
CrewAI CSVSearchTool 实战用 RAG 对 CSV 数据集做语义检索【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAICSVSearchTool是 crewai-tools 中专门用于在 CSV 文件内容中执行 RAGRetrieval-Augmented Generation语义搜索的工具。本文基于工具自带的 README 文档与仓库源码完整讲清它的两种初始化模式、入参 Schema、运行时检索参数similarity_threshold、limit、自定义模型与 Embedding 配置方式并深入到加载器CSVLoader与分块器CsvChunker层面说明一个 CSV 从文件到可检索向量的完整链路。读完后你可以直接在自己的 Agent 中挂载该工具对大型 CSV 数据集提出自然语言查询而不再依赖逐行关键词匹配。定位为什么搜索 CSV 要用 RAG工具文档给出的核心定位是在传统搜索方式效率低下的场景例如大型 CSV 数据集中CSVSearchTool允许对指定 CSV 文件的内容进行语义搜索——把自然语言查询向量化后与 CSV 内容切分出的文本块做相似度匹配返回最相关的片段而不是做字符串精确匹配。文档同时强调所有名字中带有 “Search” 的工具包括CSVSearchTool都属于 RAG 工具家族面向不同数据源PDF、JSON、XML、网站、GitHub、YouTube 等提供同一套检索能力CSVSearchTool只是其中 CSV 数据源的实现。安装方式沿用 crewai-tools 的统一入口pip install crewai[tools]从源码结构看CSVSearchTool本体非常薄位于 csv_search_tool.py核心只有三件事定义输入 Schema、在初始化时绑定固定 CSV、把检索委托给父类RagToolclass CSVSearchTool(RagTool): name: str Search a CSVs content description: str ( A tool that can be used to semantic search a query from a CSVs content. ) args_schema: type[BaseModel] CSVSearchToolSchema两种初始化模式固定 CSV 与运行时传路径README 给出的基本用法是两种等价风格的初始化from crewai_tools import CSVSearchTool # Initialize the tool with a specific CSV file. This setup allows the agent to only search the given CSV file. tool CSVSearchTool(csvpath/to/your/csvfile.csv) # OR # Initialize the tool without a specific CSV file. Agent will need to provide the CSV path at runtime. tool CSVSearchTool()这两种模式在源码中的行为差异很明确见 csv_search_tool.py指定csv初始化构造时会立即调用self.add(csv)把文件加载进知识库把工具描述改成A tool that can be used to semantic search a query the {csv} CSVs content.并把args_schema从CSVSearchToolSchema替换为FixedCSVSearchToolSchema。也就是说Agent 调用该工具时只需要提供search_query不会再要求传文件路径——工具被“锁定”在这个 CSV 上。不指定csv初始化args_schema保持为CSVSearchToolSchemaAgent 每次调用时必须额外提供csv参数文件路径或 URL运行时再动态加载。对应的两个 Pydantic Schema 定义如下csv_search_tool.pyclass FixedCSVSearchToolSchema(BaseModel): Input for CSVSearchTool. search_query: str Field( ..., descriptionMandatory search query you want to use to search the CSVs content, ) class CSVSearchToolSchema(FixedCSVSearchToolSchema): Input for CSVSearchTool. csv: str Field(..., descriptionFile path or URL of a CSV file to be searched)参数类型必填说明csv初始化参数str \| None可选初始化时固定的 CSV 路径/URL。若初始化时未指定则调用时必须提供README 原文This is a mandatory argument if the tool was initialized without a specific CSV file; otherwise, it is optional.search_querystr必填自然语言查询语句用于语义检索 CSV 内容similarity_thresholdfloat \| None可选相似度过滤阈值默认继承RagTool的0.6limitint \| None可选返回的相关文本块数量上限默认继承RagTool的5_run的实现展示了运行时传csv的处理逻辑csv_search_tool.pydef _run( self, search_query: str, csv: str | None None, similarity_threshold: float | None None, limit: int | None None, ) - str: if csv is not None: self.add(csv) return super()._run( querysearch_query, similarity_thresholdsimilarity_threshold, limitlimit )即运行时提供的 CSV 会先被add()加载进向量库随后交给父类执行查询。而add()的重写只有一行csv_search_tool.pydef add(self, csv: str) - None: super().add(csv, data_typeDataType.CSV)显式标注data_typeDataType.CSV这让底层能选择正确的加载器与分块器下一节展开。底层管线加载、分块与检索参数CSVLoader把表格转成可嵌入的文本DataType.CSV在 data_types.py 的映射表中对应CSVLoader加载与CsvChunker分块。csv_loader.py 的处理逻辑值得注意它决定了“语义检索 CSV”实际检索的是什么文本来源如果是 URL通过load_from_url下载请求头Accept: text/csv, application/csv, text/plain如果是本地路径则以 UTF-8 读取整个文件。结构化转文本用csv.DictReader解析生成如下格式的纯文本——先输出Headers: 列1 | 列2 | ...再逐行输出Row N: 列1: 值1 | 列2: 值2 | ...空值列会被跳过。这样每个 chunk 都保留了“哪一列是什么值”的语义对语义检索非常友好。元数据附带{format: csv, columns: headers, rows: N}如果解析失败如格式不合法不会抛错而是降级为原始文本并记录parse_error。一个直观例子对于name,description\ntest,This is a test CSV file加载后的文本大致是Headers: name | description -------------------------------------------------- Row 1: name: test | description: This is a test CSV fileCsvChunker按行边界切分structured_chunker.py 中CsvChunker的默认参数是chunk_size1200、chunk_overlap100分隔符优先级依次为separators [ \nRow , # Row boundaries (from CSVLoader format) \n, # Line breaks | , # Column separators , , # Comma separators , # Word breaks , # Character level ]可以看到第一个分隔符正是CSVLoader输出的Row N:行边界——加载器和分块器的输出格式是配套设计的优先保证“一个 chunk 以完整的 CSV 行为单位切分”而不是把一行数据拦腰截断。检索参数与结果格式similarity_threshold默认0.6和limit默认5定义在父类 rag_tool.pyclass RagTool(BaseTool): name: str Knowledge base description: str A knowledge base that can be used to answer questions. summarize: bool False similarity_threshold: float 0.6 limit: int 5 collection_name: str rag_tool_collectionRagTool._run的最终输出格式为Relevant Content:\n检索到的相关文本块rag_tool.py。从RagToolConfig的定义types.py可以确认向量库 provider 支持chromadb未配置时的默认值与qdrant两种。另外RagTool.add()在写入前会对文件路径和 URL 做安全校验validate_file_path/validate_url见 rag_tool.py以防止越权读取文件或 SSRF测试代码中也出现过通过环境变量CREWAI_TOOLS_ALLOW_UNSAFE_PATHS放宽限制的用法test_search_tools.py。这意味着传入的路径必须是可访问、通过校验的路径或 URL。自定义 LLM 与 EmbeddingREADME 说明工具默认使用 OpenAI 同时承担 Embedding 与摘要summarization要自定义模型可通过config字典传入llm与embedder两段配置官方示例原样保留如下tool CSVSearchTool( configdict( llmdict( providerollama, # or google, openai, anthropic, llama2, ... configdict( modelllama2, # temperature0.5, # top_p1, # streamtrue, ), ), embedderdict( providergoogle, configdict( modelmodels/embedding-001, task_typeretrieval_document, # titleEmbeddings, ), ), ) )补充源码侧的信息RagTool的config字段类型为RagToolConfigtypes.py其中还包含embedding_modelProviderSpec供 Embedding 工厂build_embedder构造 Embedding 函数与vectordbprovider: chromadb | qdrant 各自的config字典两个键。_parse_config会把embedding_model解析出的EmbeddingFunction注入到 ChromaDBConfig 或 QdrantConfig 中rag_tool.py。因此自定义 Embedding 时的两个硬约束是向量库 provider 只能是chromadb或qdrant且不同 Embedding 的维度/模型需要与向量集合保持兼容。测试用例官方如何验证这个工具仓库测试 test_search_tools.py 中的test_csv_search_tool完整覆盖了“无固定 CSV 初始化 → add → 查询”这条主链路pytest.mark.vcr() def test_csv_search_tool(): with tempfile.NamedTemporaryFile(suffix.csv, deleteFalse) as temp_file: temp_file.write(bname,description\ntest,This is a test CSV file) temp_file_path temp_file.name try: tool CSVSearchTool() tool.add(temp_file_path) result tool._run(search_querytest CSV) assert test csv in result.lower() finally: os.unlink(temp_file_path)该用例与前面讲的实现完全对应临时生成一个两列 CSVCSVSearchTool()不传csv手动add后以search_querytest CSV查询断言返回内容中能检索到该行的文本。同文件中的其他 Search 工具用例PDF、DOCX、XML、Website 等还以 mock adapter 的方式断言了统一的默认检索参数similarity_threshold0.6, limit5可以佐证 CSV 工具同样遵循这组默认值。在 Agent 中使用与适用前提把工具挂到 CrewAI 的 Agent 上即可让 Agent 在任务执行中自主发起 CSV 语义查询from crewai import Agent from crewai_tools import CSVSearchTool csv_tool CSVSearchTool(csvdata/sales.csv) # 锁定文件Agent 只需给出查询语句 researcher Agent( roleData Analyst, goalAnswer questions based on the sales CSV, tools[csv_tool], verboseTrue, )使用前的几个实际约束均来源于源码行为编码CSVLoader以 UTF-8 读取本地文件非 UTF-8 编码的 CSV 需要先转码解析降级格式异常时不会中断而是把原始文本作为内容入库并记录parse_error此时检索质量会下降建议先确认 CSV 结构规整路径/URL 安全校验运行时传入的csv会经过路径与 URL 校验路径需可读、URL 需可访问Embedding 依赖默认走 OpenAI Embedding使用离线环境时按上一节的方式通过config指定本地或自托管的 provider向量集合默认collection_name为rag_tool_collection若与其他 RAG 工具共享默认集合且想隔离数据可在构造时显式传入不同的collection_name该字段定义于 rag_tool.py。小结CSVSearchTool是 CrewAI RAG 工具家族中面向表格数据的语义检索入口初始化时可锁定单个 CSVAgent 只需提供查询语句也可留空让 Agent 运行时指定路径/URL底层由CSVLoader把“列: 值”结构化为带行号的文本、由CsvChunker按行边界1200 字符块、100 重叠切分再经 chromadb/qdrant 向量库以默认阈值 0.6、默认 Top-5 完成相似度检索。相关实现可继续追踪这几个文件工具实现、RagTool 基类、CSVLoader、CsvChunker 与 官方 README。【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考