Langchain-Chatchat 按列筛选加载 CSV:FilteredCSVLoader 设计原理与实战指南
2026/9/10 16:32:22 网站建设 项目流程

Langchain-Chatchat 按列筛选加载 CSV:FilteredCSVLoader 设计原理与实战指南

【免费下载链接】Langchain-ChatchatLangchain-Chatchat(原Langchain-ChatGLM)基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-Chatchat

本文围绕 Langchain-Chatchat 知识库文件加载体系中的自定义文档加载器FilteredCSVLoader展开:它继承 LangChain 社区的CSVLoader,允许只抽取 CSV 中指定的若干列构建文档内容、把其余列注入元数据,解决“整行塞入向量库导致噪声过大”的问题。读完本文,你将掌握该加载器全部构造参数、行级解析与异常处理细节,并能在独立脚本或 Langchain-Chatchat 知识库流水线中按列定制 CSV 加载。

一、它解决什么问题

在 RAG 知识库场景中,CSV 常被用来存放结构化语料,例如项目导出的问题单(titlefileurldetailid)、工单、日志或商品数据表。LangChain 社区默认的CSVLoader会把每一整行文本作为一条Document.page_content载入,这会导致:

  • 不需要参与语义检索的列(如idurl、行内编号)一并进入向量索引,稀释检索精度;
  • 后续文本切分与 embedding 的成本被无关内容抬高;
  • 结构化字段之间的语义关系在拼接文本中难以被模型有效利用。

FilteredCSVLoader的思路是:只把用户指定的列拼进page_content,同时把source_columnmetadata_columns指定的列放进取证更友好的metadata,让知识库的召回单元更“干净”。该类位于 FilteredCSVloader.py,是由 Langchain-Chatchat 团队针对上述场景实现的自定义加载器。

二、类定位:继承 CSVLoader 并重写加载行为

从源码头部注释“指定制定列的csv文件加载器”以及类定义可见:

from langchain.docstore.document import Document from langchain_community.document_loaders import CSVLoader from langchain_community.document_loaders.helpers import detect_file_encodings class FilteredCSVLoader(CSVLoader):

它直接继承langchain_community.document_loaders.CSVLoader,与其父类的差异体现在三处:

  1. __init__扩展了新参数columns_to_read(必需),用于声明需要读取的列名列表;
  2. load()被重写,不再使用父类默认的逐行整读逻辑,而是打开文件后转交私有方法__read_file完成列过滤式解析;
  3. 新增私有方法__read_file,内部使用csv.DictReader按列构建文档。

引入的依赖值得注意:

  • detect_file_encodings来自langchain_community.document_loaders.helpers,是编码自动检测的后端工具,与autodetect_encoding参数配套;
  • Document采用langchain.docstore.document.Document,与 Langchain-Chatchat 全库文档切分/向量化模块保持一致。

三、构造参数详解(与源码逐一对应)

