1. RAG 数据导入与解析的整体设计思路做 RAG 应用最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。我见过太多项目在 demo 阶段跑得挺欢一上真实文档就翻车——PDF 里的表格变成乱码、扫描件一个字都读不出来、Markdown 的层级结构全丢了。问题的根源往往不在模型而在数据进入向量库之前的那几步处理。这个系列的第一篇我聚焦在最基础但也最通用的场景纯文本 txt 和结构化 Markdown 的导入与解析。为什么从这两类开始因为它们是所有文档格式的“最大公约数”。你从网页抓下来的内容、从 PDF 提取出来的文字、从数据库导出的字段最终几乎都会落到 txt 或 Markdown 这两种形态上。把这两类吃透后面处理 PDF、Word、HTML 就是在此基础上加解析器的事。核心思路其实就一句话把非结构化的文本切成有语义边界、带元数据、可追溯的 Document 对象。LangChain 的 Document Loader 体系就是干这个的。但很多人用 Loader 只是loader.load()一把梭结果切出来的 chunk 要么把一句话拦腰截断要么把标题和正文混在一起检索时召回的全是噪音。这篇我会把从文件读取、编码处理、结构解析、分块策略到元数据注入的完整链路拆开讲每个环节都给出可复现的代码和参数选择的理由。适合谁看如果你正在搭 RAG 知识库手头有一堆 txt 笔记或 Markdown 文档要入库或者你用过 LangChain 但对其中的分块逻辑一知半解这篇能帮你少走至少两周弯路。我默认你有 Python 基础知道什么是向量库但不需要你精通 LangChain——所有代码我都会解释清楚每一步在干什么。2. 核心概念与工具选型解析2.1 为什么是 Document 对象而不是纯字符串LangChain 里所有 Loader 的产出都是Document对象不是裸字符串。这个设计很多人一开始不理解觉得多此一举。但等你做检索溯源的时候就明白了Document有两个核心字段page_content存文本内容metadata存元数据。元数据里可以放来源文件路径、页码、标题层级、创建时间等等。检索的时候向量库返回的是相似的 chunk但用户想知道“这段话出自哪个文件的哪一部分”靠的就是 metadata。如果一开始图省事直接存字符串后面想加溯源信息就得重新处理一遍全量数据代价极大。所以我的习惯是从导入的第一行代码开始就把 metadata 设计好哪怕暂时用不上。2.2 Loader 选型的三个判断维度LangChain 社区提供了大量 Loader光文本类就有TextLoader、UnstructuredFileLoader、MarkdownLoader、DirectoryLoader等。选哪个不是看哪个高级而是看三个维度判断维度说明对应选择文件格式是否单一单一格式用专用 Loader混合格式用通用 Loadertxt 用 TextLoadermd 用 UnstructuredMarkdownLoader是否需要保留结构需要保留标题层级、列表、代码块Markdown 必须用结构化解析器是否批量处理单文件还是整个目录目录用 DirectoryLoader 配合 glob 模式我实测下来的经验是能用专用 Loader 就别用通用的。UnstructuredFileLoader虽然什么都能读但它内部要判断文件类型、调用不同的解析后端速度和稳定性都不如专用 Loader。而且通用 Loader 对 Markdown 的结构保留往往不如专门的 Markdown 解析器。2.3 分块策略RAG 成败的关键一环数据导入里最容易被忽视、但对检索质量影响最大的就是分块。分块太大检索时召回的内容包含太多无关信息浪费上下文窗口分块太小语义被切碎检索出来的片段答非所问。LangChain 提供了多种 TextSplitter常用的有CharacterTextSplitter、RecursiveCharacterTextSplitter、MarkdownHeaderTextSplitter。我的选型逻辑是这样的纯文本 txt用RecursiveCharacterTextSplitter按段落、换行、句号、逗号逐级降级切分尽量保持语义完整。结构化 Markdown先用MarkdownHeaderTextSplitter按标题层级切再用RecursiveCharacterTextSplitter对过长的段落做二次切分。这里有个关键参数chunk_size和chunk_overlap。chunk_size不是越大越好也不是越小越好。我的经验值是中文文本 chunk_size 设在 500-800 字符overlap 设在 50-100 字符。为什么是这个范围因为中文一个字符承载的信息量比英文单词大500 字符大约对应 300-400 个汉字正好是一个完整段落的长度。overlap 的作用是防止关键信息正好落在切分边界上被切断50-100 字符能保证上下文的连续性。注意chunk_size 的单位是字符数不是 token 数。如果你用的是按 token 计费的 embedding 模型需要自己换算。中文大致 1 字符约等于 0.6-1 个 token具体取决于分词器。3. 纯文本 txt 的导入与解析实操3.1 编码问题是第一个坑处理 txt 文件十有八九会碰到编码问题。Windows 上创建的 txt 默认可能是 GBK 或 GB2312Linux 和 Mac 上一般是 UTF-8。如果你直接用TextLoader不加encoding参数遇到非 UTF-8 文件就会抛UnicodeDecodeError。我的处理方案是写一个编码探测函数用chardet库自动检测检测不出来再按优先级尝试import chardet def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read(10000) # 读前 10000 字节做检测 result chardet.detect(raw) encoding result[encoding] confidence result[confidence] # 置信度太低时按优先级回退 if confidence 0.7: for enc in [utf-8, gbk, gb2312, utf-16]: try: with open(file_path, r, encodingenc) as f: f.read(1000) return enc except UnicodeDecodeError: continue return encoding or utf-8这个函数先读前 10000 字节做统计检测因为chardet对短文本的检测准确率不高。如果置信度低于 0.7就按 utf-8、gbk、gb2312、utf-16 的顺序逐个尝试哪个能正常读就用哪个。实测下来这套逻辑能覆盖 95% 以上的中文 txt 文件。3.2 TextLoader 的正确用法拿到编码后用TextLoader加载就很简单了from langchain_community.document_loaders import TextLoader loader TextLoader( file_pathdata/notes.txt, encodingdetect_encoding(data/notes.txt), autodetect_encodingFalse # 我们已经手动检测了关掉自动检测 ) documents loader.load()这里有个细节autodetect_encoding参数。LangChain 的TextLoader支持自动检测编码但它内部用的也是chardet而且检测逻辑比较简单。我建议手动检测后传入encoding把autodetect_encoding设为False避免重复检测和潜在的误判。加载出来的documents是一个列表每个元素是一个Document对象。对于单个 txt 文件列表通常只有一个元素page_content是全文metadata里默认只有source字段值是文件路径。3.3 元数据注入让每个 chunk 都可追溯默认的 metadata 只有 source信息太少。我通常会在加载后手动补充元数据import os from datetime import datetime for doc in documents: doc.metadata.update({ file_name: os.path.basename(doc.metadata[source]), file_type: txt, load_time: datetime.now().isoformat(), char_count: len(doc.page_content), })这些字段看起来不起眼但在后续排查问题时非常有用。比如检索结果不理想你可以先看char_count如果某个 chunk 只有几十个字符那大概率是分块出了问题。load_time则能帮你定位是哪一批数据导入的。3.4 分块参数的计算与选择txt 文件加载后是一整块文本必须分块。我用RecursiveCharacterTextSplitterfrom langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size600, chunk_overlap80, length_functionlen, separators[\n\n, \n, 。, , , , , , ], ) chunks text_splitter.split_documents(documents)重点说separators这个参数。它的逻辑是先尝试用\n\n段落分隔切如果切出来的块还是超过chunk_size就用\n换行切再不行就用中文句号、感叹号、问号、分号、逗号最后才用空格和空字符串逐字符切。这个降级顺序保证了优先在语义边界处切分。为什么把中文标点放在空格前面因为中文文本里空格很少如果按默认的英文分隔符空格优先中文长句会被硬切。把中文标点提前能保证句子完整性。实测下来这套分隔符对中文 txt 的切分效果比默认配置好很多基本不会出现把一句话切成两半的情况。chunk_size600和chunk_overlap80是我在多个项目里调出来的经验值。600 字符大约对应 400 个汉字是一个中等长度段落的体量。80 字符的 overlap 约等于一句话的长度能保证跨块的语义连续性。当然这不是金标准你可以根据自己文档的特点微调。如果文档段落普遍很短chunk_size 可以降到 400如果段落很长可以升到 800。4. Markdown 结构化解析的完整流程4.1 为什么 Markdown 不能当纯文本处理Markdown 看起来是纯文本但它有隐含的结构#是一级标题##是二级标题-是无序列表是代码块。如果你用处理 txt 的方式处理 Markdown这些结构信息就全丢了。结构信息对 RAG 有多重要举个例子用户问“XX 功能的参数怎么配置”如果 chunk 里保留了## 配置参数这个标题检索时模型能明确知道这段内容属于配置章节回答的准确率会明显提升。反过来如果标题和正文被切散模型看到的就是一堆没有上下文的参数列表很容易答错。所以 Markdown 的处理必须分两步先按标题层级切分并保留层级信息再对过长的段落做二次切分。4.2 MarkdownHeaderTextSplitter 的层级保留机制LangChain 的MarkdownHeaderTextSplitter专门干这件事from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on [ (#, h1), (##, h2), (###, h3), (####, h4), ] markdown_splitter MarkdownHeaderTextSplitter( headers_to_split_onheaders_to_split_on, strip_headersFalse, # 保留标题在内容中 ) md_chunks markdown_splitter.split_text(markdown_text)headers_to_split_on定义了要识别的标题层级和对应的元数据键名。strip_headersFalse表示切分后标题文本保留在page_content里而不是只放在 metadata 中。我建议设为False因为标题本身携带重要语义信息保留在内容里能提升 embedding 的质量。切分后每个 chunk 的 metadata 里会带上它所属的标题层级。比如一个 chunk 在## 配置参数下面它的 metadata 就是{h1: 使用指南, h2: 配置参数}。这个层级链就是天然的上下文检索时能直接告诉模型这段内容的归属。4.3 二次切分处理超长段落MarkdownHeaderTextSplitter只按标题切如果一个标题下面的内容特别长比如一个章节有几千字切出来的 chunk 还是太大。这时候需要二次切分from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size600, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ], ) final_chunks text_splitter.split_documents(md_chunks)注意这里传入的是split_documents而不是split_text因为md_chunks是Document对象列表split_documents会保留原有的 metadata把标题层级信息带到二次切分后的每个 chunk 上。这一点非常关键——如果用了split_textmetadata 就丢了。4.4 代码块和表格的特殊处理Markdown 里的代码块和表格是两类特殊内容处理不当会严重影响检索质量。代码块的问题在于RecursiveCharacterTextSplitter的分隔符里没有针对代码块的保护机制一个长代码块可能被从中间切断导致语法不完整。我的处理方案是在分块前先把代码块提取出来单独处理import re def extract_code_blocks(text): 提取 Markdown 中的代码块返回代码块列表和替换后的文本 pattern r(\w*)\n(.*?) code_blocks [] def replacer(match): lang match.group(1) code match.group(2) placeholder f__CODE_BLOCK_{len(code_blocks)}__ code_blocks.append({lang: lang, code: code, placeholder: placeholder}) return placeholder text_without_code re.sub(pattern, replacer, text, flagsre.DOTALL) return code_blocks, text_without_code提取出来后代码块作为独立的 chunk 存入metadata 里标记content_type: code和language: python。这样检索时如果用户问的是代码相关的问题可以优先召回代码块类型的 chunk。表格的处理类似。Markdown 表格用|分隔如果被切断表头和表体会分离检索出来就是一堆没有列名的数字。我的做法是把整个表格作为一个不可分割的单元如果表格超过 chunk_size就单独存为一个 chunk不做二次切分。提示代码块和表格的独立存储会增加 chunk 数量但能显著提升特定类型问题的检索准确率。如果你的知识库以技术文档为主这个处理非常值得做。5. 批量导入与常见问题排查5.1 DirectoryLoader 批量处理目录实际项目里很少只处理一个文件通常是一整个目录。DirectoryLoader可以配合 glob 模式批量加载from langchain_community.document_loaders import DirectoryLoader, TextLoader loader DirectoryLoader( pathdata/docs, glob**/*.txt, loader_clsTextLoader, loader_kwargs{encoding: utf-8}, show_progressTrue, use_multithreadingTrue, max_concurrency4, ) documents loader.load()glob**/*.txt表示递归匹配所有子目录下的 txt 文件。use_multithreadingTrue开启多线程加载max_concurrency4控制并发数。这里并发数不建议设太高因为文件 IO 本身有瓶颈设到 4-8 就够了再高反而会因为线程切换降低效率。对于混合格式的目录既有 txt 又有 md需要分别用不同的 Loader 加载然后合并结果txt_loader DirectoryLoader(data/docs, glob**/*.txt, loader_clsTextLoader) md_loader DirectoryLoader(data/docs, glob**/*.md, loader_clsUnstructuredMarkdownLoader) all_docs txt_loader.load() md_loader.load()5.2 常见问题速查表我在实际项目中踩过的坑整理成一张表方便你对照排查问题现象可能原因排查方法解决方案UnicodeDecodeError文件编码非 UTF-8用 chardet 检测编码指定正确的 encoding 参数加载后内容为空文件路径错误或文件为空检查文件是否存在、大小是否为 0修正路径过滤空文件分块后语义断裂chunk_size 太小或分隔符不当打印 chunk 内容人工检查调大 chunk_size补充中文分隔符Markdown 标题丢失用了纯文本 Loader检查 metadata 是否有标题字段改用 MarkdownHeaderTextSplitter代码块被切断分块器不识别代码块检查 chunk 中是否有不完整代码提取代码块单独处理检索结果重复chunk_overlap 过大检查相邻 chunk 的重叠比例降低 overlap 到 50-80加载速度慢单线程处理大量文件统计文件数量和总大小开启多线程增大 max_concurrencymetadata 丢失用了 split_text 而非 split_documents检查 chunk 的 metadata改用 split_documents5.3 几个容易忽视的实操细节第一个细节是空文件和空白内容的过滤。目录里经常有一些空文件或者只有几个空格的文件加载后会生成空的 Document这些空 Document 进入向量库会污染检索结果。我的做法是在加载后统一过滤documents [doc for doc in documents if len(doc.page_content.strip()) 10]阈值设 10 是因为太短的内容比如只有一两个词作为独立 chunk 没有检索价值反而会增加噪音。第二个细节是文件路径的规范化。不同操作系统下路径分隔符不同Windows 是反斜杠Linux 和 Mac 是正斜杠。metadata 里的 source 字段如果直接存原始路径跨平台迁移时会出问题。我习惯统一转成正斜杠doc.metadata[source] doc.metadata[source].replace(\\, /)第三个细节是大文件的分批处理。如果单个 txt 文件超过 10MB一次性加载到内存再分块可能会占用大量内存。我的做法是先用TextLoader加载然后立即分块分块后的 chunk 列表比原始全文小得多内存压力会缓解。如果文件实在太大比如超过 100MB就需要用流式读取按行读取并累积到一定大小就切分而不是一次性读入。5.4 分块质量的验证方法分块做完后怎么知道效果好不好不能凭感觉。我通常用两个方法验证方法一人工抽样检查。随机抽 10-20 个 chunk看内容是否语义完整、是否有明显的截断、metadata 是否正确。这个方法虽然原始但最直接。方法二检索测试。准备 10-20 个典型问题用这些 chunk 建一个临时向量库跑一遍检索看召回的内容是否相关。如果某个问题的召回结果明显不相关就去检查对应的 chunk 是不是切分出了问题。我一般会写一个简单的检查脚本统计 chunk 的长度分布import statistics lengths [len(chunk.page_content) for chunk in chunks] print(fchunk 总数: {len(chunks)}) print(f平均长度: {statistics.mean(lengths):.0f}) print(f中位数长度: {statistics.median(lengths):.0f}) print(f最短: {min(lengths)}, 最长: {max(lengths)}) print(f标准差: {statistics.stdev(lengths):.0f})如果标准差很大说明 chunk 长度参差不齐可能有异常短的或异常长的 chunk需要进一步排查。理想情况下大部分 chunk 的长度应该集中在 chunk_size 附近标准差控制在 150 以内。6. 从导入到入库的完整链路串联6.1 一个可复用的处理管道把前面所有环节串起来形成一个完整的处理管道。这个管道我封装成了一个函数输入是目录路径输出是处理好的 chunk 列表from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter import os def build_ingestion_pipeline(directory): all_chunks [] # 第一步分别加载 txt 和 md for root, dirs, files in os.walk(directory): for file in files: file_path os.path.join(root, file) if file.endswith(.txt): encoding detect_encoding(file_path) loader TextLoader(file_path, encodingencoding) docs loader.load() splitter RecursiveCharacterTextSplitter( chunk_size600, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ] ) chunks splitter.split_documents(docs) elif file.endswith(.md): loader UnstructuredMarkdownLoader(file_path) docs loader.load() md_splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) md_chunks md_splitter.split_text(docs[0].page_content) splitter RecursiveCharacterTextSplitter( chunk_size600, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ] ) chunks splitter.split_documents(md_chunks) else: continue # 补充元数据 for chunk in chunks: chunk.metadata.update({ file_name: file, file_type: file.split(.)[-1], source: file_path.replace(\\, /), }) all_chunks.extend(chunks) # 过滤过短的 chunk all_chunks [c for c in all_chunks if len(c.page_content.strip()) 10] return all_chunks这个管道覆盖了从文件遍历、编码检测、格式区分、分块到元数据注入的全流程。你可以直接拿去用也可以根据自己的需求调整参数。6.2 入库前的最后检查chunk 准备好之后别急着往向量库里灌。先做几项检查第一去重。同一份文档可能被重复导入或者不同文档里有完全相同的段落。用内容哈希去重import hashlib seen set() unique_chunks [] for chunk in all_chunks: content_hash hashlib.md5(chunk.page_content.encode()).hexdigest() if content_hash not in seen: seen.add(content_hash) unique_chunks.append(chunk)第二检查 metadata 完整性。确保每个 chunk 都有 source、file_name、file_type 这几个关键字段缺失的补上默认值。第三预估 embedding 成本。统计总字符数按你的 embedding 模型的计费方式估算成本。中文大致 1 字符对应 0.6-1 个 token600 字符的 chunk 大约 400-600 token。如果总共有 10000 个 chunk那就是 400-600 万 token心里有个数。6.3 后续扩展方向这套管道目前只处理 txt 和 Markdown但它的架构是可扩展的。要加 PDF 支持只需要在文件类型判断里加一个分支用PyPDFLoader加载后面的分块和元数据逻辑可以复用。要加 HTML用UnstructuredHTMLLoader或者BSHTMLLoader同样复用后续流程。真正需要单独处理的是扫描件 PDF 和图片那涉及到 OCR是另一个技术栈。还有表格密集的 Excel 和 CSV需要专门的表格解析逻辑。这些我会在后续的文章里展开。我在实际项目里最大的体会是数据导入这一步花多少时间都值得。很多人急着调模型、调检索参数但源头的数据质量不行后面怎么调都是白费。把 txt 和 Markdown 这两类基础格式处理干净你的 RAG 知识库就已经赢在起跑线上了。