尧图网页设计 
1. 为什么数据导入是 RAG 系统的第一道生死关做 RAG 的人都有一个共识检索效果差八成不是模型不行而是数据没处理好。我见过太多团队花大价钱调 embedding 模型、换向量数据库、折腾重排序结果回头一看原始文档里全是乱码、断行、页眉页脚混在正文里检索出来的 chunk 驴唇不对马嘴。这不是模型的问题是数据导入和解析这一层就没打好地基。RAG 的全称是检索增强生成它的核心逻辑是“先找到相关资料再让模型基于资料回答”。那么问题来了资料从哪来怎么变成模型能用的格式这就是 Document Loader 和文本解析要解决的事。LangChain 里把这一步抽象成了 Document 对象和 Loader 组件但很多人只是照着文档跑了个TextLoader就以为完事了实际上不同格式的文件需要完全不同的处理策略。这篇内容适合谁看如果你正在搭建 RAG 知识库手头有一堆 txt、Markdown、PDF、Word 文档要往里灌但发现检索效果不理想或者你刚开始接触 LangChain想搞清楚 Document Loader 到底该怎么选、怎么用那这篇就是写给你的。我会从最基础的 txt 和 Markdown 入手把通用文本和结构化文档的解析逻辑拆开讲透后面再延伸到 PDF、Word 等复杂格式。这一篇先把地基打牢。2. Document 对象与 Loader 体系的核心设计逻辑2.1 Document 到底装了什么LangChain 里所有加载器最终都输出Document对象它只有两个核心字段page_content和metadata。看起来简单但这两个字段的设计直接决定了后续检索的质量。page_content是字符串就是文本内容本身。metadata是字典存的是这份文本的来源信息——文件路径、页码、标题层级、创建时间等等。很多人忽略 metadata 的重要性觉得只要文本内容对就行。但实际检索时metadata 是过滤和排序的关键依据。比如你检索到一段内容想知道它来自哪个文件的哪一页没有 metadata 就抓瞎了。from langchain_core.documents import Document doc Document( page_content这是正文内容, metadata{source: docs/intro.md, page: 1, section: 概述} )注意metadata 里的值尽量用字符串、数字、布尔值这类基础类型不要塞复杂对象。因为后续做向量化存储时metadata 往往要序列化复杂对象会直接报错。2.2 Loader 的选型逻辑LangChain 提供了几十种 Loader从TextLoader到UnstructuredMarkdownLoader再到PyPDFLoader名字五花八门。选型的核心原则只有一条根据文件格式选择能保留最多结构信息的 Loader。为什么强调“保留结构信息”因为 RAG 检索的粒度是 chunk而 chunk 的切分质量取决于原始文本的结构是否清晰。一个 Markdown 文件如果被当成纯文本加载标题、列表、代码块的层级关系全丢了切出来的 chunk 就是一堆没有上下文关联的碎片。反过来如果用专门的 Markdown Loader它能识别标题层级切分时就能按章节来切每个 chunk 都带着所属章节的上下文。我个人的选型优先级是这样的能用结构化 Loader 就不用通用 Loader能用专门格式 Loader 就不用 Unstructured 兜底。具体来说txt 用TextLoaderMarkdown 用UnstructuredMarkdownLoader或MarkdownHeaderTextSplitterPDF 用PyPDFLoader或PDFPlumberLoaderWord 用Docx2txtLoader。Unstructured 系列是万能兜底方案但它的解析质量取决于底层依赖库有时候反而不如专用 Loader 稳定。2.3 编码问题最容易被忽视的坑txt 文件加载最常见的翻车场景就是编码。Windows 上创建的 txt 默认可能是 GBK 或 GB2312Linux 和 macOS 默认 UTF-8。如果你在 Linux 服务器上跑TextLoader加载一个 GBK 编码的文件直接抛UnicodeDecodeError。from langchain_community.document_loaders import TextLoader # 指定编码避免乱码 loader TextLoader(data/example.txt, encodingutf-8) docs loader.load()如果不知道文件编码怎么办可以用chardet库检测import chardet with open(data/example.txt, rb) as f: raw f.read() result chardet.detect(raw) print(result) # {encoding: GB2312, confidence: 0.99}实测下来chardet对中文编码的检测准确率在 95% 以上但偶尔会把 GBK 误判成 GB2312。这两个编码大部分字符兼容但遇到生僻字可能出问题。稳妥的做法是检测到 GB 系列编码后统一用gb18030来解码它是 GBK 的超集兼容性最好。3. txt 文件加载看似简单细节不少3.1 TextLoader 的基本用法与参数TextLoader是 LangChain 里最简单的 Loader没有之一。它的核心参数就三个file_path、encoding、autodetect_encoding。from langchain_community.document_loaders import TextLoader loader TextLoader( file_pathdata/knowledge.txt, encodingutf-8, autodetect_encodingTrue # 自动检测编码 ) documents loader.load() print(f加载了 {len(documents)} 个文档) print(f第一个文档内容长度: {len(documents[0].page_content)}) print(fmetadata: {documents[0].metadata})autodetect_encodingTrue会让 Loader 内部用chardet检测编码省去手动指定的麻烦。但这个参数有个小坑检测本身有开销如果文件很大或者要批量加载几千个文件检测时间会累积。我的做法是先用chardet批量检测一遍把编码信息记下来后续加载直接指定不做重复检测。3.2 大文件的分块策略TextLoader默认把整个文件读成一个 Document。如果文件只有几 KB没问题。但如果是一个几十 MB 的日志文件或者长篇小说直接塞进去会导致后续 embedding 超长、检索粒度太粗。这时候需要在加载后做切分。LangChain 提供了多种 TextSplitter最常用的是RecursiveCharacterTextSplitterfrom langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) docs loader.load() split_docs text_splitter.split_documents(docs) print(f切分后得到 {len(split_docs)} 个 chunk)chunk_size500是什么概念大约 300-400 个中文字符。这个粒度适合大多数问答场景既不会太碎导致语义不完整也不会太长导致检索精度下降。chunk_overlap50是让相邻 chunk 之间有 50 个字符的重叠防止关键信息刚好被切断。separators的顺序很重要。它按优先级依次尝试用分隔符切分先尝试\n\n段落如果切出来的块还是太大再用\n换行然后是中文标点。这个顺序保证了切分尽量在语义边界发生而不是在句子中间硬切。实操心得中文文本的 separators 一定要把中文标点加进去。默认的 separators 只有英文标点和空格对中文文本来说切分点会落在很奇怪的位置。我一般会加上。、、、这几个效果立竿见影。3.3 批量加载多个 txt 文件实际项目里很少只有一个 txt 文件通常是一个目录下几十上百个文件。LangChain 提供了DirectoryLoader来批量处理from langchain_community.document_loaders import DirectoryLoader, TextLoader loader DirectoryLoader( pathdata/txt_files/, glob**/*.txt, # 递归匹配所有 txt loader_clsTextLoader, loader_kwargs{encoding: utf-8, autodetect_encoding: True}, show_progressTrue, use_multithreadingTrue, max_concurrency4 ) docs loader.load() print(f共加载 {len(docs)} 个文档)use_multithreadingTrue和max_concurrency4可以并行加载对大量小文件效果明显。但要注意如果文件在机械硬盘上并发读取反而可能因为磁头频繁寻道而变慢。SSD 上并发效果最好机械硬盘建议max_concurrency2就够了。show_progressTrue会显示进度条批量加载时很有用能直观看到处理进度。但如果在日志系统里跑进度条会刷屏这时候关掉它。4. Markdown 解析结构化文本的正确打开方式4.1 为什么 Markdown 需要专门处理Markdown 是 RAG 知识库最理想的格式之一因为它自带结构标记——标题、列表、代码块、表格都有明确的语法。如果把它当纯文本加载这些结构信息就全丢了。举个例子一个 Markdown 文件里有这样的内容## 安装步骤 ### 环境要求 - Python 3.8 - 内存 8GB 以上 ### 安装命令 bash pip install langchain如果当纯文本加载切分后可能变成“Python 3.8 内存 8GB 以上 pip install langchain”标题层级完全丢失。检索时用户问“安装环境要求是什么”可能匹配不到正确的 chunk因为 chunk 里没有“环境要求”这个标题信息。 用 MarkdownHeaderTextSplitter 就不一样了它会识别 ## 和 ###把标题作为 metadata 附加到每个 chunk 上。检索时不仅能匹配正文还能匹配标题精度提升非常明显。 ### 4.2 MarkdownHeaderTextSplitter 实战 python from langchain.text_splitter import MarkdownHeaderTextSplitter markdown_text # RAG 系统指南 ## 第一章 数据导入 ### 1.1 文本加载 使用 TextLoader 加载纯文本文件。 ### 1.2 Markdown 加载 使用 MarkdownHeaderTextSplitter 处理结构化文档。 ## 第二章 向量存储 ### 2.1 向量数据库选型 常见选择包括 Chroma、FAISS、Milvus。 headers_to_split_on [ (#, h1), (##, h2), (###, h3), ] splitter MarkdownHeaderTextSplitter( headers_to_split_onheaders_to_split_on, strip_headersFalse # 保留标题在正文中 ) chunks splitter.split_text(markdown_text) for chunk in chunks: print(内容:, chunk.page_content[:50]) print(metadata:, chunk.metadata) print(---)输出结果里每个 chunk 的 metadata 都会带上它所属的标题层级。比如“使用 TextLoader 加载纯文本文件”这个 chunk 的 metadata 是{h1: RAG 系统指南, h2: 第一章 数据导入, h3: 1.1 文本加载}。这样检索时即使用户问的是“第一章讲什么”也能通过 metadata 匹配到相关 chunk。strip_headersFalse这个参数值得说一下。默认情况下切分后标题会从正文中移除只保留在 metadata 里。但实测下来保留标题在正文中对检索效果更好因为 embedding 模型能同时看到标题和内容语义更完整。所以我一般设成False。4.3 结合 RecursiveCharacterTextSplitter 做二级切分MarkdownHeaderTextSplitter有个限制它只按标题切分不控制 chunk 大小。如果某个章节内容特别长切出来的 chunk 可能超过 embedding 模型的 token 限制。这时候需要做二级切分from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter # 第一级按标题切分 md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) md_chunks md_splitter.split_text(markdown_text) # 第二级按大小切分 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) final_chunks text_splitter.split_documents(md_chunks) print(f最终得到 {len(final_chunks)} 个 chunk) for chunk in final_chunks[:3]: print(metadata:, chunk.metadata) print(内容长度:, len(chunk.page_content)) print(---)这个两级切分策略是我目前用得最顺手的方案。第一级保证结构信息不丢失第二级保证 chunk 大小可控。split_documents会自动继承第一级产生的 metadata所以最终每个 chunk 既有标题信息又不会超长。4.4 用 UnstructuredMarkdownLoader 加载文件如果 Markdown 内容在文件里而不是字符串可以用UnstructuredMarkdownLoaderfrom langchain_community.document_loaders import UnstructuredMarkdownLoader loader UnstructuredMarkdownLoader( docs/guide.md, modesingle, # single 返回一个 Documentelements 返回多个 strategyfast # fast 速度快hi_res 精度高但慢 ) docs loader.load()modesingle把整个文件作为一个 Document 返回modeelements则按元素标题、段落、列表拆成多个 Document。做 RAG 一般用single然后自己控制切分逻辑灵活性更高。strategy参数控制解析策略。fast用简单的正则匹配速度快但可能漏掉复杂结构hi_res用更复杂的模型解析精度高但慢很多。Markdown 本身结构清晰fast就够了没必要上hi_res。5. 通用文本与结构化文档的融合处理5.1 混合格式目录的统一加载实际项目里一个知识库目录下往往 txt、Markdown、PDF、Word 混在一起。这时候需要根据文件扩展名动态选择 Loaderimport os from langchain_community.document_loaders import ( TextLoader, UnstructuredMarkdownLoader, PyPDFLoader, Docx2txtLoader ) def load_document(file_path): ext os.path.splitext(file_path)[1].lower() loader_map { .txt: lambda p: TextLoader(p, encodingutf-8, autodetect_encodingTrue), .md: lambda p: UnstructuredMarkdownLoader(p, modesingle), .pdf: lambda p: PyPDFLoader(p), .docx: lambda p: Docx2txtLoader(p), } if ext not in loader_map: raise ValueError(f不支持的文件格式: {ext}) loader loader_map[ext](file_path) return loader.load() # 批量加载目录 all_docs [] for root, dirs, files in os.walk(knowledge_base/): for file in files: file_path os.path.join(root, file) try: docs load_document(file_path) all_docs.extend(docs) print(f成功加载: {file_path} ({len(docs)} 个文档)) except Exception as e: print(f加载失败: {file_path}, 错误: {e}) print(f总计加载 {len(all_docs)} 个文档)这个模式的好处是扩展性强新增格式只需要在loader_map里加一行。异常处理也很重要批量加载时某个文件出错不应该中断整个流程而是记录错误继续处理。5.2 metadata 的统一规范不同 Loader 产生的 metadata 字段不一样。TextLoader只有sourcePyPDFLoader有source和pageUnstructuredMarkdownLoader可能有一堆额外字段。为了后续检索时能统一过滤建议在加载后做一次 metadata 规范化def normalize_metadata(doc, file_path): 统一 metadata 字段 doc.metadata[source] file_path doc.metadata[file_type] os.path.splitext(file_path)[1].lower() doc.metadata[file_name] os.path.basename(file_path) # 确保 page 字段存在 if page not in doc.metadata: doc.metadata[page] 0 return doc all_docs [normalize_metadata(doc, doc.metadata.get(source, )) for doc in all_docs]统一 metadata 之后检索时就可以按file_type过滤比如只搜 Markdown 文档或者按file_name聚合结果。这些在构建复杂 RAG 应用时非常有用。5.3 去重与质量过滤批量加载后经常遇到重复内容。比如同一个文件被复制了两份或者 PDF 和 Markdown 版本内容重叠。重复内容会导致检索结果冗余浪费 token。def deduplicate_documents(docs, similarity_threshold0.95): 基于内容哈希的简单去重 seen_hashes set() unique_docs [] for doc in docs: # 用内容的前 200 个字符做哈希 content_hash hash(doc.page_content[:200].strip()) if content_hash not in seen_hashes: seen_hashes.add(content_hash) unique_docs.append(doc) return unique_docs # 更严格的去重可以用 MinHash 或 SimHash这个简单哈希去重只能处理完全重复或开头高度相似的情况。如果要处理语义重复比如两段话意思一样但措辞不同需要上 MinHash 或 SimHash 这类局部敏感哈希算法。不过对于大多数知识库场景简单哈希去重已经能解决 80% 的问题。质量过滤也很关键。加载后要检查每个 Document 的page_content长度太短的比如少于 10 个字符直接丢弃这些通常是页眉页脚或者解析残留def filter_quality(docs, min_length10): 过滤掉内容过短的文档 return [doc for doc in docs if len(doc.page_content.strip()) min_length] all_docs filter_quality(all_docs) print(f质量过滤后剩余 {len(all_docs)} 个文档)6. 常见问题与排查技巧实录6.1 编码乱码问题速查现象可能原因解决方法中文显示为\xef\xbf\xbdUTF-8 解码失败用chardet检测真实编码中文显示为乱码字符GBK 文件用 UTF-8 读指定encodinggb18030部分字符正常部分乱码混合编码逐行检测编码分段解码读取时报UnicodeDecodeError编码不匹配加errorsignore跳过错误字节errorsignore是最后的兜底方案它会跳过无法解码的字节。但这会导致内容丢失只适合对完整性要求不高的场景。更好的做法还是正确检测编码。6.2 Markdown 切分后 metadata 丢失用MarkdownHeaderTextSplitter切分后如果再做RecursiveCharacterTextSplitter有时候 metadata 会丢。原因是split_documents方法会继承 metadata但如果你用的是split_text然后手动构造 Documentmetadata 就没了。# 错误做法metadata 丢失 texts text_splitter.split_text(md_chunk.page_content) new_docs [Document(page_contentt) for t in texts] # metadata 没了 # 正确做法用 split_documents 自动继承 new_docs text_splitter.split_documents([md_chunk]) # metadata 保留踩坑记录我一开始用split_text然后手动拼 Document结果检索时发现所有 chunk 的 metadata 都是空的排查了半天才发现是这里的问题。后来统一用split_documents再也没丢过 metadata。6.3 大文件加载内存溢出加载几百 MB 的 txt 文件时TextLoader会一次性读入内存可能导致 OOM。解决方案是流式加载from langchain_community.document_loaders import TextLoader # 流式加载逐行读取 loader TextLoader(huge_file.txt, encodingutf-8) for doc in loader.lazy_load(): # 逐块处理不一次性加载全部 process_document(doc)lazy_load()返回一个生成器每次只加载一个 Document。配合TextSplitter的split_documents也可以流式处理内存占用大幅降低。6.4 批量加载速度慢批量加载几千个文件时速度可能很慢。优化手段有几个开启多线程use_multithreadingTruemax_concurrency根据 CPU 核心数设置跳过已处理的文件记录已加载文件的哈希下次跳过用更快的 LoaderTextLoader比UnstructuredMarkdownLoader快很多如果 Markdown 结构不复杂可以直接用TextLoader加载再手动处理import hashlib import json import os def get_file_hash(file_path): 计算文件哈希 hasher hashlib.md5() with open(file_path, rb) as f: hasher.update(f.read()) return hasher.hexdigest() # 加载已处理记录 processed {} if os.path.exists(processed.json): with open(processed.json, r) as f: processed json.load(f) # 跳过已处理文件 for file_path in file_list: file_hash get_file_hash(file_path) if processed.get(file_path) file_hash: continue # 跳过 # 处理文件... processed[file_path] file_hash # 保存记录 with open(processed.json, w) as f: json.dump(processed, f)这个增量加载策略在知识库持续更新的场景下非常实用每次只需要处理新增或修改的文件不用全量重跑。6.5 Markdown 表格和代码块的特殊处理Markdown 里的表格和代码块在切分时容易被破坏。表格被从中间切断代码块被拆成两半都会影响检索质量。from langchain.text_splitter import RecursiveCharacterTextSplitter # 针对含代码块的 Markdown调整 separators splitter RecursiveCharacterTextSplitter( chunk_size800, # 调大一点容纳完整代码块 chunk_overlap100, separators[ \n\n, # 代码块边界优先 \n\n, # 段落 \n|, # 表格行 \n, # 换行 。, , , , , , ] )把代码块边界\n\n放在 separators 最前面切分时会优先在代码块之间切而不是在代码块内部切。表格行\n|也加进去尽量保持表格完整。当然如果表格特别大还是可能被切开这时候可以考虑把表格单独提取出来做特殊处理。7. 从加载到入库的完整链路演示7.1 完整代码示例把前面讲的串起来一个完整的从文件加载到切分入库的流程import os from langchain_community.document_loaders import ( TextLoader, UnstructuredMarkdownLoader, DirectoryLoader ) from langchain.text_splitter import ( MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter ) from langchain_core.documents import Document def load_and_split(knowledge_dir): 加载目录下所有文档并切分 all_docs [] # 1. 加载 txt 文件 txt_loader DirectoryLoader( knowledge_dir, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8, autodetect_encoding: True}, show_progressTrue ) all_docs.extend(txt_loader.load()) # 2. 加载 Markdown 文件 md_loader DirectoryLoader( knowledge_dir, glob**/*.md, loader_clsUnstructuredMarkdownLoader, loader_kwargs{mode: single}, show_progressTrue ) md_docs md_loader.load() # 3. Markdown 按标题切分 md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) md_chunks [] for doc in md_docs: chunks md_splitter.split_text(doc.page_content) # 继承原始 metadata for chunk in chunks: chunk.metadata.update(doc.metadata) md_chunks.extend(chunks) all_docs.extend(md_chunks) # 4. 统一二级切分 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) final_docs text_splitter.split_documents(all_docs) # 5. 质量过滤 final_docs [d for d in final_docs if len(d.page_content.strip()) 10] print(f最终得到 {len(final_docs)} 个 chunk) return final_docs # 执行 docs load_and_split(knowledge_base/)7.2 入库前的最后检查切分完成后入库前建议做一次快速检查确认 chunk 质量def inspect_chunks(docs, sample_size5): 抽样检查 chunk 质量 import random samples random.sample(docs, min(sample_size, len(docs))) for i, doc in enumerate(samples): print(f 样本 {i1} ) print(f来源: {doc.metadata.get(source, unknown)}) print(f长度: {len(doc.page_content)}) print(f内容预览: {doc.page_content[:100]}...) print() inspect_chunks(docs)重点看几个指标chunk 长度是否在预期范围内400-600 字符metadata 是否完整内容开头结尾是否在语义边界。如果发现大量 chunk 长度只有几十个字符说明切分参数需要调整如果 metadata 大量缺失说明加载或切分环节有问题。7.3 接入向量数据库切分好的 Document 列表可以直接喂给向量数据库。以 Chroma 为例from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings # 初始化 embedding 模型 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 存入 Chroma vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./chroma_db, collection_nameknowledge_base ) print(f已存入 {vectorstore._collection.count()} 个向量)normalize_embeddingsTrue对 BGE 系列模型很重要它会把向量归一化到单位长度这样余弦相似度计算更准确。不同 embedding 模型对归一化的要求不一样用之前查一下模型文档。8. 一些实战中攒下来的经验数据导入和解析这件事看起来是 RAG 系统里最没技术含量的环节但实际上它决定了整个系统的上限。我踩过的坑里至少有一半跟数据质量有关。有几个经验值得单独拎出来说。第一不要迷信自动检测。autodetect_encoding和strategyhi_res这些自动化的东西在 demo 阶段很好用但生产环境一定要做人工抽检。我遇到过chardet把 UTF-8 文件误判成 GBK 的情况导致整个文件乱码而且不报错静默地产生了一堆垃圾 chunk。第二metadata 比正文更重要。正文内容 embedding 模型能理解但 metadata 是过滤和排序的唯一依据。加载阶段就要把能收集的 metadata 都收集上——文件路径、修改时间、文件类型、标题层级、页码。后面做检索优化时这些字段就是你的武器库。第三切分参数没有万能值。chunk_size500只是一个起点。技术文档可以小一点300-400叙事性内容可以大一点800-1000。最好的办法是拿一批典型 query 做检索测试看不同参数下的召回率和准确率用数据说话。第四增量更新要提前设计。知识库不是一次性的后续会不断有新文档加入、旧文档修改。如果每次更新都全量重跑成本会越来越高。用文件哈希做增量检测只处理变化的文件这个机制越早做越好。第五保留原始文件路径。metadata 里的source字段一定要存原始文件的绝对路径或可访问的 URL。检索到结果后用户往往想点进去看原文没有路径就只能干瞪眼。如果文件在对象存储里存 URL如果在本机存绝对路径。这套流程我在几个项目里反复迭代过从最开始用TextLoader一把梭到后面针对不同格式做精细化处理检索效果提升非常明显。同样的 embedding 模型和向量数据库只是把数据导入这一层做扎实了问答准确率就能从 60% 出头拉到 85% 以上。数据质量这件事投入产出比远比调模型参数高。 LangChain Document Loader 实战：txt 与 Markdown 数据导入解析指南
1. 为什么数据导入是 RAG 系统的第一道生死关做 RAG 的人都有一个共识检索效果差八成不是模型不行而是数据没处理好。我见过太多团队花大价钱调 embedding 模型、换向量数据库、折腾重排序结果回头一看原始文档里全是乱码、断行、页眉页脚混在正文里检索出来的 chunk 驴唇不对马嘴。这不是模型的问题是数据导入和解析这一层就没打好地基。RAG 的全称是检索增强生成它的核心逻辑是“先找到相关资料再让模型基于资料回答”。那么问题来了资料从哪来怎么变成模型能用的格式这就是 Document Loader 和文本解析要解决的事。LangChain 里把这一步抽象成了 Document 对象和 Loader 组件但很多人只是照着文档跑了个TextLoader就以为完事了实际上不同格式的文件需要完全不同的处理策略。这篇内容适合谁看如果你正在搭建 RAG 知识库手头有一堆 txt、Markdown、PDF、Word 文档要往里灌但发现检索效果不理想或者你刚开始接触 LangChain想搞清楚 Document Loader 到底该怎么选、怎么用那这篇就是写给你的。我会从最基础的 txt 和 Markdown 入手把通用文本和结构化文档的解析逻辑拆开讲透后面再延伸到 PDF、Word 等复杂格式。这一篇先把地基打牢。2. Document 对象与 Loader 体系的核心设计逻辑2.1 Document 到底装了什么LangChain 里所有加载器最终都输出Document对象它只有两个核心字段page_content和metadata。看起来简单但这两个字段的设计直接决定了后续检索的质量。page_content是字符串就是文本内容本身。metadata是字典存的是这份文本的来源信息——文件路径、页码、标题层级、创建时间等等。很多人忽略 metadata 的重要性觉得只要文本内容对就行。但实际检索时metadata 是过滤和排序的关键依据。比如你检索到一段内容想知道它来自哪个文件的哪一页没有 metadata 就抓瞎了。from langchain_core.documents import Document doc Document( page_content这是正文内容, metadata{source: docs/intro.md, page: 1, section: 概述} )注意metadata 里的值尽量用字符串、数字、布尔值这类基础类型不要塞复杂对象。因为后续做向量化存储时metadata 往往要序列化复杂对象会直接报错。2.2 Loader 的选型逻辑LangChain 提供了几十种 Loader从TextLoader到UnstructuredMarkdownLoader再到PyPDFLoader名字五花八门。选型的核心原则只有一条根据文件格式选择能保留最多结构信息的 Loader。为什么强调“保留结构信息”因为 RAG 检索的粒度是 chunk而 chunk 的切分质量取决于原始文本的结构是否清晰。一个 Markdown 文件如果被当成纯文本加载标题、列表、代码块的层级关系全丢了切出来的 chunk 就是一堆没有上下文关联的碎片。反过来如果用专门的 Markdown Loader它能识别标题层级切分时就能按章节来切每个 chunk 都带着所属章节的上下文。我个人的选型优先级是这样的能用结构化 Loader 就不用通用 Loader能用专门格式 Loader 就不用 Unstructured 兜底。具体来说txt 用TextLoaderMarkdown 用UnstructuredMarkdownLoader或MarkdownHeaderTextSplitterPDF 用PyPDFLoader或PDFPlumberLoaderWord 用Docx2txtLoader。Unstructured 系列是万能兜底方案但它的解析质量取决于底层依赖库有时候反而不如专用 Loader 稳定。2.3 编码问题最容易被忽视的坑txt 文件加载最常见的翻车场景就是编码。Windows 上创建的 txt 默认可能是 GBK 或 GB2312Linux 和 macOS 默认 UTF-8。如果你在 Linux 服务器上跑TextLoader加载一个 GBK 编码的文件直接抛UnicodeDecodeError。from langchain_community.document_loaders import TextLoader # 指定编码避免乱码 loader TextLoader(data/example.txt, encodingutf-8) docs loader.load()如果不知道文件编码怎么办可以用chardet库检测import chardet with open(data/example.txt, rb) as f: raw f.read() result chardet.detect(raw) print(result) # {encoding: GB2312, confidence: 0.99}实测下来chardet对中文编码的检测准确率在 95% 以上但偶尔会把 GBK 误判成 GB2312。这两个编码大部分字符兼容但遇到生僻字可能出问题。稳妥的做法是检测到 GB 系列编码后统一用gb18030来解码它是 GBK 的超集兼容性最好。3. txt 文件加载看似简单细节不少3.1 TextLoader 的基本用法与参数TextLoader是 LangChain 里最简单的 Loader没有之一。它的核心参数就三个file_path、encoding、autodetect_encoding。from langchain_community.document_loaders import TextLoader loader TextLoader( file_pathdata/knowledge.txt, encodingutf-8, autodetect_encodingTrue # 自动检测编码 ) documents loader.load() print(f加载了 {len(documents)} 个文档) print(f第一个文档内容长度: {len(documents[0].page_content)}) print(fmetadata: {documents[0].metadata})autodetect_encodingTrue会让 Loader 内部用chardet检测编码省去手动指定的麻烦。但这个参数有个小坑检测本身有开销如果文件很大或者要批量加载几千个文件检测时间会累积。我的做法是先用chardet批量检测一遍把编码信息记下来后续加载直接指定不做重复检测。3.2 大文件的分块策略TextLoader默认把整个文件读成一个 Document。如果文件只有几 KB没问题。但如果是一个几十 MB 的日志文件或者长篇小说直接塞进去会导致后续 embedding 超长、检索粒度太粗。这时候需要在加载后做切分。LangChain 提供了多种 TextSplitter最常用的是RecursiveCharacterTextSplitterfrom langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) docs loader.load() split_docs text_splitter.split_documents(docs) print(f切分后得到 {len(split_docs)} 个 chunk)chunk_size500是什么概念大约 300-400 个中文字符。这个粒度适合大多数问答场景既不会太碎导致语义不完整也不会太长导致检索精度下降。chunk_overlap50是让相邻 chunk 之间有 50 个字符的重叠防止关键信息刚好被切断。separators的顺序很重要。它按优先级依次尝试用分隔符切分先尝试\n\n段落如果切出来的块还是太大再用\n换行然后是中文标点。这个顺序保证了切分尽量在语义边界发生而不是在句子中间硬切。实操心得中文文本的 separators 一定要把中文标点加进去。默认的 separators 只有英文标点和空格对中文文本来说切分点会落在很奇怪的位置。我一般会加上。、、、这几个效果立竿见影。3.3 批量加载多个 txt 文件实际项目里很少只有一个 txt 文件通常是一个目录下几十上百个文件。LangChain 提供了DirectoryLoader来批量处理from langchain_community.document_loaders import DirectoryLoader, TextLoader loader DirectoryLoader( pathdata/txt_files/, glob**/*.txt, # 递归匹配所有 txt loader_clsTextLoader, loader_kwargs{encoding: utf-8, autodetect_encoding: True}, show_progressTrue, use_multithreadingTrue, max_concurrency4 ) docs loader.load() print(f共加载 {len(docs)} 个文档)use_multithreadingTrue和max_concurrency4可以并行加载对大量小文件效果明显。但要注意如果文件在机械硬盘上并发读取反而可能因为磁头频繁寻道而变慢。SSD 上并发效果最好机械硬盘建议max_concurrency2就够了。show_progressTrue会显示进度条批量加载时很有用能直观看到处理进度。但如果在日志系统里跑进度条会刷屏这时候关掉它。4. Markdown 解析结构化文本的正确打开方式4.1 为什么 Markdown 需要专门处理Markdown 是 RAG 知识库最理想的格式之一因为它自带结构标记——标题、列表、代码块、表格都有明确的语法。如果把它当纯文本加载这些结构信息就全丢了。举个例子一个 Markdown 文件里有这样的内容## 安装步骤 ### 环境要求 - Python 3.8 - 内存 8GB 以上 ### 安装命令 bash pip install langchain如果当纯文本加载切分后可能变成“Python 3.8 内存 8GB 以上 pip install langchain”标题层级完全丢失。检索时用户问“安装环境要求是什么”可能匹配不到正确的 chunk因为 chunk 里没有“环境要求”这个标题信息。 用 MarkdownHeaderTextSplitter 就不一样了它会识别 ## 和 ###把标题作为 metadata 附加到每个 chunk 上。检索时不仅能匹配正文还能匹配标题精度提升非常明显。 ### 4.2 MarkdownHeaderTextSplitter 实战 python from langchain.text_splitter import MarkdownHeaderTextSplitter markdown_text # RAG 系统指南 ## 第一章 数据导入 ### 1.1 文本加载 使用 TextLoader 加载纯文本文件。 ### 1.2 Markdown 加载 使用 MarkdownHeaderTextSplitter 处理结构化文档。 ## 第二章 向量存储 ### 2.1 向量数据库选型 常见选择包括 Chroma、FAISS、Milvus。 headers_to_split_on [ (#, h1), (##, h2), (###, h3), ] splitter MarkdownHeaderTextSplitter( headers_to_split_onheaders_to_split_on, strip_headersFalse # 保留标题在正文中 ) chunks splitter.split_text(markdown_text) for chunk in chunks: print(内容:, chunk.page_content[:50]) print(metadata:, chunk.metadata) print(---)输出结果里每个 chunk 的 metadata 都会带上它所属的标题层级。比如“使用 TextLoader 加载纯文本文件”这个 chunk 的 metadata 是{h1: RAG 系统指南, h2: 第一章 数据导入, h3: 1.1 文本加载}。这样检索时即使用户问的是“第一章讲什么”也能通过 metadata 匹配到相关 chunk。strip_headersFalse这个参数值得说一下。默认情况下切分后标题会从正文中移除只保留在 metadata 里。但实测下来保留标题在正文中对检索效果更好因为 embedding 模型能同时看到标题和内容语义更完整。所以我一般设成False。4.3 结合 RecursiveCharacterTextSplitter 做二级切分MarkdownHeaderTextSplitter有个限制它只按标题切分不控制 chunk 大小。如果某个章节内容特别长切出来的 chunk 可能超过 embedding 模型的 token 限制。这时候需要做二级切分from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter # 第一级按标题切分 md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) md_chunks md_splitter.split_text(markdown_text) # 第二级按大小切分 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) final_chunks text_splitter.split_documents(md_chunks) print(f最终得到 {len(final_chunks)} 个 chunk) for chunk in final_chunks[:3]: print(metadata:, chunk.metadata) print(内容长度:, len(chunk.page_content)) print(---)这个两级切分策略是我目前用得最顺手的方案。第一级保证结构信息不丢失第二级保证 chunk 大小可控。split_documents会自动继承第一级产生的 metadata所以最终每个 chunk 既有标题信息又不会超长。4.4 用 UnstructuredMarkdownLoader 加载文件如果 Markdown 内容在文件里而不是字符串可以用UnstructuredMarkdownLoaderfrom langchain_community.document_loaders import UnstructuredMarkdownLoader loader UnstructuredMarkdownLoader( docs/guide.md, modesingle, # single 返回一个 Documentelements 返回多个 strategyfast # fast 速度快hi_res 精度高但慢 ) docs loader.load()modesingle把整个文件作为一个 Document 返回modeelements则按元素标题、段落、列表拆成多个 Document。做 RAG 一般用single然后自己控制切分逻辑灵活性更高。strategy参数控制解析策略。fast用简单的正则匹配速度快但可能漏掉复杂结构hi_res用更复杂的模型解析精度高但慢很多。Markdown 本身结构清晰fast就够了没必要上hi_res。5. 通用文本与结构化文档的融合处理5.1 混合格式目录的统一加载实际项目里一个知识库目录下往往 txt、Markdown、PDF、Word 混在一起。这时候需要根据文件扩展名动态选择 Loaderimport os from langchain_community.document_loaders import ( TextLoader, UnstructuredMarkdownLoader, PyPDFLoader, Docx2txtLoader ) def load_document(file_path): ext os.path.splitext(file_path)[1].lower() loader_map { .txt: lambda p: TextLoader(p, encodingutf-8, autodetect_encodingTrue), .md: lambda p: UnstructuredMarkdownLoader(p, modesingle), .pdf: lambda p: PyPDFLoader(p), .docx: lambda p: Docx2txtLoader(p), } if ext not in loader_map: raise ValueError(f不支持的文件格式: {ext}) loader loader_map[ext](file_path) return loader.load() # 批量加载目录 all_docs [] for root, dirs, files in os.walk(knowledge_base/): for file in files: file_path os.path.join(root, file) try: docs load_document(file_path) all_docs.extend(docs) print(f成功加载: {file_path} ({len(docs)} 个文档)) except Exception as e: print(f加载失败: {file_path}, 错误: {e}) print(f总计加载 {len(all_docs)} 个文档)这个模式的好处是扩展性强新增格式只需要在loader_map里加一行。异常处理也很重要批量加载时某个文件出错不应该中断整个流程而是记录错误继续处理。5.2 metadata 的统一规范不同 Loader 产生的 metadata 字段不一样。TextLoader只有sourcePyPDFLoader有source和pageUnstructuredMarkdownLoader可能有一堆额外字段。为了后续检索时能统一过滤建议在加载后做一次 metadata 规范化def normalize_metadata(doc, file_path): 统一 metadata 字段 doc.metadata[source] file_path doc.metadata[file_type] os.path.splitext(file_path)[1].lower() doc.metadata[file_name] os.path.basename(file_path) # 确保 page 字段存在 if page not in doc.metadata: doc.metadata[page] 0 return doc all_docs [normalize_metadata(doc, doc.metadata.get(source, )) for doc in all_docs]统一 metadata 之后检索时就可以按file_type过滤比如只搜 Markdown 文档或者按file_name聚合结果。这些在构建复杂 RAG 应用时非常有用。5.3 去重与质量过滤批量加载后经常遇到重复内容。比如同一个文件被复制了两份或者 PDF 和 Markdown 版本内容重叠。重复内容会导致检索结果冗余浪费 token。def deduplicate_documents(docs, similarity_threshold0.95): 基于内容哈希的简单去重 seen_hashes set() unique_docs [] for doc in docs: # 用内容的前 200 个字符做哈希 content_hash hash(doc.page_content[:200].strip()) if content_hash not in seen_hashes: seen_hashes.add(content_hash) unique_docs.append(doc) return unique_docs # 更严格的去重可以用 MinHash 或 SimHash这个简单哈希去重只能处理完全重复或开头高度相似的情况。如果要处理语义重复比如两段话意思一样但措辞不同需要上 MinHash 或 SimHash 这类局部敏感哈希算法。不过对于大多数知识库场景简单哈希去重已经能解决 80% 的问题。质量过滤也很关键。加载后要检查每个 Document 的page_content长度太短的比如少于 10 个字符直接丢弃这些通常是页眉页脚或者解析残留def filter_quality(docs, min_length10): 过滤掉内容过短的文档 return [doc for doc in docs if len(doc.page_content.strip()) min_length] all_docs filter_quality(all_docs) print(f质量过滤后剩余 {len(all_docs)} 个文档)6. 常见问题与排查技巧实录6.1 编码乱码问题速查现象可能原因解决方法中文显示为\xef\xbf\xbdUTF-8 解码失败用chardet检测真实编码中文显示为乱码字符GBK 文件用 UTF-8 读指定encodinggb18030部分字符正常部分乱码混合编码逐行检测编码分段解码读取时报UnicodeDecodeError编码不匹配加errorsignore跳过错误字节errorsignore是最后的兜底方案它会跳过无法解码的字节。但这会导致内容丢失只适合对完整性要求不高的场景。更好的做法还是正确检测编码。6.2 Markdown 切分后 metadata 丢失用MarkdownHeaderTextSplitter切分后如果再做RecursiveCharacterTextSplitter有时候 metadata 会丢。原因是split_documents方法会继承 metadata但如果你用的是split_text然后手动构造 Documentmetadata 就没了。# 错误做法metadata 丢失 texts text_splitter.split_text(md_chunk.page_content) new_docs [Document(page_contentt) for t in texts] # metadata 没了 # 正确做法用 split_documents 自动继承 new_docs text_splitter.split_documents([md_chunk]) # metadata 保留踩坑记录我一开始用split_text然后手动拼 Document结果检索时发现所有 chunk 的 metadata 都是空的排查了半天才发现是这里的问题。后来统一用split_documents再也没丢过 metadata。6.3 大文件加载内存溢出加载几百 MB 的 txt 文件时TextLoader会一次性读入内存可能导致 OOM。解决方案是流式加载from langchain_community.document_loaders import TextLoader # 流式加载逐行读取 loader TextLoader(huge_file.txt, encodingutf-8) for doc in loader.lazy_load(): # 逐块处理不一次性加载全部 process_document(doc)lazy_load()返回一个生成器每次只加载一个 Document。配合TextSplitter的split_documents也可以流式处理内存占用大幅降低。6.4 批量加载速度慢批量加载几千个文件时速度可能很慢。优化手段有几个开启多线程use_multithreadingTruemax_concurrency根据 CPU 核心数设置跳过已处理的文件记录已加载文件的哈希下次跳过用更快的 LoaderTextLoader比UnstructuredMarkdownLoader快很多如果 Markdown 结构不复杂可以直接用TextLoader加载再手动处理import hashlib import json import os def get_file_hash(file_path): 计算文件哈希 hasher hashlib.md5() with open(file_path, rb) as f: hasher.update(f.read()) return hasher.hexdigest() # 加载已处理记录 processed {} if os.path.exists(processed.json): with open(processed.json, r) as f: processed json.load(f) # 跳过已处理文件 for file_path in file_list: file_hash get_file_hash(file_path) if processed.get(file_path) file_hash: continue # 跳过 # 处理文件... processed[file_path] file_hash # 保存记录 with open(processed.json, w) as f: json.dump(processed, f)这个增量加载策略在知识库持续更新的场景下非常实用每次只需要处理新增或修改的文件不用全量重跑。6.5 Markdown 表格和代码块的特殊处理Markdown 里的表格和代码块在切分时容易被破坏。表格被从中间切断代码块被拆成两半都会影响检索质量。from langchain.text_splitter import RecursiveCharacterTextSplitter # 针对含代码块的 Markdown调整 separators splitter RecursiveCharacterTextSplitter( chunk_size800, # 调大一点容纳完整代码块 chunk_overlap100, separators[ \n\n, # 代码块边界优先 \n\n, # 段落 \n|, # 表格行 \n, # 换行 。, , , , , , ] )把代码块边界\n\n放在 separators 最前面切分时会优先在代码块之间切而不是在代码块内部切。表格行\n|也加进去尽量保持表格完整。当然如果表格特别大还是可能被切开这时候可以考虑把表格单独提取出来做特殊处理。7. 从加载到入库的完整链路演示7.1 完整代码示例把前面讲的串起来一个完整的从文件加载到切分入库的流程import os from langchain_community.document_loaders import ( TextLoader, UnstructuredMarkdownLoader, DirectoryLoader ) from langchain.text_splitter import ( MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter ) from langchain_core.documents import Document def load_and_split(knowledge_dir): 加载目录下所有文档并切分 all_docs [] # 1. 加载 txt 文件 txt_loader DirectoryLoader( knowledge_dir, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8, autodetect_encoding: True}, show_progressTrue ) all_docs.extend(txt_loader.load()) # 2. 加载 Markdown 文件 md_loader DirectoryLoader( knowledge_dir, glob**/*.md, loader_clsUnstructuredMarkdownLoader, loader_kwargs{mode: single}, show_progressTrue ) md_docs md_loader.load() # 3. Markdown 按标题切分 md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) md_chunks [] for doc in md_docs: chunks md_splitter.split_text(doc.page_content) # 继承原始 metadata for chunk in chunks: chunk.metadata.update(doc.metadata) md_chunks.extend(chunks) all_docs.extend(md_chunks) # 4. 统一二级切分 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) final_docs text_splitter.split_documents(all_docs) # 5. 质量过滤 final_docs [d for d in final_docs if len(d.page_content.strip()) 10] print(f最终得到 {len(final_docs)} 个 chunk) return final_docs # 执行 docs load_and_split(knowledge_base/)7.2 入库前的最后检查切分完成后入库前建议做一次快速检查确认 chunk 质量def inspect_chunks(docs, sample_size5): 抽样检查 chunk 质量 import random samples random.sample(docs, min(sample_size, len(docs))) for i, doc in enumerate(samples): print(f 样本 {i1} ) print(f来源: {doc.metadata.get(source, unknown)}) print(f长度: {len(doc.page_content)}) print(f内容预览: {doc.page_content[:100]}...) print() inspect_chunks(docs)重点看几个指标chunk 长度是否在预期范围内400-600 字符metadata 是否完整内容开头结尾是否在语义边界。如果发现大量 chunk 长度只有几十个字符说明切分参数需要调整如果 metadata 大量缺失说明加载或切分环节有问题。7.3 接入向量数据库切分好的 Document 列表可以直接喂给向量数据库。以 Chroma 为例from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings # 初始化 embedding 模型 embeddings HuggingFaceEmbeddings( model_nameBAAI/bge-small-zh-v1.5, model_kwargs{device: cpu}, encode_kwargs{normalize_embeddings: True} ) # 存入 Chroma vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./chroma_db, collection_nameknowledge_base ) print(f已存入 {vectorstore._collection.count()} 个向量)normalize_embeddingsTrue对 BGE 系列模型很重要它会把向量归一化到单位长度这样余弦相似度计算更准确。不同 embedding 模型对归一化的要求不一样用之前查一下模型文档。8. 一些实战中攒下来的经验数据导入和解析这件事看起来是 RAG 系统里最没技术含量的环节但实际上它决定了整个系统的上限。我踩过的坑里至少有一半跟数据质量有关。有几个经验值得单独拎出来说。第一不要迷信自动检测。autodetect_encoding和strategyhi_res这些自动化的东西在 demo 阶段很好用但生产环境一定要做人工抽检。我遇到过chardet把 UTF-8 文件误判成 GBK 的情况导致整个文件乱码而且不报错静默地产生了一堆垃圾 chunk。第二metadata 比正文更重要。正文内容 embedding 模型能理解但 metadata 是过滤和排序的唯一依据。加载阶段就要把能收集的 metadata 都收集上——文件路径、修改时间、文件类型、标题层级、页码。后面做检索优化时这些字段就是你的武器库。第三切分参数没有万能值。chunk_size500只是一个起点。技术文档可以小一点300-400叙事性内容可以大一点800-1000。最好的办法是拿一批典型 query 做检索测试看不同参数下的召回率和准确率用数据说话。第四增量更新要提前设计。知识库不是一次性的后续会不断有新文档加入、旧文档修改。如果每次更新都全量重跑成本会越来越高。用文件哈希做增量检测只处理变化的文件这个机制越早做越好。第五保留原始文件路径。metadata 里的source字段一定要存原始文件的绝对路径或可访问的 URL。检索到结果后用户往往想点进去看原文没有路径就只能干瞪眼。如果文件在对象存储里存 URL如果在本机存绝对路径。这套流程我在几个项目里反复迭代过从最开始用TextLoader一把梭到后面针对不同格式做精细化处理检索效果提升非常明显。同样的 embedding 模型和向量数据库只是把数据导入这一层做扎实了问答准确率就能从 60% 出头拉到 85% 以上。数据质量这件事投入产出比远比调模型参数高。