FilteredCSVLoader.__init__的完整签名如下(源码 FilteredCSVloader.py#L13-L31):

def __init__( self, file_path: str, columns_to_read: List[str], source_column: Optional[str] = None, metadata_columns: List[str] = [], csv_args: Optional[Dict] = None, encoding: Optional[str] = None, autodetect_encoding: bool = False, ): super().__init__( file_path=file_path, source_column=source_column, metadata_columns=metadata_columns, csv_args=csv_args, encoding=encoding, autodetect_encoding=autodetect_encoding, ) self.columns_to_read = columns_to_read

各参数含义与默认值归纳如下:

参数类型默认值作用
file_pathstr必传CSV 文件路径,须指向磁盘上存在且可读的文件
columns_to_readList[str]必传需要读取并写入page_content的列名列表;列名必须在 CSV 中存在,否则抛ValueError
source_columnOptional[str]None指定作为数据源信息的列名;为None或列不存在时回退为文件路径作为 source
metadata_columnsList[str][]需要原样搬进metadata的列名列表,这些列不参与正文内容拼接
csv_argsOptional[Dict]None透传给csv.DictReader的参数字典,例如自定义delimiterquotechar
encodingOptional[str]None打开文件时使用的编码(会传入内置open),处理非 UTF-8 文件时使用
autodetect_encodingboolFalse当解码失败时是否调用detect_file_encodings自动探测编码并重试

__init__先调用父类构造器完成除columns_to_read之外全部参数的传递,再把columns_to_read保存为实例属性,供后续__read_file使用。这样既复用了父类对source_columnmetadata_columnscsv_args的既有处理契约,又保证了本类参数体系与父类兼容。

四、load():打开文件、编码兜底与统一错误出口

load()方法(FilteredCSVloader.py#L33-L57)是加载流程的入口,其行为可拆解为三层:

第一层:正常路径。使用open(self.file_path, newline="", encoding=self.encoding)打开文件并调用self.__read_file(csvfile)解析。注意newline=""csv模块推荐的写法,避免跨平台行结束符差异导致解析错乱。

第二层:UnicodeDecodeError 分支。若首次按指定编码打开失败,且autodetect_encoding=True,则调用detect_file_encodings(self.file_path)获得候选编码序列,逐个尝试打开并解析,直到某一次成功即break;若未开启自动检测,则直接抛出RuntimeError(f"Error loading {self.file_path}")(以原始异常为 cause)。

第三层:统一错误出口。解析过程中的任何其他异常都被包装为RuntimeError上抛,并携带文件路径便于排障。

需要留意的一个实现细节:编码探测分支内部仅except UnicodeDecodeErrorcontinue,若全部候选编码都失败,load()将返回空列表而非报错。因此生产环境下对未知来源的 CSV,建议在调用侧额外判断返回的docs是否为空。

五、__read_file:行级列筛选的底层原理

私有方法__read_file(FilteredCSVloader.py#L59-L87)负责真正的解析,整个算法按行推进:

def __read_file(self, csvfile: TextIOWrapper) -> List[Document]: docs = [] csv_reader = csv.DictReader(csvfile, **self.csv_args) for i, row in enumerate(csv_reader): content = [] for col in self.columns_to_read: if col in row: content.append(f"{col}:{str(row[col])}") else: raise ValueError( f"Column '{self.columns_to_read[0]}' not found in CSV file." ) content = "\n".join(content) source = ( row.get(self.source_column, None) if self.source_column is not None else self.file_path ) metadata = {"source": source, "row": i} for col in self.metadata_columns: if col in row: metadata[col] = row[col] doc = Document(page_content=content, metadata=metadata) docs.append(doc) return docs

其关键行为可以提炼为 5 条规则,均为源码可直接验证的实现事实:

  1. csv.DictReader把每一行读成“列名 → 值”的字典,行号i从 0 开始计数(注意第一行表头不计入行号);
  2. 正文按“列名:值”格式逐列拼接,多列之间用\n连接,最终整体作为page_content。因此检索单元会携带列名语义,比裸文本更利于模型理解字段含义;
  3. 缺失列即抛ValueError。注意实现细节:columns_to_read中任一列缺失都会报错,但错误信息引用的是self.columns_to_read[0](第一个列名),排障时需要自行核对是哪一个列名拼写不符;
  4. source 解析三级回退source_columnNone→ 使用file_path;指定了source_column但该列在当前行不存在 →row.get(..., None)None;指定且存在 → 取该列值。metadata始终携带{"source": ..., "row": i}两个字段,metadata_columns中存在的列会追加进同一字典;
  5. 每个数据行独立封装为一个Document,追加进docs返回,天然适合后续按行切分与入库。

正因为__read_file是私有方法(名称以下划线开头),它只允许被类内部调用,官方文档也明确提示不应从类外部直接调用。

六、输出形态:代码真实产物与文档示例的差异

原类文档给出的输出示例是“干净文本”风格的示意:

[ Document(page_content="这是第一行的内容", metadata={"source": "example.csv", "row": 0}), Document(page_content="这是第二行的内容", metadata={"source": "example.csv", "row": 1}), ]

而对照源码,实际产物中的page_content是**“列名:值”并以换行连接**的文本。以仓库自带示例数据 langchain-ChatGLM_closed.csv 为例,其表头为:

,title,file,url,detail,id 0,加油~以及一些建议,2023-03-31.0002,https://github.com/.../issues/2,加油,我认为你的方向是对的。,0

若设置columns_to_read=["title", "detail"],第一行数据的真实产物应为:

Document( page_content="title:加油~以及一些建议\ndetail:加油,我认为你的方向是对的。", metadata={"source": "<csv 文件路径>", "row": 0}, )

理解这一点对后续知识库效果调优很重要:page_content的字段前缀会一并进入文本切分与 embedding 计算,若希望检索内容更贴近问答语料,可选择只读真正的“正文”列(如detail),把titleurl等放入metadata_columns用于溯源与引用。

七、在 Langchain-Chatchat 知识库流水线中的定位

把视角从“单类加载器”拉高到 Langchain-Chatchat 的知识库文件处理流水线,可以看清它的边界。

知识库的工具函数定义在 knowledge_base/utils.py:

  • LOADER_DICT声明“文件扩展名 → 加载器名”的映射,.csv默认映射到CSVLoader,而FilteredCSVLoader那一行处于被注释状态,注释即说明了设计意图:# "FilteredCSVLoader": [".csv"], 如果使用自定义分割csv。也就是说:默认流程下 CSV 交由 LangChain 社区原版CSVLoader整行加载,只有需要“自定义按列拆分”时才启用FilteredCSVLoader
  • SUPPORTED_EXTSLOADER_DICT拍平生成,.csv是受支持的文件类型之一;
  • KnowledgeFile(同文件 utils.py#L312)在初始化时通过get_LoaderClass(self.ext)决定加载器名称,其file2docs()再调用get_loader(...)拿到加载器实例并执行.load(),得到原始Document列表;
  • get_loader()(utils.py#L169)内置了自定义加载器名单(RapidOCRPDFLoaderRapidOCRLoaderFilteredCSVLoaderRapidOCRDocLoaderRapidOCRPPTLoader),对这些名称会优先从chatchat.server.file_rag.document_loaders包内解析,其他名称回退到langchain_community.document_loaders
  • 同一函数对CSVLoader有专门的编码处理:若用户未显式传入encoding,会先用chardet.detect探测二进制内容并写入loader_kwargs["encoding"],避免加载中文 CSV 时出现编码错误。

因此在实际的 Langchain-Chatchat 部署副本中若要用FilteredCSVLoader替代默认 CSV 加载,需要在知识库文件加载相关配置(即LOADER_DICT所在配置处)把FilteredCSVLoader.csv关联、并传入columns_to_read等 loader 参数,使其经由get_loader的名单分支被实例化;代码注释以“如果使用自定义分割 csv”一句话点明这正是它存在的原因。

八、可直接运行的完整示例

场景 A:独立脚本中使用

下面的代码把示例 CSV 中titledetail两列作为正文,urlid作为元数据,并让url同时承担 source 溯源:

from chatchat.server.file_rag.document_loaders.FilteredCSVloader import FilteredCSVLoader loader = FilteredCSVLoader( file_path="langchain-ChatGLM_closed.csv", columns_to_read=["title", "detail"], source_column="url", metadata_columns=["id"], csv_args={"delimiter": ",", "quotechar": '"'}, encoding="utf-8", autodetect_encoding=True, ) docs = loader.load() for doc in docs[:2]: print(doc.page_content) print(doc.metadata)

输出示意:

title:加油~以及一些建议 detail:加油,我认为你的方向是对的。 {'source': 'https://github.com/.../issues/2', 'row': 0, 'id': '0'}

场景 B:非 UTF-8 文件的编码兜底

若 CSV 实际是 GBK 编码而encoding未指定,可开启自动探测:

loader = FilteredCSVLoader( file_path="gbk_file.csv", columns_to_read=["question", "answer"], encoding="utf-8", # 首次尝试 autodetect_encoding=True, # 失败后探测候选编码逐个重试 )

需要注意前置条件:autodetect_encoding依赖 LangChain 侧的detect_file_encodings辅助函数,其探测成本与文件大小相关,仅对“编码不确定”的文件开启即可;自动检测全部失败时方法不会抛错而是返回空列表,调用方应自行判空。

场景 C:中文 RAG 中的列取舍

结构化 CSV 进入知识库前,建议把真正承载语义的列交给columns_to_read,把用于展示/溯源的列交给metadata_columns,例如对表格型语料只读正文列,避免id、序号等无意义内容参与向量化——这正是FilteredCSVLoader相对默认整行加载的核心价值。

九、注意事项与最佳实践小结

综合类文档(FilteredCSVloader.md的“注意”段落)与源码实现,使用时有几条必须记住的约束:

  • 文件前提file_path必须真实存在且可读,否则会以RuntimeError包装上抛;
  • 列前提columns_to_read中的每个列名都必须存在于表头中,缺失时抛出ValueError;错误信息只引用第一个列名,排障时按完整列表逐项核对;
  • 编码前提:未指定encoding且未开启autodetect_encoding时,非目标编码文件会直接抛RuntimeError;仅autodetect_encoding=True才触发逐候选编码重试;
  • 私有方法边界__read_file是私有实现,请通过load()间接调用;
  • 大文件内存load()一次性返回全部Document,超大 CSV 建议自行做流式分块或切分后再入库;
  • 行号语义metadata["row"]从 0 开始、且不计表头,与 Excel 中看到的行号差 2,做溯源展示时需要换算。

十、结语

FilteredCSVLoader是 Langchain-Chatchat 在 LangChain 加载器之上做“按列抽取”定制的代表性组件:它以最小改动继承CSVLoader,通过columns_to_read控制正文语义、metadata_columns控制溯源信息、source_column控制来源列,配合encoding/autodetect_encoding双保险解决中文 CSV 编码这一高频痛点。若你正在为知识库投喂带噪点的结构化表格数据,把这篇文档与 源码实现 及 加载器注册逻辑 对照阅读,即可快速落地一套按列筛选、字段化、可溯源的 CSV 知识库加载方案。

【免费下载链接】Langchain-ChatchatLangchain-Chatchat(原Langchain-ChatGLM)基于 Langchain 与 ChatGLM, Qwen 与 Llama 等语言模型的 RAG 与 Agent 应用 | Langchain-Chatchat (formerly langchain-ChatGLM), local knowledge based LLM (like ChatGLM, Qwen and Llama) RAG and Agent app with langchain项目地址: https://gitcode.com/GitHub_Trending/la/Langchain-Chatchat

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询