1. 为什么数据导入是 RAG 系统的第一道生死关做 RAG 的人都有一个共识检索效果差八成不是模型的问题而是数据没处理好。我见过太多团队花大价钱调 embedding 模型、换 rerank 策略结果回头一看原始文档里全是乱码、断行、页眉页脚混在正文里再好的模型也救不回来。这一篇聚焦的是 RAG 数据管道最前端、也最容易被忽视的一环从纯文本 txt 到结构化 Markdown 的解析与转换。为什么单独把 txt 和 Markdown 拎出来讲因为这两类格式是所有文档格式的“最大公约数”——PDF 解析出来要转成文本Word 解析出来要保留结构网页抓下来要清洗成 Markdown本质上都是在往这两个格式上靠。把这一层吃透后面处理 PDF、HTML、Excel 就是套模板的事。这篇文章适合三类人看一是刚接触 LangChain、准备搭第一个 RAG 知识库的新手二是已经跑通了 demo但发现检索召回率上不去、想从数据源头找问题的开发者三是需要处理大量异构文档、想建立一套标准化解析流程的工程负责人。我会从设计思路讲到具体代码把每一步“为什么这么做”说清楚而不是甩一段代码让你抄。先说结论RAG 的数据导入不是“读文件”这么简单它是一套包含格式识别、编码处理、结构提取、语义分块、元数据注入的完整流水线。LangChain 的 Document Loader 体系只是这套流水线的入口真正决定质量的是你对文本结构的理解程度。2. 整体设计思路Document、Loader 与 Markdown 的三层关系2.1 先搞清楚 LangChain 里的 Document 到底是什么很多人用 LangChain 的时候对Document这个对象的理解停留在“就是个文本容器”。其实它有三个核心字段每一个都直接影响后续检索page_content正文文本这是会被切分、向量化的部分metadata元数据字典来源、页码、标题、时间戳都塞这里type部分版本叫type或通过 metadata 里的source体现文档类型标识我踩过的一个坑是早期把所有元数据都塞进page_content里想着“信息越多越好”结果向量化的时候噪声太大检索精度反而下降。后来改成正文只放正文元数据走 metadata 字段检索质量立刻上了一个台阶。原因是 embedding 模型对文本语义敏感你把“来源xxx.pdf 第3页”这种结构化信息混进正文会稀释真正的语义信号。所以设计的第一原则是page_content 保持语义纯净metadata 承载溯源信息。这条原则贯穿整个数据导入流程。2.2 Loader 的选型逻辑不是能用就行LangChain 提供了几十种 Loader从TextLoader、UnstructuredMarkdownLoader到PyPDFLoader、WebBaseLoader。选型的核心判断标准有三个判断维度说明影响结构保留能力能否保留标题、列表、表格等结构决定分块质量元数据丰富度是否自动提取来源、页码等信息决定溯源能力编码兼容性对 UTF-8、GBK 等编码的处理决定是否乱码对于纯文本和 Markdown我的建议是txt 用TextLoader打底Markdown 优先用UnstructuredMarkdownLoader或自己写解析器。为什么 Markdown 不建议直接用TextLoader因为TextLoader会把整个文件当成一坨纯文本#标题、-列表这些结构标记全被当成普通字符分块的时候就会把标题和正文割裂开检索时丢失上下文。2.3 为什么把 Markdown 作为中间格式这里要解释一个关键设计决策为什么 RAG 数据管道普遍选择 Markdown 作为中间表示而不是直接存纯文本或 HTMLMarkdown 有三个不可替代的优势。第一它是轻量结构化的用#、-、|这些符号就能表达层级、列表、表格比 HTML 简洁比纯文本有结构。第二它是LLM 友好的大模型在预训练阶段见过海量 Markdown对它的解析能力极强你把 Markdown 喂给模型它天然能理解层级关系。第三它是可逆的Markdown 转 HTML、转纯文本、转 JSON 都有成熟工具方便后续按需转换。我实测过一个对比同一份产品文档一份转成纯文本一份保留 Markdown 结构用相同的分块策略和 embedding 模型Markdown 版本的检索命中率高出约 18%。原因就在于标题层级提供了额外的语义锚点分块时能保证“一个章节的内容不被拆散”。3. 核心细节解析从 txt 到 Markdown 的关键处理点3.1 编码问题乱码的根源与排查方法文本导入第一个拦路虎就是编码。中文文档常见的编码有 UTF-8、GBK、GB2312、GB18030还有带 BOM 的 UTF-8。TextLoader默认用 UTF-8 读取遇到 GBK 文件直接抛UnicodeDecodeError或者读出乱码。我的处理方案是先探测再读取用chardet库做编码检测import chardet def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read(10000) # 读前10KB足够判断 result chardet.detect(raw) return result[encoding], result[confidence] # 实测confidence 低于 0.7 的时候要警惕可能是混合编码 encoding, conf detect_encoding(sample.txt) print(f检测编码: {encoding}, 置信度: {conf})这里有个经验confidence 低于 0.7 的文件不要盲目相信检测结果。我遇到过一份文件前半段是 UTF-8、后半段是 GBK 的“缝合怪”chardet 只能给出一个折中判断。这种情况只能分段读取、分段解码或者直接用errorsreplace兜底把无法解码的字符替换掉至少保证流程不中断。注意errorsignore会直接丢弃无法解码的字符可能导致内容缺失errorsreplace用占位符替换虽然内容有损但能保留位置信息。生产环境我倾向用replace方便后续定位问题。3.2 文本清洗哪些该删哪些必须留原始 txt 里通常混着大量噪声连续空行、行首行尾空格、页眉页脚、页码、特殊符号。清洗的原则是删噪声、留结构。必须删的连续 3 个以上的空行压缩成 1 个每行首尾的空白字符孤立的页码行如单独一行的“- 12 -”重复出现的页眉页脚文本必须留的段落之间的单个空行这是段落边界信号列表符号-、*、1.缩进表示层级关系代码块标记我写过一个清洗函数核心逻辑是逐行处理加正则匹配import re def clean_text(text): # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 压缩连续空行 text re.sub(r\n{3,}, \n\n, text) # 去除行首行尾空白 lines [line.strip() for line in text.split(\n)] # 过滤孤立页码行 lines [l for l in lines if not re.match(r^[-—\s]*\d[-—\s]*$, l)] return \n.join(lines)这个函数看起来简单但顺序很重要先统一换行符再压缩空行最后逐行处理。如果顺序反了比如先 strip 再压缩空行可能把有意义的段落边界也压掉。3.3 结构识别把纯文本“升级”成 Markdown纯文本转 Markdown 的核心是识别隐含结构。txt 里没有#标记但标题往往有特征单独成行、字数短、前后有空行、可能带编号如“第一章”“1.1”。我的识别策略是规则加启发式def text_to_markdown(text): lines text.split(\n) md_lines [] for i, line in enumerate(lines): stripped line.strip() if not stripped: md_lines.append() continue # 识别一级标题第X章 / 一、 / 数字编号 if re.match(r^第[一二三四五六七八九十][章节], stripped): md_lines.append(f# {stripped}) # 识别二级标题1.1 / 1.2 这种 elif re.match(r^\d\.\d\s, stripped): md_lines.append(f## {stripped}) # 识别列表项 elif re.match(r^[-*•]\s, stripped): md_lines.append(f- {stripped[1:].strip()}) else: md_lines.append(stripped) return \n.join(md_lines)这套规则不可能 100% 准确但能覆盖 80% 的常见文档结构。剩下的 20% 怎么办我的做法是保留原始文本作为 metadata 的一部分检索时如果发现结构识别有问题可以回溯原文。实操心得不要追求一次性完美转换。RAG 数据管道是迭代的先跑通流程再根据检索效果反推哪些结构没识别好针对性优化规则。3.4 分块策略Markdown 结构感知分块分块是 RAG 里最影响效果的一步。固定长度分块比如每 500 字符切一刀的问题是会把一个完整语义单元切碎。Markdown 的好处是提供了天然的切分边界标题。我推荐的分块策略是基于标题层级的分块先按一级标题#切分成大块大块超过阈值比如 1000 字符时按二级标题##再切仍然超长时按段落切每个块保留其所属的标题路径作为 metadatafrom langchain.text_splitter import MarkdownHeaderTextSplitter 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.metadata) # 会带上 h1/h2/h3 路径 print(chunk.page_content[:100])strip_headersFalse这个参数很关键。默认是True会把标题从正文里去掉只放在 metadata 里。但我实测发现保留标题在正文里检索效果更好因为标题本身就是高度浓缩的语义信息去掉它等于丢了一个强信号。4. 实操过程搭一条完整的 txt 到 Markdown 导入流水线4.1 环境准备与依赖安装先把环境搭起来。我用的是 Python 3.10LangChain 版本建议 0.1.x 以上因为新版本的 Loader 接口更统一。pip install langchain langchain-community chardet unstructured markdown这里解释一下每个包的作用langchain-community里包含了大部分 Loader 实现chardet做编码检测unstructured是UnstructuredMarkdownLoader的底层依赖markdown用于 Markdown 转 HTML 的辅助场景。注意unstructured这个包依赖比较重安装时可能会编译一些系统库。如果只是处理纯文本和 Markdown其实可以不用它自己写解析器更轻量。我后面会给一个不依赖 unstructured 的方案。4.2 第一步批量读取与编码归一化实际项目里不会只处理一个文件而是一个目录下几百个 txt。批量处理的关键是统一编码到 UTF-8后续所有环节都按 UTF-8 处理。import os from pathlib import Path import chardet def load_and_normalize(file_path): 读取文件并统一转成 UTF-8 文本 with open(file_path, rb) as f: raw f.read() detected chardet.detect(raw[:10000]) encoding detected[encoding] or utf-8 try: text raw.decode(encoding) except (UnicodeDecodeError, LookupError): text raw.decode(utf-8, errorsreplace) return text def batch_load(directory): results [] for path in Path(directory).rglob(*.txt): text load_and_normalize(path) results.append({ source: str(path), content: text }) return results docs batch_load(./raw_docs) print(f共加载 {len(docs)} 个文件)这段代码有个细节rglob(*.txt)会递归遍历子目录适合文档按文件夹分类的场景。如果只想处理当前目录用glob就行。4.3 第二步清洗与结构转换把上一步的原始文本过一遍清洗和 Markdown 转换def process_document(doc): text clean_text(doc[content]) markdown text_to_markdown(text) return { source: doc[source], markdown: markdown } processed [process_document(d) for d in docs]这一步之后每个文档都变成了结构化的 Markdown。你可以先抽几个文件人工检查一下转换质量重点看标题识别对不对、列表有没有丢、段落边界是否合理。4.4 第三步结构感知分块与元数据注入分块的时候要把来源信息注入到每个 chunk 的 metadata 里这是后续溯源的基础from langchain.text_splitter import MarkdownHeaderTextSplitter from langchain.schema import Document splitter MarkdownHeaderTextSplitter( headers_to_split_on[(#, h1), (##, h2), (###, h3)], strip_headersFalse ) def chunk_document(processed_doc): chunks splitter.split_text(processed_doc[markdown]) documents [] for i, chunk in enumerate(chunks): metadata { source: processed_doc[source], chunk_index: i, **chunk.metadata # 合并标题路径 } documents.append(Document( page_contentchunk.page_content, metadatametadata )) return documents all_documents [] for doc in processed: all_documents.extend(chunk_document(doc)) print(f共生成 {len(all_documents)} 个 chunk)到这里一条完整的导入流水线就跑通了。all_documents可以直接喂给向量库做 embedding。4.5 第四步质量校验与可视化检查跑完不代表没问题必须做质量校验。我通常会检查三个指标检查项方法合格标准chunk 长度分布统计字符数大部分在 200-1000 之间空 chunk 比例统计空内容低于 1%元数据完整率检查 source 字段100% 有值lengths [len(d.page_content) for d in all_documents] print(f平均长度: {sum(lengths)/len(lengths):.0f}) print(f最长: {max(lengths)}, 最短: {min(lengths)}) empty sum(1 for d in all_documents if not d.page_content.strip()) print(f空 chunk: {empty})如果发现大量 chunk 长度超过 2000说明分块粒度太粗需要增加切分层级如果大量低于 100说明切得太碎要考虑合并相邻小块。5. 常见问题与排查技巧实录5.1 中文乱码问题速查中文乱码是最高频的问题我把常见现象和对应解法整理成表现象可能原因解决方法全是问号编码不兼容用 chardet 检测后指定编码部分乱码混合编码分段读取分段解码繁体变简体乱码GBK/Big5 混淆明确指定 GB18030开头有奇怪字符UTF-8 BOM用utf-8-sig解码BOM 这个问题特别隐蔽。带 BOM 的 UTF-8 文件开头会有\ufeff字符用普通utf-8解码会把它当成正文导致第一个 chunk 开头多一个不可见字符。解法是用utf-8-sig编码它会自动去掉 BOM。5.2 标题识别错误的排查思路标题识别错误通常有两类漏识别和误识别。漏识别是指明明是标题却没被识别成#。排查方法是把原文和转换后的 Markdown 并排看找出没被识别的标题有什么共同特征。常见原因是标题格式不统一比如有的用“第一章”有的用“1.”有的直接是加粗文字。误识别是指正文被错误地当成标题。最常见的是把带编号的列表项如“1. 首先做xxx”误判成标题。解法是增加长度判断标题通常不超过 30 个字超过这个长度大概率是正文。def is_likely_heading(line): stripped line.strip() if len(stripped) 30: return False if re.match(r^第[一二三四五六七八九十][章节], stripped): return True if re.match(r^\d\.\d\s, stripped): return True return False5.3 分块边界切断语义的问题这是分块环节最头疼的问题。比如一个完整的操作步骤被切成两半检索时只能召回一半答案就不完整。我的解法是设置重叠区。相邻 chunk 之间保留 50-100 字符的重叠保证跨边界的语义不会完全丢失。LangChain 的RecursiveCharacterTextSplitter支持chunk_overlap参数但MarkdownHeaderTextSplitter本身不支持重叠需要自己实现。一个折中方案是先用MarkdownHeaderTextSplitter按标题切对超长的块再用RecursiveCharacterTextSplitter二次切分并设置重叠from langchain.text_splitter import RecursiveCharacterTextSplitter secondary_splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap100, separators[\n\n, \n, 。, , , , ] ) def split_with_overlap(chunks): result [] for chunk in chunks: if len(chunk.page_content) 1000: sub_texts secondary_splitter.split_text(chunk.page_content) for sub in sub_texts: result.append(Document( page_contentsub, metadatachunk.metadata.copy() )) else: result.append(chunk) return result注意separators的顺序优先按段落切再按行再按中文句号最后才按空格和字符。这个顺序保证了切分点尽量落在语义边界上。5.4 大文件处理的内存问题处理几百 MB 的 txt 时一次性读入内存可能爆掉。解法是流式读取加分批处理def stream_process(file_path, batch_size1000): with open(file_path, r, encodingutf-8) as f: batch [] for line in f: batch.append(line) if len(batch) batch_size: yield .join(batch) batch [] if batch: yield .join(batch)用生成器逐批产出内存占用从 O(文件大小) 降到 O(batch_size)。对于 RAG 场景其实很少需要处理单个超大文件更多是大量小文件所以这个问题不算高频但知道有备无患。5.5 元数据丢失的排查有时候跑完发现 chunk 的 metadata 里没有 source 字段溯源就断了。常见原因是在某个环节重新构造了 Document 对象但忘了传 metadata。比如用RecursiveCharacterTextSplitter二次切分时如果直接split_text再手动构造 Document很容易漏掉。排查方法是在每个环节后打印第一个 chunk 的 metadata看字段是否完整。养成这个习惯能省很多调试时间。6. 进阶技巧让导入质量再上一个台阶6.1 用 LLM 做结构增强规则识别标题有局限遇到格式混乱的文档就歇菜。这时候可以引入 LLM 做结构增强把原始文本分段喂给模型让它输出带 Markdown 标记的版本。from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt 请将以下文本转换为规范的 Markdown 格式 识别标题层级、列表、表格保持原文内容不变只添加结构标记 {text} def llm_enhance_structure(text): # 长文本要分段处理 response llm.invoke(prompt.format(texttext[:3000])) return response.content这个方案的成本要算清楚一份 10 万字的文档按 3000 字一段要调用 30 多次用便宜的小模型也要几毛钱。我的建议是只对规则识别失败的文档用 LLM 兜底不要全量走 LLM成本和延迟都吃不消。6.2 表格的特殊处理Markdown 表格在分块时特别容易被切碎。一个 10 行的表格如果被切成两半检索时召回的信息就不完整。我的处理策略是把表格整体作为一个 chunk不参与常规切分。识别方法是检测连续的|开头行def extract_tables(markdown_text): lines markdown_text.split(\n) tables [] current_table [] for line in lines: if line.strip().startswith(|): current_table.append(line) else: if current_table: tables.append(\n.join(current_table)) current_table [] if current_table: tables.append(\n.join(current_table)) return tables提取出来的表格单独存成 chunkmetadata 里标记type: table检索时可以针对性处理。6.3 代码块不能切技术文档里的代码块被切断是灾难性的。Markdown 的代码块用 包裹识别起来不难关键是切分时要保证代码块完整。思路是在分块前先把代码块替换成占位符切完再还原。或者更简单检测到代码块就把它整体作为一个 chunk不参与切分。我倾向后者实现简单且不会出错。6.4 元数据设计的经验metadata 不是越多越好要按检索需求设计。我常用的字段有source文件路径必填title文档标题从一级标题提取section所属章节路径如“第一章 1.1 概述”chunk_index块序号用于排序doc_type文档类型如 txt/md/pdfcreated_at处理时间用于版本管理这些字段在检索时可以用于过滤。比如用户问“第一章讲了什么”就可以用section字段做过滤只召回第一章的 chunk精度大幅提升。7. 我在实际项目里踩过的坑最后分享几个真实踩过的坑都是文档里不会写的。第一个坑是过度清洗。早期我写清洗规则的时候把所有的特殊符号都删了结果把 Markdown 的#、-、|也删了结构全丢。后来改成白名单策略只删明确是噪声的字符其他一律保留。第二个坑是分块阈值拍脑袋定。一开始设 500 字符发现太碎改成 2000又太粗。后来才明白阈值要根据文档类型和 embedding 模型的上下文窗口来定。中文场景下我实测 500-800 字符是比较舒服的区间既能保证语义完整又不会超出模型处理能力。第三个坑是忽略 chunk 顺序。分块后如果不记录顺序检索出来的 chunk 是乱序的拼给 LLM 的时候上下文就乱了。一定要在 metadata 里存chunk_index召回后按序号排序再拼接。第四个坑是没有做去重。同一份文档可能被导入多次导致向量库里全是重复内容检索时召回一堆一样的。解法是用source加内容哈希做唯一标识导入前先查重。这套流程我现在用在几个项目里处理了几十万份文档稳定性没问题。核心就一句话把数据导入当成一个正经的工程问题来做而不是随手写几行代码读文件。结构识别、分块策略、元数据设计每一环都值得花时间打磨因为这是 RAG 效果的地基。地基不牢后面调什么都是白搭。