1. 为什么数据导入是 RAG 系统的第一道生死关做 RAG 的人都有一个共识检索效果差八成问题出在数据导入和解析环节而不是模型本身。我见过太多团队花大价钱调 embedding 模型、换向量库、折腾重排序最后发现原始文档解析出来就是一堆乱码或者断句错乱后面再怎么优化都是白搭。这个项目标题聚焦的是 RAG 数据导入与解析的第一环——从纯文本 txt 到结构化 Markdown 的通用文本与结构化解析。说白了就是把各种格式的原始文档通过 LangChain 的 Document Loader 体系统一转换成带元数据的 Document 对象并且尽可能保留原文的层级结构标题、列表、表格、代码块为后续的切分和向量化打好基础。为什么单独把 txt 和 Markdown 拎出来讲因为这两个格式是所有文档解析的最小公倍数。你从 PDF、Word、HTML 里解析出来的内容最终都要落到纯文本或类 Markdown 的结构上。如果连 txt 和 Markdown 的解析都没搞明白直接上 PDF 解析那基本就是给自己挖坑。这篇文章适合刚接触 RAG 的开发者、正在搭建知识库的技术负责人以及被文档解析折磨过的运维同学。我会把 LangChain 的 Loader 体系拆开讲透配上可直接复现的代码和踩坑记录。2. LangChain Document Loader 体系的核心设计逻辑2.1 Document 对象到底装了什么LangChain 里所有 Loader 的产出都是Document对象这个对象只有两个核心字段page_content和metadata。看起来简单但这两个字段的设计直接决定了你后面能不能做好检索。page_content是字符串存的是文档的实际文本内容。metadata是字典存的是这条内容的来源信息——文件路径、页码、标题层级、创建时间等等。很多人只关注page_content把metadata当摆设这是大错特错。在实际检索场景里metadata是你做过滤检索和结果溯源的唯一依据。比如用户问2023 年的财报里营收是多少你如果没有在 metadata 里存年份和文档类型就只能靠语义相似度硬匹配召回率会惨不忍睹。我个人的经验是metadata 的设计要在导入阶段就定好不要等到检索阶段再补。因为一旦向量化完成再想给已有的向量补 metadata就得全量重新 embedding成本极高。2.2 为什么 Loader 要分这么多种LangChain 提供了几十种 Loader从TextLoader、UnstructuredMarkdownLoader到PyPDFLoader、CSVLoader看起来冗余其实每一种都对应一类文档的解析特性。txt 文件没有结构解析逻辑最简单但编码问题最头疼。Markdown 有明确的语法结构#标题、-列表、|表格解析时要决定是保留原始 Markdown 标记还是转成纯文本。PDF 有版式信息需要处理分栏、页眉页脚、扫描件 OCR。CSV 有行列结构要决定每一行是一个 Document 还是整个表是一个 Document。这个项目标题选择从 txt 和 Markdown 入手我认为是非常务实的路径。因为这两个格式的解析逻辑是其他所有格式的基础PDF 解析出来本质上是带页码的文本HTML 解析出来本质上是带标签的文本Word 解析出来本质上是带样式的文本。你把 txt 和 Markdown 的解析吃透了其他格式只是多了一层格式转换的壳。2.3 通用解析与结构化解析的分界线标题里提到通用文本与结构化解析这其实是两种不同的处理策略。通用文本解析的目标是把内容完整取出来不关心结构产出的是连续的文本流。TextLoader就是典型代表它把整个文件读成一个字符串塞进一个 Document 里。这种方式适合内容本身没有明显层级、或者你打算用固定长度切分的场景。结构化解析的目标是把内容按层级拆开产出的是带结构信息的多个 Document 或带层级 metadata 的 Document。UnstructuredMarkdownLoader配合modeelements就是典型代表它会把每个标题、每个段落、每个列表项都拆成独立的 element并标注类型。这种方式适合需要精确定位、按章节检索的场景。选择哪种策略取决于你的检索需求。如果你做的是整篇文档问答通用解析就够了。如果你做的是精确定位到某一节那必须用结构化解析。我后面会给出两种策略的完整代码和效果对比。3. 从 txt 到 Markdown核心解析细节与实操要点3.1 TextLoader 的编码陷阱与参数配置TextLoader看起来是最简单的 Loader但它的坑一点都不少。最典型的就是编码问题。中文文档在 Windows 上经常是 GBK 或 GB2312 编码而TextLoader默认用 UTF-8 读取遇到非 UTF-8 文件直接抛UnicodeDecodeError。from langchain_community.document_loaders import TextLoader # 错误示范不指定编码遇到 GBK 文件直接崩 loader TextLoader(财报.txt) docs loader.load() # UnicodeDecodeError # 正确做法显式指定编码 loader TextLoader(财报.txt, encodingutf-8) docs loader.load() # 如果文件是 GBK需要这样处理 loader TextLoader(财报.txt, encodinggbk) docs loader.load()但问题是你不可能提前知道每个文件的编码。我的做法是写一个编码探测函数用chardet库自动识别然后传给TextLoader。import chardet from langchain_community.document_loaders import TextLoader def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read(10000) # 只读前 10KB 做探测避免大文件慢 result chardet.detect(raw) return result[encoding] file_path 财报.txt encoding detect_encoding(file_path) loader TextLoader(file_path, encodingencoding) docs loader.load()注意chardet对短文本的探测准确率不高如果文件很小小于 1KB建议直接尝试 UTF-8失败再回退到 GBK。另外TextLoader的autodetect_encoding参数在部分版本里可用但实测下来不如手动探测稳。还有一个容易被忽略的点TextLoader默认把整个文件读成一个 Document。如果你的 txt 文件有 10MB那page_content就是一个 10MB 的字符串后面切分的时候会非常慢。我的建议是在导入阶段就做一次粗切分比如按空行或按固定字符数切避免单个 Document 过大。3.2 Markdown 解析的两种模式单文档 vs 元素级Markdown 的解析比 txt 复杂因为 Markdown 本身有结构。LangChain 提供了UnstructuredMarkdownLoader它有两种模式默认模式和modeelements。默认模式下整个 Markdown 文件被读成一个 Documentpage_content是去掉 Markdown 标记后的纯文本。这种模式适合整篇问答但丢失了标题层级信息。from langchain_community.document_loaders import UnstructuredMarkdownLoader # 默认模式整个文件一个 Document loader UnstructuredMarkdownLoader(技术文档.md) docs loader.load() print(len(docs)) # 1 print(docs[0].page_content[:200]) # 纯文本无 Markdown 标记modeelements模式下每个 Markdown 元素标题、段落、列表项、代码块都被拆成独立的 Document并且 metadata 里会标注元素类型。# 元素级模式每个元素一个 Document loader UnstructuredMarkdownLoader(技术文档.md, modeelements) docs loader.load() print(len(docs)) # 可能是几十个 for doc in docs[:5]: print(doc.metadata[category], |, doc.page_content[:50])输出大概是这样Title | 第一章 系统概述 NarrativeText | 本系统采用微服务架构... Title | 1.1 核心模块 NarrativeText | 核心模块包括... ListItem | 用户管理模块这种模式的好处是标题层级被保留在 metadata 里你可以根据category做过滤比如只检索NarrativeText类型的内容跳过Title。坏处是 Document 数量暴增如果后面不做合并向量库会被大量短文本撑爆。我的实操经验是元素级解析后一定要做一次标题合并。把每个Title和它下面的NarrativeText合并成一个 Document这样既保留了层级信息又不会产生太多碎片。def merge_by_title(docs): merged [] current_title current_content [] for doc in docs: if doc.metadata[category] Title: if current_content: merged.append({ title: current_title, content: \n.join(current_content) }) current_title doc.page_content current_content [] else: current_content.append(doc.page_content) if current_content: merged.append({ title: current_title, content: \n.join(current_content) }) return merged3.3 Markdown 表格与代码块的特殊处理Markdown 里的表格和代码块是两个特殊存在。表格在UnstructuredMarkdownLoader里会被识别为Table类型但page_content里的内容是制表符分隔的文本不是 Markdown 表格语法。代码块会被识别为CodeSnippet类型内容保留原始代码。这两个类型在检索时有个共同问题语义相似度匹配效果差。表格里的数字和代码里的符号embedding 模型很难理解。我的做法是给这两类内容单独打标签在检索时要么排除要么用专门的检索策略。# 给表格和代码块单独打标签 for doc in docs: if doc.metadata[category] Table: doc.metadata[content_type] table elif doc.metadata[category] CodeSnippet: doc.metadata[content_type] code else: doc.metadata[content_type] text提示如果你的知识库里有大量表格建议在导入阶段就把表格转成自然语言描述。比如把| 年份 | 营收 |转成2023 年营收为 1000 万元。这个转换可以用 LLM 做虽然增加成本但检索效果提升非常明显。4. 完整实操流程从文件扫描到 Document 入库4.1 目录扫描与文件类型分发实际项目里你面对的不是单个文件而是一个目录树。第一步是扫描目录根据文件扩展名分发到不同的 Loader。import os from pathlib import Path from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader def scan_directory(root_dir): files [] for path in Path(root_dir).rglob(*): if path.is_file(): files.append(str(path)) return files def load_file(file_path): ext os.path.splitext(file_path)[1].lower() if ext .txt: encoding detect_encoding(file_path) loader TextLoader(file_path, encodingencoding) return loader.load() elif ext in [.md, .markdown]: loader UnstructuredMarkdownLoader(file_path, modeelements) return loader.load() else: return [] def load_directory(root_dir): all_docs [] for file_path in scan_directory(root_dir): docs load_file(file_path) # 给每个 Document 补充来源信息 for doc in docs: doc.metadata[source] file_path doc.metadata[file_name] os.path.basename(file_path) all_docs.extend(docs) return all_docs这段代码看起来简单但有几个细节要注意。rglob(*)会递归扫描所有子目录如果目录里有.git、node_modules这种无关目录会浪费大量时间。建议加一个忽略列表。IGNORE_DIRS {.git, node_modules, __pycache__, .venv} def scan_directory(root_dir): files [] for path in Path(root_dir).rglob(*): if any(ignore in path.parts for ignore in IGNORE_DIRS): continue if path.is_file(): files.append(str(path)) return files4.2 元数据标准化让每条 Document 都可溯源元数据标准化是导入阶段最容易被忽视、但后期最影响体验的环节。我建议至少包含这几个字段字段名类型说明是否必填sourcestring文件绝对路径是file_namestring文件名是file_typestring文件类型txt/md是categorystring元素类型Title/NarrativeText等结构化解析时必填title_pathstring标题层级路径如第一章 1.1 核心模块结构化解析时建议填create_timestring文件创建时间建议填content_typestring内容类型text/table/code建议填title_path这个字段特别有用。它记录了当前内容所属的完整标题路径检索时可以直接展示给用户这段内容来自《第一章 1.1 核心模块》溯源体验直接拉满。def build_title_path(docs): title_stack [] for doc in docs: if doc.metadata.get(category) Title: # 根据标题层级调整栈 level doc.metadata.get(level, 1) title_stack title_stack[:level-1] title_stack.append(doc.page_content) doc.metadata[title_path] .join(title_stack) return docs4.3 切分策略从 Document 到 Chunk 的过渡导入阶段产出的 Document 还不能直接向量化因为很多 Document 太长比如一个 10MB 的 txt。需要先切分成 Chunk。LangChain 提供了RecursiveCharacterTextSplitter它按字符递归切分优先在段落、句子边界切尽量保持语义完整。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ] ) chunks splitter.split_documents(docs)chunk_size500和chunk_overlap50是我常用的起点。chunk_size太小语义不完整太大检索精度下降。chunk_overlap是为了避免关键信息刚好被切在边界上。中文场景下separators里一定要加中文标点否则切分会在句子中间断开。注意RecursiveCharacterTextSplitter会保留原 Document 的 metadata所以切分后的每个 chunk 都带着source、title_path等信息溯源不会断。4.4 向量化与入库的衔接切分完成后就可以调 embedding 模型向量化然后存入向量库。这一步虽然不属于导入解析但导入阶段的设计直接影响这一步的效率。from langchain_community.embeddings import OllamaEmbeddings from langchain_community.vectorstores import Chroma embeddings OllamaEmbeddings(modelnomic-embed-text) vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db )这里有个经验批量向量化比逐条快得多。Chroma.from_documents内部会做批处理但如果你自己写循环逐条add_documents速度会慢好几倍。另外如果 chunk 数量超过几千建议分批入库避免内存爆掉。5. 常见问题与排查技巧实录5.1 编码乱码问题速查编码问题是 txt 解析的头号杀手。我整理了一个速查表现象可能原因解决方法UnicodeDecodeError文件非 UTF-8 编码用 chardet 探测编码中文显示为乱码编码探测错误手动指定 gbk/gb2312部分字符丢失编码不兼容转成 UTF-8 后再处理读取速度极慢文件过大分块读取或先切分我踩过最坑的一次是一个 GBK 文件被 chardet 误判为 ISO-8859-1结果中文全变成乱码但程序不报错。这种问题最难排查因为不抛异常。我的建议是导入后抽样检查随机打印几条page_content肉眼确认内容正常。5.2 Markdown 解析后 Document 数量暴增怎么办modeelements模式下一个 100KB 的 Markdown 可能产出上千个 Document。如果直接全部向量化向量库会被大量短文本比如单个列表项撑爆检索时也会返回一堆碎片。解决方法有两个。一是前面提到的标题合并把同一标题下的内容合并成一个 Document。二是过滤短文本把长度小于 20 个字符的 Document 丢掉。docs [doc for doc in docs if len(doc.page_content.strip()) 20]但过滤要小心有些短文本可能是关键信息比如是、否这种表格值。我的做法是对NarrativeText和ListItem做长度过滤对Title和Table不过滤。5.3 标题层级丢失的补救方案UnstructuredMarkdownLoader在部分版本里不会在 metadata 里标注标题层级level字段导致title_path构建失败。这时候需要自己解析 Markdown 的#数量。import re def extract_title_level(text): match re.match(r^(#)\s, text) if match: return len(match.group(1)) return None如果连category都没有那就只能退回到默认模式用正则手动提取标题然后自己构建 Document 列表。这条路虽然麻烦但可控性最强。5.4 大文件导入的内存与速度优化导入大文件时内存和速度是两个瓶颈。我的优化清单流式读取不要一次性f.read()用for line in f逐行读分批处理每处理 100 个文件就入库一次清空内存并行解析用concurrent.futures多线程解析IO 密集型任务提速明显跳过已处理文件用文件哈希做去重避免重复导入import hashlib def file_hash(file_path): hasher hashlib.md5() with open(file_path, rb) as f: for chunk in iter(lambda: f.read(8192), b): hasher.update(chunk) return hasher.hexdigest()把文件哈希存到数据库每次导入前先查哈希已存在就跳过。这个简单的机制能省掉大量重复工作尤其是在调试阶段反复导入同一批文件时。5.5 结构化解析后检索效果反而变差的排查有时候用了结构化解析检索效果反而比通用解析差。原因通常是切分粒度太细导致单个 chunk 的语义信息不足。比如一个列表项只有用户管理模块五个字embedding 出来就是一个模糊的向量匹配不到具体问题。排查思路先看检索返回的 chunk 内容如果都是很短的片段那就是切分粒度问题。解决方法是在切分前先合并把同一标题下的内容合并成一个较长的 Document再切分。或者调整chunk_size让它至少覆盖一个完整的语义单元。6. 我个人的实操体会与后续扩展方向这套从 txt 到 Markdown 的导入解析流程我在三个知识库项目里都用过最深的体会是导入阶段多花一小时做元数据标准化检索阶段能省十小时排查。很多人急着把数据灌进去看效果结果检索不准回头改导入逻辑又要全量重新向量化得不偿失。另外一个小技巧在导入阶段就做一次检索模拟。随便拿几个预期问题用刚导入的数据跑一次检索看看返回的 chunk 是不是你期望的。如果不对趁数据量还小赶紧调别等到几万条数据入库了才发现问题。这个系列后续还可以往几个方向扩展。一是 PDF 和 Word 的解析重点讲版式还原和表格提取。二是 HTML 和网页内容的解析重点讲正文提取和噪声过滤。三是多模态内容的处理比如图片 OCR 和图表理解。每一类格式都有自己的坑但底层逻辑是一样的把非结构化数据转成带元数据的结构化 Document为检索服务。把 txt 和 Markdown 这两个基础格式吃透后面的扩展就是水到渠成的事。