RAG 系统落地时最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。我见过太多团队在 Demo 阶段用几个干净的 PDF 跑通了全流程一到真实业务场景就翻车——扫描件 OCR 乱码、表格结构丢失、Markdown 标题层级混乱、编码格式不统一导致切片后语义断裂。这些问题不会在技术选型评审时暴露但会在上线后以检索结果不相关回答质量不稳定的形式反复折磨你。这篇内容聚焦 RAG 数据管道的第一段从最朴素的 txt 纯文本到具备层级结构的 Markdown如何做通用文本与结构化解析。适合正在搭建 RAG 知识库的工程师、需要处理多格式文档的数据从业者以及想理解为什么我的 RAG 效果差的开发者。我会把解析环节的坑、选型逻辑、代码级操作和实测经验都摊开讲不绕弯子。1. 为什么 txt 和 Markdown 是 RAG 解析的起点而非终点1.1 纯文本的简单是个陷阱很多人觉得 txt 最好处理——没有格式、没有标签、读进来就是字符串。但实际项目中txt 恰恰是最容易埋雷的格式。我接手过一个企业知识库项目客户提供了 3000 多个 txt 文件来源包括系统导出日志、人工整理的 FAQ、从其他平台复制的文章。表面看都是纯文本实际打开后发现有的用 GBK 编码有的用 UTF-8 with BOM有的换行符是\r\n有的是\n还有的整个文件就是一行超长字符串没有任何换行。这些差异在人工阅读时几乎无感但进入 RAG 管道后会直接导致切片失败。比如按\n\n分段时如果文件用的是\r\n\r\n正则匹配不到整个文档会被当成一个 chunk向量化后语义被稀释检索时什么都召不回来。再比如编码问题Python 默认用 UTF-8 读取遇到 GBK 文件直接抛UnicodeDecodeError如果没做异常捕获整个批处理任务中断。所以 txt 解析的核心不是读文件而是编码探测、换行归一化、空行清理、超长段落切分这一整套预处理动作。这些动作做扎实了后续的 Markdown 解析才有稳定的输入基础。1.2 Markdown 的结构化价值被严重低估Markdown 在 RAG 场景里的地位很特殊。它比纯文本多了标题层级、列表、代码块、表格这些结构信息又比 HTML/PDF 轻量得多解析成本低。但大多数团队只是把 Markdown 当带符号的文本处理直接整篇丢进切分器白白浪费了#、##、-、|这些符号携带的语义边界。举个实际例子。一份技术文档的 Markdown 源文件里## 配置说明下面跟着### 数据库配置和### 缓存配置。如果按固定字符数切分很可能把两个子章节的内容混在一个 chunk 里检索缓存怎么配时返回的片段里一半是数据库配置大模型生成答案时容易被干扰。但如果解析时识别标题层级以###为边界切分每个 chunk 就是一个完整的配置项说明检索精度会明显提升。Markdown 的另一个价值是元数据提取。标题层级天然构成了文档的目录树你可以把##作为一级分类、###作为二级分类写入每个 chunk 的 metadata。检索时先按 metadata 过滤再向量匹配这在多产品线、多版本文档共存的场景里非常有用。1.3 从 txt 到 Markdown 的解析链路设计一个稳健的解析链路应该长这样原始文件进入后先做格式识别和编码归一化统一转成 UTF-8 的中间态然后根据文件类型分流txt 走纯文本预处理Markdown 走结构化解析最后输出统一的中间格式——我通常用带 metadata 的 JSON 或 Markdown 本身作为中间格式方便后续切片器消费。这个链路里有个关键决策要不要把 txt 也转成 Markdown。我的建议是转。原因很简单统一格式能大幅降低后续切片器的复杂度。txt 转 Markdown 的操作就是给段落加空行、给疑似标题的行加#、给列表项加-虽然粗糙但能让切片器用同一套逻辑处理所有文档。当然如果 txt 本身就是无结构的流水账强行加标题反而引入噪声这时候保持纯文本、用语义切分更合适。2. 编码探测与文本归一化的实操细节2.1 编码探测不能只靠 chardetchardet是 Python 里最常用的编码探测库但它在短文本上准确率堪忧。一个只有几十个字节的中文 txtchardet 可能返回GB2312、GBK、GB18030甚至ISO-8859-1而这些编码对中文的兼容性不同选错了就会乱码。我的做法是组合策略先用chardet探测如果置信度低于 0.8就用候选编码列表逐个尝试解码哪个能成功解码且解码后的文本中文字符占比合理就用哪个。候选列表按优先级排utf-8-sig、utf-8、gb18030、gbk、big5。注意gb18030要排在gbk前面因为它是超集能覆盖更多字符。import chardet def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read() # 先试 UTF-8 with BOM if raw.startswith(b\xef\xbb\xbf): return utf-8-sig # chardet 探测 result chardet.detect(raw) if result[confidence] 0.8 and result[encoding]: return result[encoding] # 候选编码逐个尝试 candidates [utf-8, gb18030, gbk, big5] for enc in candidates: try: text raw.decode(enc) # 检查中文字符占比 chinese_count sum(1 for c in text if \u4e00 c \u9fff) if chinese_count / max(len(text), 1) 0.1: return enc except UnicodeDecodeError: continue return utf-8 # 兜底这段代码的关键在于中文占比校验。纯英文文本用任何编码解出来都差不多但中文文本用错编码会产生大量乱码字符中文占比会异常低。设一个 0.1 的阈值能过滤掉大部分错误解码。2.2 换行符与空白字符的归一化编码搞定后下一步是归一化。Windows 的\r\n、老 Mac 的\r、Unix 的\n统一转成\n。连续多个空行压缩成一个空行。行首行尾的空白字符去掉但要注意保留 Markdown 里代码块的缩进——所以归一化要在解析之前做且不能无差别 strip。import re def normalize_text(text): # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 压缩连续空行3个以上变2个 text re.sub(r\n{3,}, \n\n, text) # 去掉行尾空白保留行首缩进 text \n.join(line.rstrip() for line in text.split(\n)) # 去掉首尾空白 text text.strip() return text这里有个细节不要用strip()处理每一行。Markdown 的代码块依赖行首缩进如果每行都 strip代码块的层级就丢了。只去行尾空白是安全的。2.3 超长段落的预切分有些 txt 文件整篇没有换行或者某个段落长达几万字。这种文本直接送进切片器如果切片器是按字符数硬切会在句子中间断开语义受损。我的做法是在归一化阶段就做一次粗切分按句号、问号、感叹号、分号这些句子边界切把超长段落拆成句子列表再按语义聚合。def split_long_paragraph(text, max_len1000): if len(text) max_len: return [text] # 按中文和英文句子边界切分 sentences re.split(r(?[。.!?;])\s*, text) chunks [] current for sent in sentences: if len(current) len(sent) max_len: current sent else: if current: chunks.append(current) current sent if current: chunks.append(current) return chunks这个预切分不是最终切片只是把超长文本拆成合理粒度的段落方便后续按语义或结构切分。max_len设 1000 是个经验值对应大约 500-700 个 token留足余量给后续的 embedding 模型。3. Markdown 结构化解析从标题层级到 chunk 元数据3.1 标题层级是天然的切分边界Markdown 解析的核心思路是把标题层级映射为文档树。#是一级节点##是二级节点以此类推。每个标题下面的内容属于该节点直到遇到同级或更高级的标题。实现上我推荐用markdown-it-py或mistune这类解析器把 Markdown 转成 AST抽象语法树然后遍历 AST 提取标题和内容。不要用正则去匹配^#{1,6}\s因为代码块里也可能出现#开头的行正则会误判。from markdown_it import MarkdownIt def parse_markdown_structure(md_text): md MarkdownIt() tokens md.parse(md_text) sections [] current_section None current_content [] for token in tokens: if token.type heading_open: # 保存上一个 section if current_section: current_section[content] \n.join(current_content).strip() sections.append(current_section) level int(token.tag[1]) # h1 - 1, h2 - 2 current_section {level: level, title: , content: } current_content [] elif token.type inline and current_section and not current_section[title]: current_section[title] token.content elif token.type inline: current_content.append(token.content) if current_section: current_section[content] \n.join(current_content).strip() sections.append(current_section) return sections这段代码输出的是一个扁平的 section 列表每个 section 带 level、title、content。后续可以根据 level 构建树形结构也可以直接按 section 切片。3.2 标题路径作为 chunk 的 metadata扁平 section 列表有个问题一个### 缓存配置的 section脱离上下文后你不知道它属于哪个##章节。解决办法是维护标题路径栈每个 section 记录从根到当前的完整路径。def build_section_paths(sections): path_stack [] for sec in sections: level sec[level] # 弹出比当前层级深的 while path_stack and path_stack[-1][level] level: path_stack.pop() path_stack.append({level: level, title: sec[title]}) sec[path] .join(item[title] for item in path_stack) return sections这样每个 section 就有了类似配置说明 缓存配置的路径。切片时把这个路径写入 chunk 的 metadata检索时可以用它做过滤生成答案时也可以把它作为上下文提示大模型。3.3 代码块、表格、列表的特殊处理Markdown 里的代码块、表格、列表不能当普通文本切。代码块被切断后无法执行表格被切断后行列错位列表被切断后层级丢失。我的处理原则是代码块和表格作为原子单元不切分列表按顶层项切分保留子项完整。在 AST 遍历时fence类型是代码块table_open到table_close之间是表格bullet_list_open到bullet_list_close之间是列表。识别这些边界把它们的内容整体提取出来作为一个独立的 chunk 或附加到所属 section。def extract_special_blocks(tokens): blocks [] i 0 while i len(tokens): token tokens[i] if token.type fence: blocks.append({type: code, content: token.content, lang: token.info}) elif token.type table_open: # 收集到 table_close table_content [] i 1 while i len(tokens) and tokens[i].type ! table_close: if tokens[i].type inline: table_content.append(tokens[i].content) i 1 blocks.append({type: table, content: \n.join(table_content)}) i 1 return blocks代码块还要注意语言标记。token.info里存的是python、bash这类语言名写入 metadata 后检索时可以按语言过滤比如用户问Python 怎么读文件优先召回langpython的代码块。4. 通用解析器的工程化封装与批量处理4.1 统一入口与格式路由实际项目里输入目录往往混杂着 txt、md、甚至 csv、json。解析器需要一个统一入口根据扩展名路由到不同的处理函数输出统一的中间格式。import os from pathlib import Path class DocumentParser: def __init__(self): self.handlers { .txt: self.parse_txt, .md: self.parse_markdown, .markdown: self.parse_markdown, } def parse(self, file_path): ext Path(file_path).suffix.lower() handler self.handlers.get(ext) if not handler: raise ValueError(fUnsupported format: {ext}) encoding detect_encoding(file_path) with open(file_path, r, encodingencoding) as f: raw f.read() text normalize_text(raw) return handler(text, file_path) def parse_txt(self, text, file_path): paragraphs [p for p in text.split(\n\n) if p.strip()] chunks [] for para in paragraphs: for sub in split_long_paragraph(para): chunks.append({ content: sub, metadata: {source: file_path, type: txt} }) return chunks def parse_markdown(self, text, file_path): sections parse_markdown_structure(text) sections build_section_paths(sections) chunks [] for sec in sections: if not sec[content].strip(): continue chunks.append({ content: f{sec[title]}\n{sec[content]}, metadata: { source: file_path, type: markdown, path: sec[path], level: sec[level] } }) return chunks这个封装的好处是新增格式只需加一个 handler不影响已有逻辑。metadata 里保留 source 和 type方便后续溯源和过滤。4.2 批量处理的并发与容错几千个文件的批量处理串行跑太慢但并发也不能无脑开线程池。我的经验是IO 密集型用线程池CPU 密集型用进程池。解析主要是 IO 和字符串操作用ThreadPoolExecutor开 4-8 个线程就够了开太多反而因为 GIL 和磁盘 IO 竞争导致性能下降。容错方面单个文件解析失败不能中断整个批次。用 try-except 包住每个文件的处理失败的文件记录到错误日志继续处理下一个。from concurrent.futures import ThreadPoolExecutor, as_completed def batch_parse(file_paths, max_workers4): parser DocumentParser() results [] errors [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_path {executor.submit(parser.parse, p): p for p in file_paths} for future in as_completed(future_to_path): path future_to_path[future] try: chunks future.result() results.extend(chunks) except Exception as e: errors.append({path: path, error: str(e)}) return results, errors错误日志要记录文件路径和异常信息方便后续人工排查。我通常会把失败文件单独复制到一个failed/目录修完编码或格式后重新跑。4.3 解析质量的抽检与验证批量解析完不能直接进切片器要先抽检。抽检的维度包括chunk 数量是否合理太少说明切分粒度太粗太多说明太细、metadata 是否完整、内容是否有乱码、代码块和表格是否完整。我一般写一个简单的统计脚本输出每个文件的 chunk 数、平均 chunk 长度、metadata 字段缺失率。如果某个文件的 chunk 数异常比如一个 10KB 的 md 只切出 1 个 chunk就要单独看它的解析结果。def quality_check(chunks): from collections import defaultdict stats defaultdict(lambda: {count: 0, total_len: 0, missing_meta: 0}) for chunk in chunks: source chunk[metadata].get(source, unknown) stats[source][count] 1 stats[source][total_len] len(chunk[content]) if not chunk[metadata].get(path) and chunk[metadata].get(type) markdown: stats[source][missing_meta] 1 for source, s in stats.items(): avg_len s[total_len] / max(s[count], 1) print(f{source}: {s[count]} chunks, avg {avg_len:.0f} chars, missing meta {s[missing_meta]})平均 chunk 长度在 300-800 字符之间比较健康低于 200 说明切太碎高于 1500 说明切太粗。metadata 缺失率高的话要检查解析逻辑是不是漏了某些 section。5. 解析环节的常见坑与排查链路5.1 乱码问题的完整排查过程乱码是解析环节最高频的问题。我遇到过一次客户提供的 txt 文件用chardet探测出来是GB2312但解码后部分生僻字还是乱码。排查链路是这样的第一步确认乱码位置。用十六进制查看器打开文件找到乱码对应的字节序列。发现是0x81 0x40这种双字节GB2312里没有这个组合。第二步换编码尝试。用GBK解码还是乱码用GB18030解码正常了。原因是GB18030是GBK的超集覆盖了更多生僻字和少数民族文字。第三步修正探测逻辑。在候选编码列表里把GB18030提到GBK前面并且降低chardet的置信度阈值让更多文件走候选编码尝试流程。这个坑的教训是中文编码探测不能只信 chardet要有候选编码兜底且 GB18030 优先级要高于 GBK。5.2 Markdown 标题识别失败的几种情况Markdown 解析时标题识别失败通常有三种原因。第一种是标题符号和文字之间没有空格比如##配置说明而不是## 配置说明。标准 Markdown 要求#后面必须有空格但很多人生成的内容不规范。解决办法是在解析前做一次预处理用正则给^#{1,6}[^#\s]的行插入空格。第二种是标题出现在代码块里。比如一段 shell 脚本的注释# 这是注释被误识别为一级标题。这就是为什么不能用正则而要用 AST 解析器——AST 能区分代码块和正文。第三种是 Setext 风格标题用和---下划线表示一级和二级标题。这种写法现在少见但老文档里还有。markdown-it-py默认支持但如果你自己写解析逻辑要额外处理。5.3 表格和代码块被切断的修复表格被切断的典型表现是检索结果里出现半截表格只有表头没有数据行或者行列错位。根因是切片器按字符数硬切没识别表格边界。修复方案是在解析阶段就把表格提取为独立 chunk并在 metadata 里标记type: table。切片器遇到type: table的 chunk 直接跳过不参与二次切分。代码块同理标记type: code切片器跳过。如果表格特别大比如几百行一个 chunk 放不下可以按行切分但每个子 chunk 都要带上表头。这个逻辑要在解析阶段做不能留给切片器。5.4 元数据丢失的排查元数据丢失通常发生在格式转换环节。比如 txt 转 Markdown 时如果只是简单加#没有保留原始文件名和路径后续 chunk 的 metadata 里 source 字段就是空的。排查方法是在解析链路的每个环节打印 metadata看在哪一步丢的。我的习惯是在解析器入口、格式转换后、切片前各打一次日志对比 metadata 字段的变化。如果切片前还有、切片后没了说明是切片器的问题如果格式转换后就没了说明是转换逻辑没传递 metadata。修复原则是metadata 要像接力棒一样在各个环节传递不能中途重新构造。每个处理函数接收上游的 metadata补充自己的字段再传给下游。6. 从解析到切片的衔接策略6.1 解析输出的中间格式设计解析器的输出直接决定切片器的输入。我推荐的中间格式是一个 JSON 列表每个元素包含content和metadata两个字段。content是文本内容metadata是字典至少包含source、type、path三个字段。这个格式的好处是与切片器解耦。切片器不需要知道原始文件是 txt 还是 md只需要处理 content 和 metadata。新增格式时只要解析器输出同样的格式切片器不用改。如果后续要接入向量数据库这个格式也能直接映射。content送 embedding 模型metadata作为 payload 存储检索时按 metadata 过滤。6.2 切片粒度与解析粒度的匹配解析粒度和切片粒度要匹配。如果解析时已经按###切成了 section切片器就不应该再按字符数硬切而应该把每个 section 作为一个 chunk或者只在 section 内部做语义切分。我的做法是在 metadata 里标记parsed_level表示这个 chunk 是解析阶段切出来的切片器看到这个标记就跳过硬切逻辑只做长度校验——如果超过 embedding 模型的最大长度才做二次切分。def slice_chunks(chunks, max_tokens512): final_chunks [] for chunk in chunks: if chunk[metadata].get(parsed_level): # 解析阶段已切好只做长度校验 if estimate_tokens(chunk[content]) max_tokens: final_chunks.append(chunk) else: # 超长做语义切分 sub_chunks semantic_split(chunk[content], max_tokens) for sub in sub_chunks: final_chunks.append({ content: sub, metadata: {**chunk[metadata], sub_split: True} }) else: # 未解析的走常规切片 final_chunks.extend(regular_split(chunk, max_tokens)) return final_chunks6.3 解析质量对检索效果的影响验证解析质量好不好最终要看检索效果。我通常做一个小规模验证准备 20-30 个典型问题用解析后的 chunk 建索引跑一遍检索看 Top-5 结果里有多少是相关的。如果检索效果差先排查解析环节。常见原因包括chunk 太长导致语义稀释、metadata 缺失导致无法过滤、代码块和表格被切断导致内容不完整。逐个修复后再跑验证对比修复前后的召回率变化。这个验证不需要标注数据人工看 Top-5 结果就能判断。我一般会记录修复前后的对比作为解析器迭代的依据。7. 一些实战中攒下来的经验解析器的健壮性比性能重要。我宁愿解析慢一点也不愿意因为一个文件解析失败导致整批任务中断。所以每个环节都要有 try-except每个异常都要记录日志每个失败文件都要能单独重跑。编码探测的候选列表要定期更新。不同来源的文件编码分布不同我一般会在项目初期跑一遍全量文件的编码统计看看主要编码类型然后调整候选列表的优先级。Markdown 解析不要自己写正则。我早期图省事用正则匹配标题结果被代码块里的#坑了好几次。后来换成markdown-it-py虽然多了一个依赖但省了无数排查时间。metadata 的设计要提前想清楚。哪些字段用于过滤哪些字段用于展示哪些字段用于溯源在解析阶段就要规划好。后期补 metadata 比前期设计麻烦得多因为要重新跑一遍全量解析。解析日志要保留原始文件路径和行号。出问题时能快速定位到具体文件的哪一行排查效率会高很多。我一般会在 chunk 的 metadata 里加source_line字段记录这个 chunk 对应原文的起始行号。最后解析器的输出要可复现。同样的输入文件跑两次解析输出的 chunk 应该完全一致。如果用了随机数或并发导致顺序不稳定要在输出前做一次排序按 source 和行号排。这样后续调试和对比才有意义。