1. 为什么 RAG 的第一步永远是“把数据喂干净”
做 RAG 的人都有一个共识:检索效果差,八成不是模型不行,而是数据没处理好。我见过太多人一上来就折腾向量库选型、Embedding 模型对比、重排序策略,结果召回的内容驴唇不对马嘴,回头一查原始数据——PDF 里全是断行、表格错位、页眉页脚混进正文,这种料喂给再强的模型也白搭。
这一篇只聊 RAG 数据导入与解析里最基础、也最容易被跳过的一环:通用文本(txt)和结构化文档(Markdown)的解析与清洗。为什么先讲这两种?因为它们构成了知识库的“地基格式”。txt 是最原始的纯文本载体,几乎所有格式最终都能降级成它;Markdown 则是带轻量结构标记的文本,既能保留标题层级、列表、代码块这些语义信息,又不像 HTML/PDF 那样解析起来一堆坑。把这两种吃透,后面处理 PDF、Word、Excel、网页就有了统一的参照系。
这篇文章适合谁看?如果你正在搭自己的 RAG 知识库,卡在“文档丢进去检索不出来”这一步;或者你打算写一个通用的文档导入管道,但不确定每种格式该怎么切、怎么洗;再或者你只是好奇“为什么我导入的 txt 检索效果这么差”——那这篇就是给你写的。我会把解析思路、切分策略、清洗规则、踩过的坑全部摊开讲,代码能直接抄。
先说一个核心判断:RAG 的数据导入不是“读文件”,而是“把非结构化信息重构成机器可检索的语义单元”。读文件谁都会,open().read()一行搞定,但那只是拿到了字符串。真正决定检索质量的是:这段字符串怎么切、切完保留什么元数据、脏数据怎么过滤、结构信息怎么不丢。下面按这个逻辑一层层拆。
2. 通用文本解析:txt 看着简单,坑全在编码和切分上
2.1 txt 解析的第一个拦路虎:编码识别
很多人觉得 txt 最好处理,直接读就行。我一开始也这么想,直到有一次导入一批中文小说 txt,检索出来全是乱码,排查半天发现是 GBK 编码被当成 UTF-8 读了。txt 没有自描述编码信息,这是它最大的坑。
处理方案是先探测再解码。Python 里我常用charset-normalizer(chardet的继任者,维护更活跃),它能给出编码置信度:
from charset_normalizer import from_path def detect_encoding(file_path): result = from_path(file_path).best() if result is None: return "utf-8" # 兜底 return result.encoding def read_txt(file_path): encoding = detect_encoding(file_path) with open(file_path, "r", encoding=encoding, errors="replace") as f: return f.read()注意errors="replace"这个参数。探测不可能 100% 准,遇到个别无法解码的字节,用 replace 替换成占位符,比直接抛异常中断整个导入流程要好。但这里有个经验:如果替换字符(\ufffd)占比超过 1%,说明编码探测大概率错了,应该报警而不是静默吞掉。我一般会加一个统计:
def read_txt_safe(file_path): encoding = detect_encoding(file_path) with open(file_path, "r", encoding=encoding, errors="replace") as f: text = f.read() bad_ratio = text.count("\ufffd") / max(len(text), 1) if bad_ratio > 0.01: raise ValueError(f"编码探测可能失败: {file_path}, 异常字符占比 {bad_ratio:.2%}") return text提示:中文场景下最常见的三种编码是 UTF-8、GBK、GB18030。GB18030 是 GBK 的超集,遇到疑似 GBK 的文件,直接用 GB18030 解码兼容性更好。
2.2 换行符统一:别让\r\n毁掉你的切分
Windows 出来的 txt 是\r\n,Linux/Mac 是\n,老 Mac 是\r。如果你按\n切段落,\r\n会在每行末尾留一个\r,检索时匹配不上。统一处理:
text = text.replace("\r\n", "\n").replace("\r", "\n")这一步看着不起眼,但我在实际项目里遇到过因为\r残留导致 BM25 分词结果异常的情况,排查了两小时。导入管道里所有文本进切分器之前,必须先做换行归一化,这是铁律。
2.3 切分策略:固定长度 vs 语义切分
txt 没有结构标记,切分只能靠启发式规则。常见三种:
| 策略 | 做法 | 优点 | 缺点 |
|---|---|---|---|
| 固定字符切分 | 每 N 字符一刀 | 实现简单、块大小均匀 | 容易切断句子、语义不完整 |
| 递归字符切分 | 按段落→句子→字符逐级降级 | 尽量保持语义边界 | 块大小不均 |
| 语义切分 | 用 Embedding 判断句子相似度断点 | 语义最完整 | 慢、成本高 |
我的建议是默认用递归字符切分,LangChain 的RecursiveCharacterTextSplitter就是这个思路,分隔符优先级设为["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""]。中文场景一定要把中文标点加进去,默认的英文分隔符对中文几乎无效。
关于块大小,很多人纠结 chunk_size 设多少。我的经验值:中文 300~500 字,英文 500~800 字符。为什么?因为主流 Embedding 模型(如 bge、m3e)的上下文窗口通常在 512 token 左右,中文一个字约 1~1.5 token,500 字差不多到上限。设太大,超出部分被截断,信息丢失;设太小,一个完整语义被切碎,检索出来是残句。
overlap 我一般设 chunk_size 的 10%~15%,也就是 50 字左右。overlap 的作用是防止关键信息正好落在切分点上被割裂,但设太大又会导致大量重复内容进库,浪费存储还拉低检索精度。
from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=400, chunk_overlap=50, separators=["\n\n", "\n", "。", "!", "?", ";", ",", " ", ""], length_function=len, ) chunks = splitter.split_text(text)2.4 元数据:切完不记来源,检索出来就是孤儿
这是新手最容易忽略的一点。切完的 chunk 如果不带元数据,检索出来你根本不知道它来自哪个文件、哪一段。元数据至少要包含:
source:文件路径或文件名chunk_index:在原文中的序号char_start/char_end:字符偏移量,方便回溯原文file_type:txt / md / pdf 等
chunks_with_meta = [] offset = 0 for i, chunk in enumerate(chunks): start = text.find(chunk, offset) chunks_with_meta.append({ "text": chunk, "metadata": { "source": file_path, "chunk_index": i, "char_start": start, "char_end": start + len(chunk), "file_type": "txt", } }) offset = start + len(chunk)注意:
text.find在 chunk 有重复内容时可能定位不准,更稳妥的做法是用 splitter 返回的create_documents接口,它内部会维护偏移量。但如果你自己手写切分,记得处理这个边界。
3. Markdown 解析:结构信息是宝藏,别当纯文本读
3.1 为什么 Markdown 不能按 txt 处理
Markdown 的价值在于它用极低的成本携带了结构语义:#是标题层级,-是列表,```是代码块,|是表格。如果你把它当纯文本切,这些标记要么被当成噪声,要么被切得七零八落,标题和它下面的正文分到不同 chunk,检索时上下文就断了。
正确做法是先解析成 AST(抽象语法树),再按结构切分。Python 里我用markdown-it-py或mistune,它们能把 Markdown 解析成 token 流,每个 token 带类型和层级信息。
from markdown_it import MarkdownIt md = MarkdownIt() tokens = md.parse(markdown_text) for token in tokens: print(token.type, token.tag, token.level, token.content[:50])输出会类似heading_open h1 0、inline None 1 标题内容、paragraph_open p 0这样。有了这个,你就能知道每个内容块属于哪个标题下。
3.2 按标题层级切分:让每个 chunk 自带“面包屑”
我的做法是以标题为切分锚点,把标题路径作为元数据附加到 chunk 上。比如一个 chunk 来自“## 3. Markdown 解析 > ### 3.2 按标题层级切分”,那它的元数据里就记heading_path: "3. Markdown 解析 > 3.2 按标题层级切分"。检索时把这个路径拼到 chunk 前面,能显著提升召回准确率,因为标题本身就是高度浓缩的语义。
def split_markdown_by_heading(md_text): tokens = MarkdownIt().parse(md_text) chunks = [] heading_stack = [] # 维护当前标题路径 current_content = [] for token in tokens: if token.type == "heading_open": # 遇到新标题,先把之前累积的内容存起来 if current_content: chunks.append({ "text": "\n".join(current_content), "heading_path": " > ".join(heading_stack), }) current_content = [] level = int(token.tag[1]) # h1 -> 1 # 更新标题栈 heading_stack = heading_stack[:level-1] # 下一个 inline token 是标题文本 elif token.type == "inline" and heading_stack is not None: if token.level == 1 and not current_content: heading_stack.append(token.content) elif token.type in ("paragraph_open", "fence", "table_open"): current_content.append(token.content if token.content else "") if current_content: chunks.append({ "text": "\n".join(current_content), "heading_path": " > ".join(heading_stack), }) return chunks上面是简化版逻辑,实际用的时候建议直接用langchain的MarkdownHeaderTextSplitter,它已经把这套逻辑封装好了:
from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "h1"), ("##", "h2"), ("###", "h3"), ] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) docs = splitter.split_text(md_text) # 每个 doc 的 metadata 里会带 h1/h2/h3 字段3.3 代码块和表格:特殊内容特殊对待
Markdown 里的代码块(```)和表格(|)如果被普通切分器处理,很容易被切断。我的处理原则是:
- 代码块整体保留,不切分。一个代码块通常是一个完整逻辑单元,切断了就没法用。如果代码块超过 chunk_size,宁可单独成块也不切。
- 表格转成文本描述。表格直接进向量库检索效果很差,因为列名和单元格是分离的。我一般把表格转成“列名: 值”的键值对文本,或者用 LLM 生成一句表格摘要。
def table_to_text(table_token): # 简化示例:把 markdown 表格转成自然语言描述 lines = table_token.content.strip().split("\n") headers = [h.strip() for h in lines[0].strip("|").split("|")] rows = [] for line in lines[2:]: cells = [c.strip() for c in line.strip("|").split("|")] row_desc = ",".join(f"{h}为{c}" for h, c in zip(headers, cells)) rows.append(row_desc) return ";".join(rows)3.4 数学公式:别让$符号干扰解析
Markdown 里的数学公式($...$行内、$$...$$块级)在解析时容易被当成普通文本,$和\这些符号还会干扰后续处理。我的做法是在解析前先把公式提取出来,用占位符替换,解析完再还原。这样公式内容不会被切分器破坏,检索时也能作为独立单元。
import re def extract_math(text): math_blocks = [] def replacer(match): math_blocks.append(match.group(0)) return f"__MATH_{len(math_blocks)-1}__" # 先匹配块级公式,再匹配行内公式 text = re.sub(r"\$\$(.+?)\$\$", replacer, text, flags=re.DOTALL) text = re.sub(r"\$(.+?)\$", replacer, text) return text, math_blocks def restore_math(text, math_blocks): for i, block in enumerate(math_blocks): text = text.replace(f"__MATH_{i}__", block) return text提示:如果你的知识库涉及大量公式(比如技术文档、论文),建议把公式单独存一份,检索时用专门的公式检索方案,不要和普通文本混在一起。
4. 从解析到入库:完整管道怎么串
4.1 统一入口:按文件类型分发解析器
一个健壮的导入管道应该有一个统一入口,根据文件扩展名分发到不同解析器,输出统一的数据结构。我定义的结构是:
@dataclass class ParsedChunk: text: str metadata: dict # metadata 至少包含 source, chunk_index, file_type分发逻辑:
PARSERS = { ".txt": parse_txt, ".md": parse_markdown, ".markdown": parse_markdown, } def parse_file(file_path): ext = os.path.splitext(file_path)[1].lower() parser = PARSERS.get(ext) if parser is None: raise ValueError(f"不支持的文件类型: {ext}") return parser(file_path)这样后面加 PDF、Word 解析器,只要往PARSERS里注册就行,主流程不用动。
4.2 清洗规则:哪些内容该丢,哪些该留
解析出来的文本不能直接入库,得先清洗。我总结了一份清洗清单,按优先级排列:
| 清洗项 | 处理方式 | 是否默认开启 |
|---|---|---|
| 多余空白 | 连续空格/换行压缩成一个 | 是 |
| 页眉页脚 | 按行频统计,高频重复行删除 | 是(PDF 场景) |
| 控制字符 | 删除\x00-\x08等不可见字符 | 是 |
| 超短块 | 少于 20 字的 chunk 合并到相邻块 | 是 |
| 纯符号块 | 只有标点/数字的块删除 | 是 |
| 重复内容 | 相似度 > 0.95 的块去重 | 可选 |
超短块合并这个特别重要。我见过很多知识库检索出来一个 chunk 就俩字“如下:”,完全没用。合并逻辑:
def merge_short_chunks(chunks, min_len=20): merged = [] buffer = "" for chunk in chunks: if len(chunk["text"]) < min_len: buffer += chunk["text"] else: if buffer: chunk["text"] = buffer + chunk["text"] buffer = "" merged.append(chunk) if buffer and merged: merged[-1]["text"] += buffer return merged4.3 去重:别让同一段内容进库十遍
重复内容对 RAG 是灾难。同一段文本进库多次,检索时全被它占满,其他相关内容反而排不上。去重我分两层:
- 精确去重:用文本的 hash(如 MD5)做 key,完全相同的直接丢。
- 近似去重:用 SimHash 或 MinHash,相似度超过阈值的保留一个。
import hashlib def exact_dedup(chunks): seen = set() result = [] for chunk in chunks: h = hashlib.md5(chunk["text"].encode("utf-8")).hexdigest() if h not in seen: seen.add(h) result.append(chunk) return result近似去重成本高,我一般只在数据源本身有大量重复(比如多个版本的同一文档)时才开。
4.4 入库前的最后一道关:质量抽检
管道跑完别急着全量入库,先抽 20~30 个 chunk 人工看一眼。重点检查:
- 有没有乱码
- 有没有被切断的句子
- 元数据是否完整
- 标题路径是否正确
我踩过的坑:有一次 Markdown 解析器把代码块里的#当成标题,导致标题栈错乱,所有 chunk 的 heading_path 全错了。这种问题不抽检根本发现不了,等检索效果差再回头查,成本高十倍。
5. 常见问题与排查技巧实录
5.1 检索效果差,怎么定位是不是解析的锅
排查顺序我一般这样走:
- 看召回内容本身:检索出来的 chunk 是不是完整句子?有没有乱码?如果 chunk 本身就是残句,那问题在切分。
- 看元数据:source 对不对?heading_path 有没有?如果元数据缺失,问题在解析阶段。
- 看重复率:随机抽 100 个 chunk,统计有多少是重复或高度相似的。超过 10% 说明去重没做好。
- 看块大小分布:如果大量 chunk 长度远小于 chunk_size,说明切分器没生效或者分隔符设置有问题。
5.2 常见问题速查表
| 现象 | 可能原因 | 解决方向 |
|---|---|---|
| 检索结果全是乱码 | 编码探测错误 | 检查 charset-normalizer 结果,手动指定编码 |
| chunk 被从句子中间切断 | 分隔符列表缺中文标点 | 加入。!?;, |
| 标题和正文分到不同 chunk | 按固定长度切分 | 改用 MarkdownHeaderTextSplitter |
| 代码块被切碎 | 普通切分器不识别代码块 | 代码块整体保留,不切 |
| 表格检索不出来 | 表格被当普通文本切 | 表格转文本描述或单独处理 |
| 同一内容检索出多条 | 去重没做 | 加 MD5 精确去重 |
| 短 chunk 太多 | 切分过细 | 合并超短块,或调大 chunk_size |
| 公式符号干扰解析 | $被当普通字符 | 解析前提取公式,占位符替换 |
5.3 几个我踩过的坑
坑一:errors="ignore"比errors="replace"更危险。ignore 会直接丢掉无法解码的字节,导致文本内容缺失,而且你完全不知道丢了什么。replace 至少留个占位符,能统计出来。
坑二:Markdown 的---分隔线被当成标题。有些解析器会把---识别成 setext 标题的下划线,导致后面一行被误判为标题。处理办法是在解析前把独立的---行替换成空行。
坑三:chunk_overlap 设太大导致检索结果重复。overlap 是为了防止边界信息丢失,但如果设成 chunk_size 的 50%,相邻 chunk 有一半内容重复,检索时这两条会同时被召回,浪费上下文窗口。10%~15% 足够。
坑四:忘了处理 BOM。UTF-8 with BOM 的文件开头会有\ufeff,这个字符会混进第一个 chunk,影响检索。读取时用encoding="utf-8-sig"可以自动去掉。
坑五:元数据里的路径用了绝对路径。换台机器或者迁移知识库时,绝对路径全失效。统一用相对路径,或者只存文件名。
6. 一些实操心得和后续扩展方向
关于 chunk_size 的选择,我再补充一个实测经验:不要迷信固定值,要按文档类型调。技术文档、API 文档这种信息密度高的,chunk_size 可以小一点(300 字),因为每句话都可能是独立知识点;小说、散文这种叙事性的,chunk_size 要大一点(500~600 字),因为上下文依赖强,切太碎反而丢语义。
还有一个容易被忽略的点:解析和切分是两个独立阶段,不要耦合在一起。我见过有人把解析和切分写在一个函数里,结果想换切分策略时得重写整个解析逻辑。正确的做法是解析器只负责“文件 → 纯文本 + 结构信息”,切分器只负责“纯文本 + 结构信息 → chunks”,两者通过中间数据结构解耦。这样你换切分器、换 chunk_size,都不用动解析代码。
这个系列后面还会讲 PDF、Word、Excel、HTML 的解析,那些格式的坑比 txt 和 Markdown 多得多——PDF 的版面分析、Word 的样式继承、Excel 的多 sheet 处理,每一个都能单独写一篇。但不管处理什么格式,最终都要归到这篇讲的这套逻辑上:解析出文本和结构,按语义切分,带上元数据,清洗去重,抽检入库。把 txt 和 Markdown 这两个基础格式跑通,后面的格式只是解析器不同,管道骨架是一样的。
最后分享一个我一直在用的小技巧:给每个 chunk 的文本前面拼上它的 heading_path。比如一个 chunk 原文是“chunk_size 建议设 300~500 字”,拼上路径后变成“RAG 数据导入 > 切分策略 > chunk_size 建议设 300~500 字”。这样即使 chunk 本身没提“切分”,检索“切分策略”时也能命中它。实测下来,这个小改动能让召回率提升 10% 以上,成本几乎为零。