RAG 数据导入与解析实战:从 txt 到 Markdown 的结构化处理
1. RAG 数据导入与解析的整体设计思路1.1 为什么数据导入是 RAG 系统的第一道生死关做 RAG 的人都有一个共识检索效果差八成问题出在数据导入和解析环节而不是模型本身。我见过太多团队花大价钱调 embedding 模型、换向量库、优化 prompt结果回头一看原始文档解析出来全是乱码、表格错位、段落粘连检索出来的 chunk 根本没法用。RAG 的全称是 Retrieval-Augmented Generation检索增强生成。它的核心逻辑是用户提问 → 检索相关文档片段 → 把片段作为上下文喂给大模型 → 生成回答。这个链条里检索质量直接决定了生成质量而检索质量的上限在你把文档导入系统的那一刻就已经被锁死了。打个比方RAG 系统就像一座图书馆。向量数据库是书架embedding 模型是图书管理员大模型是帮你写研究报告的专家。但如果送进图书馆的书本身就是缺页的、装订错乱的、字迹模糊的那管理员再厉害、专家再聪明也写不出靠谱的报告。数据导入与解析就是“把书整理好再上架”的过程。这一篇主要聚焦在通用文本与结构化文档的解析具体来说就是从最朴素的 txt 文件到带层级结构的 Markdown 文件怎么把它们干净、完整、有结构地导入到 RAG 流程里。选这个切入点是因为txt 和 Markdown 是所有文档格式的“最大公约数”——PDF 解析完要转成文本Word 解析完要保留结构HTML 抓下来要清洗最终都会落到纯文本或类 Markdown 的中间态。把这两种格式吃透后面处理 PDF、Word、Excel 就是举一反三的事。适合谁看如果你正在用 LangChain 搭 RAG 知识库或者准备从零做一个文档问答系统又或者你已经跑通了 demo 但发现检索效果不理想这篇内容都能给你直接的参考。不需要你精通 LangChain但至少要跑过一两个 Chain知道 Document 对象长什么样。1.2 从 txt 到 Markdown为什么解析策略要分层次很多人会问txt 和 Markdown 不都是纯文本吗直接open().read()不就完了为什么还要搞一套 Loader这个问题问到点子上了。纯文本不等于无结构文本。txt 文件确实没有显式结构标记但它的内容里隐含了结构——段落之间的空行、章节标题的独立成行、列表项的缩进、代码块的边界。这些隐含结构如果不在导入阶段提取出来后面 chunk 切分时就会把不相关的段落切到一起检索时自然召回不准。Markdown 则更进一步它用#、##、-、**这些符号显式标注了层级结构。这些标记本身就是天然的语义边界。一个##标题下面的内容大概率属于同一个主题一个代码块里的内容不应该和外面的说明文字混在一个 chunk 里。所以解析策略要分层次第一层字符编码与格式清洗。处理 BOM 头、换行符统一\r\n→\n、特殊字符转义。这一步不做后面全是坑。第二层结构识别。从文本中识别出标题、段落、列表、代码块、表格等元素并保留它们的层级关系。第三层元数据注入。给每个解析出来的 Document 对象打上来源、标题路径、时间等标签方便后续过滤和溯源。第四层分块策略适配。根据文档结构选择合适的分块方式而不是无脑按固定字符数切。LangChain 的 Document Loader 体系就是为这四层服务的。它把不同格式的文档统一转换成Document对象包含page_content和metadata两个核心字段让后续的 splitter、embedding、vectorstore 都能用同一套接口处理。统一接口是 LangChain 最大的价值之一你不需要为每种文件格式写一套下游逻辑。1.3 方案选型自研解析 vs LangChain Loader在实际项目里我一般会分两种情况选型情况一格式单一、结构简单。比如你只有几百个 txt 文件内容就是纯段落没有表格和代码。这种情况直接自己写 Python 脚本读取、清洗、切分反而比引入 LangChain 更轻量。代码量可能就几十行可控性还更强。情况二格式多样、结构复杂。比如你要同时处理 txt、Markdown、PDF、Word、HTML还要保留层级结构、表格、代码块。这种情况强烈建议用 LangChain 的 Loader 体系因为它已经内置了几十种 Loader覆盖主流格式所有 Loader 输出统一的Document对象下游不用改社区活跃遇到问题容易找到解决方案可以自定义 Loader继承BaseLoader实现自己的解析逻辑我自己的项目里通常是混合策略用 LangChain 的TextLoader和UnstructuredMarkdownLoader做基础解析然后在它们之上包一层自定义的清洗和元数据注入逻辑。这样既享受了框架的便利又保留了业务定制的空间。提示不要迷信“一个 Loader 走天下”。LangChain 的 Loader 质量参差不齐有些 Loader 对中文支持很差有些会把表格解析成乱码。选 Loader 之前一定拿真实数据跑一遍看解析结果再决定。2. 核心细节解析与实操要点2.1 txt 文件解析看似简单坑最多txt 文件是 RAG 数据导入里最容易被低估的格式。很多人觉得TextLoader一行代码就搞定了但实际跑起来问题一堆。我整理了几个最常见的坑和对应的处理方式。坑一编码问题。中文 txt 文件可能是 UTF-8、GBK、GB2312、GB18030 等编码。如果编码识别错了读出来就是乱码。LangChain 的TextLoader默认用 UTF-8遇到 GBK 文件直接报错或输出乱码。处理方式用chardet库自动检测编码或者手动指定encoding参数。我一般会写一个编码探测函数先读前 10000 字节检测编码再用检测到的编码读取全文。import chardet def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read(10000) result chardet.detect(raw) return result[encoding] encoding detect_encoding(data.txt) with open(data.txt, r, encodingencoding) as f: content f.read()坑二换行符不统一。Windows 的\r\n、Linux 的\n、老 Mac 的\r混在一起会让后续的段落切分逻辑失效。统一替换成\n是最稳妥的做法。坑三BOM 头。有些 Windows 记事本保存的 UTF-8 文件会带 BOM\ufeff这个字符会出现在文本开头影响后续处理。读取时用encodingutf-8-sig可以自动去掉 BOM。坑四段落边界模糊。txt 里段落之间可能用空行分隔也可能用缩进还可能连续多行没有空行。如果直接按\n\n切分遇到没有空行的文档就切不开。我的经验是先按空行切如果切出来的段落平均长度超过 1000 字再按单换行切同时用标点符号做二次判断。坑五超长单行。有些 txt 是从网页复制下来的整篇内容挤在一行里没有任何换行。这种情况需要先做“软换行”处理在句号、问号、感叹号后面插入换行符。import re def normalize_text(text): text text.replace(\r\n, \n).replace(\r, \n) text text.replace(\ufeff, ) text re.sub(r([。]), r\1\n, text) text re.sub(r\n{3,}, \n\n, text) return text.strip()2.2 Markdown 解析结构就是信息Markdown 的解析核心思路和 txt 完全不同。txt 是“从无结构中推断结构”Markdown 是“从显式标记中提取结构”。这个区别决定了 Markdown 的解析质量上限更高但也更依赖解析器对 Markdown 语法的支持程度。LangChain 里处理 Markdown 主要有两个 LoaderUnstructuredMarkdownLoader基于unstructured库能识别标题、列表、代码块、表格等元素输出带category元数据的 Document 列表。MarkdownHeaderTextSplitter专门按 Markdown 标题层级切分把标题路径写入 metadata非常适合做结构化 chunk。我一般会组合使用先用UnstructuredMarkdownLoader把 Markdown 解析成元素列表再用MarkdownHeaderTextSplitter按标题切分最后合并两者的元数据。MarkdownHeaderTextSplitter的用法很直观from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on [ (#, Header 1), (##, Header 2), (###, Header 3), ] splitter MarkdownHeaderTextSplitter(headers_to_split_onheaders_to_split_on) chunks splitter.split_text(markdown_content) for chunk in chunks: print(chunk.metadata) print(chunk.page_content[:100])这样切出来的每个 chunkmetadata 里都带着它所属的标题路径比如{Header 1: 第一章, Header 2: 1.1 节}。检索时如果用户问的是“1.1 节讲了什么”你可以直接用 metadata 过滤精准度比纯向量检索高得多。Markdown 解析的几个关键细节代码块要单独处理。代码块里的内容不应该和说明文字混在一个 chunk 里因为它们的语义完全不同。我一般会把代码块单独抽出来作为一个独立的 Documentmetadata 里标记type: code。表格要保留结构。Markdown 表格转成纯文本后行列关系会丢失。如果表格内容重要建议转成 HTML 或 JSON 格式保留结构或者用专门的表格解析逻辑。图片和链接要提取。和[text](url)里的 URL 和 alt 文本是有价值的信息可以提取到 metadata 里方便后续做多模态检索或引用溯源。数学公式要保护。$...$和$$...$$里的内容如果被当成普通文本切分公式会被切碎。建议在切分前用占位符替换公式切分后再还原。2.3 Document 对象的元数据设计不管用哪种 Loader最终输出的都是Document对象。page_content是文本内容metadata是元数据字典。元数据设计得好不好直接决定了后续检索的灵活度。我一般会注入这几类元数据元数据字段说明示例source文件来源路径/data/docs/intro.mdfile_type文件类型markdowntitle_path标题层级路径第一章 1.1 节chunk_index在文档中的序号3char_count字符数512created_at文档创建时间2024-01-15category内容类别code/text/table这些元数据在检索阶段非常有用。比如你可以做混合检索先用向量检索召回一批 chunk再用title_path做二次过滤把不属于目标章节的 chunk 剔除。或者做时间过滤只检索最近一年的文档。注意元数据不要塞太多。每个 chunk 的 metadata 都会存进向量库字段太多会显著增加存储和检索开销。我一般控制在 5-8 个字段只保留真正会用到的。2.4 分块策略结构感知切分优于固定长度切分分块chunking是 RAG 里最容易被忽视但影响最大的环节。很多人直接用RecursiveCharacterTextSplitter设个chunk_size1000、chunk_overlap200就完事了。这种做法在通用场景下能用但在结构化文档上就是浪费信息。我的建议是结构感知切分优先按文档结构切。Markdown 按标题层级切txt 按段落切。这样每个 chunk 的语义是完整的。结构切完后如果 chunk 太大再按长度二次切。比如一个##标题下的内容有 3000 字那就按 800-1000 字再切但保留标题路径作为 metadata。结构切完后如果 chunk 太小做合并。比如连续几个###小节都只有一两句话可以合并成一个 chunk避免碎片化。RecursiveCharacterTextSplitter支持自定义分隔符优先级可以模拟结构感知切分from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap100, separators[\n## , \n### , \n\n, \n, 。, , ], length_functionlen, )这个分隔符列表的意思是优先按##标题切其次按###标题切再按空行切再按单换行切再按句号切最后才按字符切。这样切出来的 chunk 会尽量保持结构完整。chunk_size 怎么定没有标准答案但有几个参考原则中文场景下800-1200 字符是比较稳妥的范围。太小了语义不完整太大了检索精度下降。如果 embedding 模型支持长文本比如 8192 token可以适当放大到 1500-2000 字符。代码类内容建议单独设置更小的 chunk_size300-500因为代码的语义密度高切太大反而不好检索。chunk_overlap 怎么定一般是 chunk_size 的 10%-20%。overlap 的作用是防止关键信息刚好被切在边界上。但 overlap 太大会导致重复内容多检索时召回一堆相似 chunk。我一般用 100-150 字符。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭起来。我用的 Python 版本是 3.10LangChain 相关包建议用最新稳定版。pip install langchain langchain-community langchain-text-splitters pip install unstructured markdown chardet如果你要处理 Markdown 里的表格和复杂元素还需要装unstructured[md]pip install unstructured[md]unstructured这个库依赖比较多安装时可能会遇到系统级依赖问题。在 Ubuntu 上可能需要先装libmagic-dev、poppler-utils等。如果只是处理 txt 和 Markdown其实不装unstructured也能跑用TextLoader加自定义解析就够了。3.2 通用文本加载器实现先写一个通用的文本加载函数处理编码检测、换行统一、BOM 清理import chardet from langchain_core.documents import Document def load_text_file(file_path, encodingNone): if encoding is None: with open(file_path, rb) as f: raw f.read(10000) encoding chardet.detect(raw)[encoding] or utf-8 with open(file_path, r, encodingencoding, errorsreplace) as f: content f.read() content content.replace(\r\n, \n).replace(\r, \n) content content.replace(\ufeff, ) metadata { source: file_path, file_type: text, encoding: encoding, } return Document(page_contentcontent, metadatametadata)这个函数比 LangChain 自带的TextLoader多了编码自动检测和换行统一实测下来对中文文档的兼容性好很多。3.3 Markdown 结构化解析实现Markdown 的解析分两步先按标题切分再对每个 chunk 做内容清洗和元数据增强。from langchain_text_splitters import MarkdownHeaderTextSplitter def load_markdown_file(file_path): with open(file_path, r, encodingutf-8) as f: content f.read() headers_to_split_on [ (#, h1), (##, h2), (###, h3), (####, h4), ] splitter MarkdownHeaderTextSplitter( headers_to_split_onheaders_to_split_on, strip_headersFalse, ) chunks splitter.split_text(content) documents [] for i, chunk in enumerate(chunks): title_path .join([ chunk.metadata.get(h, ) for h in [h1, h2, h3, h4] if chunk.metadata.get(h) ]) metadata { source: file_path, file_type: markdown, title_path: title_path, chunk_index: i, char_count: len(chunk.page_content), } documents.append(Document( page_contentchunk.page_content, metadatametadata, )) return documentsstrip_headersFalse这个参数很关键。如果设为True标题行会从page_content里去掉只保留在 metadata 里。但实际检索时标题本身也是重要的语义信息去掉后反而影响召回。所以我一般保留标题在正文里。3.4 代码块与表格的单独处理Markdown 里的代码块和表格如果混在普通文本里切分效果很差。我的做法是在切分前先把它们抽出来单独处理。import re def extract_code_blocks(markdown_text): pattern r(\w*)\n(.*?) blocks [] for match in re.finditer(pattern, markdown_text, re.DOTALL): lang match.group(1) or unknown code match.group(2) blocks.append({ language: lang, code: code, start: match.start(), end: match.end(), }) return blocks抽出来的代码块每个作为一个独立的 Documentmetadata 里标记type: code、language: python。这样检索时如果用户问的是代码相关问题可以优先召回代码块。表格的处理类似用正则匹配|...|格式的表格转成 JSON 或 HTML 保留结构。如果表格内容不多也可以直接转成自然语言描述比如“下表列出了 A、B、C 三列第一行是表头……”。3.5 完整流程串联把上面的步骤串起来形成一个完整的导入流程def load_documents(file_paths): all_docs [] for path in file_paths: if path.endswith(.md): docs load_markdown_file(path) elif path.endswith(.txt): doc load_text_file(path) docs [doc] else: continue all_docs.extend(docs) return all_docs def split_documents(documents, chunk_size800, chunk_overlap100): splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n## , \n### , \n\n, \n, 。, , ], ) return splitter.split_documents(documents)这个流程跑下来txt 和 Markdown 都能被解析成带元数据的 Document 列表再经过结构感知切分就可以送进 embedding 和向量库了。4. 常见问题与排查技巧实录4.1 解析结果乱码或内容缺失这是最常见的问题排查思路按优先级来现象可能原因排查方法解决方案中文显示为乱码编码识别错误用chardet检测实际编码手动指定正确编码开头多出奇怪字符BOM 头未清理检查文件前几个字节用utf-8-sig读取内容被截断文件读取缓冲区问题检查文件大小和读取方式用read()全量读取段落粘连换行符不统一检查\r\n和\n统一替换为\n特殊符号丢失编码不支持该字符用errorsreplace观察换用 UTF-8 编码我踩过最坑的一次是一批从旧系统导出的 txt 文件编码是 GB18030但文件头没有 BOMchardet检测出来是 GBK。GBK 是 GB18030 的子集大部分字符能正常解码但少数生僻字会变成乱码。后来我强制用 GB18030 读取才解决。所以编码检测不能全信关键数据要人工抽查。4.2 Markdown 标题层级识别错误MarkdownHeaderTextSplitter对标题的识别依赖正则匹配如果 Markdown 文件里标题格式不规范就会识别失败。常见的不规范情况#标题井号后没有空格——标准 Markdown 要求井号后必须有空格# 标题 #标题末尾有闭合井号——有些解析器不认标题行前面有空格——会被当成普通文本用了 Setext 风格标题和---下划线——MarkdownHeaderTextSplitter不支持处理方式在解析前先做一轮 Markdown 规范化把不规范的标题修正成标准格式。def normalize_markdown_headings(text): text re.sub(r^(#{1,6})([^#\s]), r\1 \2, text, flagsre.MULTILINE) text re.sub(r^(#{1,6}\s.?)\s*#\s*$, r\1, text, flagsre.MULTILINE) return text4.3 chunk 切分后语义不完整这个问题通常有两个原因一是chunk_size设得太小二是分隔符优先级没配好。排查方法随机抽 10 个 chunk人工读一遍看是否有“半句话被切断”的情况。如果有先调大chunk_size再检查分隔符列表里有没有把句号、问号加进去。我的一般配置是chunk_size800、chunk_overlap100、分隔符包含\n\n、\n、。、、、。这样切出来的 chunk 基本都能保持句子完整。提示如果你的文档里有大量列表项建议把\n-和\n1.也加入分隔符避免列表项被从中间切断。4.4 元数据丢失或覆盖用split_documents切分时原始 Document 的 metadata 会自动继承到子 chunk 上。但如果你在切分后又手动创建了新的 Documentmetadata 就可能丢失。每次创建 Document 时都要显式传入 metadata不要依赖默认值。另一个坑是 metadata 键名冲突。比如两个 Loader 都往 metadata 里写source字段但含义不同合并时就会覆盖。建议在 metadata 键名上加前缀比如md_source、txt_source避免冲突。4.5 性能问题大文件解析慢如果单个文件超过 10MB或者文件数量超过几千个解析速度会成为瓶颈。优化方向并行解析用concurrent.futures.ThreadPoolExecutor并行处理多个文件。IO 密集型任务用多线程就够了。增量导入记录已导入文件的 hash 值只处理新增或修改过的文件。懒加载不要一次性把所有 Document 加载到内存用生成器逐批产出。from concurrent.futures import ThreadPoolExecutor def load_documents_parallel(file_paths, max_workers4): with ThreadPoolExecutor(max_workersmax_workers) as executor: results executor.map(load_single_file, file_paths) return [doc for docs in results for doc in docs]实测下来4 线程并行比单线程快 3 倍左右。但注意不要开太多线程否则磁盘 IO 会成为新瓶颈。4.6 中文标点与英文标点混用中文文档里经常混用中英文标点比如逗号用,而不是句号用.而不是。。这会影响基于标点的切分逻辑。我的处理方式是在切分前做一轮标点规范化把英文标点转成中文标点在中文语境下或者反过来。具体转哪边取决于你的文档以哪种为主。def normalize_punctuation(text): text text.replace(,, ).replace(;, ) text text.replace(!, ).replace(?, ) return text但要注意如果文档里有代码或英文内容这个替换会破坏它们。所以标点规范化要分区域进行只对中文段落做替换代码块和英文段落跳过。5. 从解析到入库的衔接要点5.1 embedding 前的最后一道清洗Document 切分完、准备送 embedding 之前还有几件事要做去除空白 chunk有些 chunk 只有几个空格或换行符送进 embedding 纯属浪费。去除重复 chunk用 hash 去重避免同一内容被多次索引。长度过滤太短的 chunk少于 20 字符直接丢弃太长的 chunk超过模型最大长度要截断或二次切分。特殊字符清理连续多个空格、制表符、不可见字符统一清理掉。def clean_chunks(documents, min_length20, max_length2000): seen set() cleaned [] for doc in documents: content doc.page_content.strip() content re.sub(r\s, , content) if len(content) min_length or len(content) max_length: continue content_hash hash(content) if content_hash in seen: continue seen.add(content_hash) doc.page_content content cleaned.append(doc) return cleaned5.2 向量库写入的批量策略送进向量库时不要一条一条写要批量写。LangChain 的VectorStore.add_documents支持批量传入底层会做批处理。我一般每批 100-500 个 chunk具体看 embedding 模型的吞吐和 API 限流。如果是本地 embedding 模型比如 BGE、M3E批量大小可以设大一点500-1000 都行。如果是调用远程 API要注意限流批次之间加个time.sleep(0.5)。5.3 导入后的验证清单数据导入完成后一定要做验证。我一般会检查这几项检查项方法合格标准chunk 总数统计向量库记录数与预期数量偏差 5%元数据完整性随机抽 20 个 chunk 检查 metadata关键字段无缺失内容质量随机抽 10 个 chunk 人工阅读语义完整、无乱码检索效果用 5-10 个典型问题测试召回内容相关去重效果检查是否有重复内容重复率 1%这个验证清单看着简单但能拦住 80% 的低级问题。我见过太多人导入完直接上线结果用户一问就露馅。6. 一些实操心得与后续扩展6.1 我踩过的三个印象最深的坑第一个坑以为 txt 不需要解析。早期项目里我直接用open().read()读 txt结果遇到 GBK 编码的文件全是乱码排查了半天才发现是编码问题。从那以后任何文本读取我都先做编码检测。第二个坑Markdown 标题层级太深。有个文档用了#####五级标题MarkdownHeaderTextSplitter配置里只写到####结果五级标题被当成普通文本切分全乱了。后来我把标题配置扩展到六级并且加了规范化预处理。第三个坑chunk_overlap 设太大。有次为了“保险”把 overlap 设成 300chunk_size 是 800。结果检索时同一个内容被召回三四次去重后有效信息反而少了。后来老老实实按 10%-15% 设置。6.2 后续可以扩展的方向这篇主要覆盖了 txt 和 Markdown 的解析。实际项目里还会遇到 PDF、Word、Excel、HTML、PPT 等格式。后续可以沿着这几个方向扩展PDF 解析重点解决表格提取、扫描件 OCR、多栏排版的问题。推荐试试PyMuPDF和pdfplumber。Word 解析保留标题样式、表格、批注。python-docx是基础unstructured对 Word 的支持更全面。HTML 解析重点是正文提取和噪声去除。trafilatura和readability-lxml是两个好用的库。多模态文档图片、图表、公式的提取和描述。这块目前还在快速发展可以关注一些专门的多模态解析工具。6.3 一个实用的小技巧如果你不确定自己的分块策略好不好可以用一个简单的方法验证把切分后的 chunk 随机打乱然后人工尝试把它们重新拼回原文。如果能拼回去说明切分没有丢信息如果拼不回去说明有些内容在切分过程中丢失了。这个方法虽然笨但非常有效我每次调整分块参数后都会做一遍。另外建议在项目初期就建立一套解析质量评估集选 20-30 个有代表性的文档人工标注出关键信息点每次调整解析逻辑后检查这些信息点是否还能被正确提取。这套评估集能帮你快速判断改动是优化还是劣化避免凭感觉调参。数据导入与解析这件事说到底就是“把原始文档变成检索友好的 chunk”的过程。它没有太多高深的技术但细节极多每一个细节都可能成为 RAG 效果的瓶颈。把 txt 和 Markdown 这两个基础格式吃透后面的 PDF、Word 就是水到渠成的事。