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 编码探测不能只靠 chardet
chardet是 Python 里最常用的编码探测库，但它在短文本上准确率堪忧。一个只有几十个字节的中文 txt，chardet 可能返回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_len=1000): 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 怎么读文件"，优先召回lang=python的代码块。
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(f"Unsupported format: {ext}") encoding = detect_encoding(file_path) with open(file_path, 'r', encoding=encoding) 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_workers=4): parser = DocumentParser() results = [] errors = [] with ThreadPoolExecutor(max_workers=max_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_tokens=512): 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 和行号排。这样后续调试和对比才有意义。
