1. RAG 数据导入与解析的整体设计思路1.1 为什么数据导入是 RAG 系统的隐形瓶颈很多人做 RAG 项目注意力全放在向量库选型、Embedding 模型对比、检索策略调优上结果上线之后发现回答质量一塌糊涂。排查半天问题根本不在检索环节而是喂进去的数据本身就是一锅粥。我见过太多团队花两周调 prompt最后发现原始文档里全是乱码、页眉页脚、断行错位再好的模型也救不回来。RAG 的本质是“先找资料再回答”资料的质量直接决定回答的上限。数据导入与解析这一步决定了你的知识库里到底存了什么。存进去的是干净的段落检索就能命中要害存进去的是碎片化的字符串检索出来的东西连人都读不懂模型自然只能胡编。这个系列我打算从最基础的纯文本和 Markdown 入手把数据导入这件事拆透。为什么从这两种格式开始因为它们看起来最简单实际上最能暴露问题。txt 没有结构你得自己造结构Markdown 有结构但结构不规范时比 txt 还难处理。把这两种吃透了后面处理 PDF、Word、HTML 就是套框架的事。1.2 通用文本与结构化解析的分界线在哪我习惯把数据导入分成两条线一条是通用文本线处理那些没有明确层级、以自然语言为主的原始内容比如小说 txt、会议记录、聊天日志另一条是结构化解析线处理那些自带层级标记、需要保留语义关系的内容比如 Markdown 文档、带标题的说明手册。这两条线的核心差异在于通用文本线关注的是“怎么切分才不破坏语义”结构化解析线关注的是“怎么把结构信息提取出来并映射到元数据”。前者是切分问题后者是映射问题。举个具体例子。一段小说 txt你要做的是找到合适的切分点让每个 chunk 是一个完整的场景或对话而不是切在句子中间。而一份 Markdown 文档你要做的是把##标题提取出来作为 section 元数据把代码块单独标记把表格转成结构化数据。两者的处理逻辑完全不同但最终都要落到同一个目标上让每个 chunk 携带足够的上下文信息同时保持语义完整。1.3 解析管道的分层架构我在实际项目中会把解析管道拆成四层每层职责单一方便排查问题层级职责典型操作读取层把文件读进内存处理编码编码检测、BOM 处理、换行符统一清洗层去掉噪声保留有效内容去页眉页脚、去空行、去控制字符结构层识别并提取结构信息标题识别、列表识别、代码块识别切分层按语义切分并附加元数据递归切分、重叠窗口、元数据注入这四层分开的好处是当最终检索效果不好时你可以逐层排查是读取时编码就错了还是清洗时把有用内容删了还是切分粒度不对如果全揉在一起写出了问题只能靠猜。提示不要跳过清洗层直接做切分。我踩过的坑是原始 txt 里混了大量全角空格和零宽字符切分时看起来正常存进向量库后检索命中率极低排查了两天才定位到是零宽字符在作怪。2. 纯文本 txt 的读取与清洗实操2.1 编码检测别假设你的文件都是 UTF-8处理 txt 第一个坑就是编码。中文互联网上的 txt 文件来源五花八门GBK、GB2312、UTF-8、UTF-8 with BOM 都有甚至还有 Big5。如果你直接用open(file, r)默认编码读遇到非 UTF-8 文件直接抛异常或者读出乱码。我的做法是用chardet做编码探测然后显式指定编码读取。但 chardet 不是万能的短文本检测准确率会下降。所以我会加一个兜底策略先尝试 UTF-8失败则用 chardet 检测再失败就用 GBK 硬解最后用errorsreplace保证不中断流程。import chardet def read_text_file(filepath): with open(filepath, rb) as f: raw f.read() # 先尝试 UTF-8 try: return raw.decode(utf-8) except UnicodeDecodeError: pass # chardet 检测 detected chardet.detect(raw) encoding detected.get(encoding, gbk) confidence detected.get(confidence, 0) # 置信度太低时用 GBK 兜底 if confidence 0.7: encoding gbk return raw.decode(encoding, errorsreplace)这里有个细节errorsreplace会把无法解码的字符替换成虽然不完美但至少保证流程不中断。后续在清洗层可以统一把这些替换字符去掉。2.2 换行符与空白字符的统一处理Windows 的\r\n、Linux 的\n、老 Mac 的\r这三种换行符混在一起是常态。更麻烦的是全角空格\u3000、零宽空格\u200b、不间断空格\xa0这些字符肉眼看不出来但会严重影响后续的切分和检索。我的清洗流程是这样的import re def normalize_whitespace(text): # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 去掉零宽字符 text re.sub(r[\u200b\u200c\u200d\ufeff], , text) # 全角空格转半角 text text.replace(\u3000, ) # 不间断空格转普通空格 text text.replace(\xa0, ) # 连续空行压缩为一个 text re.sub(r\n{3,}, \n\n, text) # 行首行尾空白去掉 lines [line.strip() for line in text.split(\n)] return \n.join(lines)这段代码看起来简单但每一步都有原因。零宽字符是很多网页复制粘贴带进来的全角空格在中文文档里特别常见连续空行会让切分时产生大量空 chunk。这些处理不做后面切分出来的东西质量会差很多。2.3 噪声过滤页眉页脚与无关内容txt 文件里常见的噪声包括重复的页眉页脚、页码、网址、广告文本、扫描件的 OCR 残留。这些东西如果不清掉会被当成正文存进知识库检索时可能被命中污染回答。识别页眉页脚的思路是统计每行文本在文档中出现的频率如果某行反复出现且位置固定比如总是在每 N 行出现一次大概率是页眉页脚。具体实现可以用行频统计加位置分析。from collections import Counter def remove_repeated_lines(text, threshold3): lines text.split(\n) line_counts Counter(lines) # 找出重复出现且长度较短的行 repeated { line for line, count in line_counts.items() if count threshold and len(line) 50 and line.strip() } filtered [line for line in lines if line not in repeated] return \n.join(filtered)这个方法的局限是如果正文里某句话恰好重复了多次也会被误删。所以阈值要调一般设 3 到 5 比较稳妥。对于更复杂的场景可以结合位置信息比如只删除出现在每页固定位置的行。注意清洗力度要克制。我见过有人为了“干净”把所有的短行都删了结果对话体小说里的人物对话全没了。清洗的目标是去噪声不是去内容每加一条规则都要拿实际数据验证。3. Markdown 结构化解析的核心要点3.1 Markdown 的层级提取与元数据映射Markdown 相比 txt 最大的优势是自带结构标记。#到######表示标题层级-和1.表示列表表示代码块|表示表格。这些标记在解析时不能简单删掉而应该提取出来作为 chunk 的元数据。我的做法是解析出每个内容块的“标题路径”。比如一个 chunk 位于## 第二章下面的### 2.1 节里那这个 chunk 的元数据就包含section_path: 第二章 2.1 节。检索时这个路径信息可以帮助模型理解 chunk 的上下文位置。import re def parse_markdown_structure(text): lines text.split(\n) structure [] current_headers {} # level - header_text for line in lines: header_match re.match(r^(#{1,6})\s(.)$, line) if header_match: level len(header_match.group(1)) title header_match.group(2).strip() current_headers[level] title # 清除更低层级的标题 for k in list(current_headers.keys()): if k level: del current_headers[k] structure.append({ type: header, level: level, title: title, path: .join( current_headers[k] for k in sorted(current_headers.keys()) ) }) else: structure.append({ type: content, text: line, path: .join( current_headers[k] for k in sorted(current_headers.keys()) ) if current_headers else }) return structure这段代码的核心逻辑是维护一个current_headers字典遇到新标题时更新对应层级并清除更深的层级。这样每个内容块都能拿到自己所属的完整标题路径。3.2 代码块与表格的特殊处理Markdown 里的代码块和表格是两类特殊内容不能按普通文本切分。代码块被切碎后完全失去意义表格被切碎后行列关系就乱了。代码块的处理策略是识别包裹的区域整块保留不参与常规切分。如果代码块超过 chunk 大小限制按函数或逻辑块切分而不是按行数硬切。def extract_code_blocks(text): pattern r(\w*)\n(.*?) blocks [] for match in re.finditer(pattern, text, re.DOTALL): lang match.group(1) or unknown code match.group(2) blocks.append({ type: code, language: lang, content: code, start: match.start(), end: match.end() }) return blocks表格的处理更复杂一些。Markdown 表格用|分隔列用---分隔表头和数据。解析时可以转成二维数组存进元数据里。检索时如果命中表格 chunk可以把整个表格作为上下文返回而不是只返回某一行。处理对象切分策略元数据普通段落按语义边界切分标题路径代码块整块保留超长按逻辑切语言类型表格整表保留列名、行数列表按列表项聚合列表类型3.3 数学公式与特殊语法的兼容技术类 Markdown 文档里经常有 LaTeX 数学公式行内公式用$...$块级公式用$$...$$。这些公式如果被切分破坏检索时完全无法匹配。处理方式和代码块类似识别出来整块保留标记类型。另外还有 callout 语法 [!NOTE]、脚注、引用块等。这些特殊语法在解析时都要单独识别不能当普通文本处理。我的原则是任何有语义边界的结构都要在切分时保护起来。def protect_special_blocks(text): 把特殊块替换成占位符切分后再还原 placeholders {} counter [0] def replace_block(match, block_type): key f__BLOCK_{counter[0]}__ counter[0] 1 placeholders[key] { type: block_type, content: match.group(0) } return key # 保护代码块 text re.sub(r[\s\S]*?, lambda m: replace_block(m, code), text) # 保护块级公式 text re.sub(r\$\$[\s\S]*?\$\$, lambda m: replace_block(m, math), text) return text, placeholders这种“占位符保护”的思路在处理混合内容时特别有用。先把特殊块替换成不会被打断的短字符串切分完再把原始内容还原回去。4. 语义切分的参数计算与实操4.1 chunk size 与 overlap 的取舍逻辑切分参数是 RAG 里最容易被拍脑袋决定的东西。很多人直接抄一个chunk_size500, overlap50就用结果效果不好也不知道为什么。chunk size 的选择取决于三个因素Embedding 模型的最大输入长度、内容的语义密度、检索的粒度需求。Embedding 模型一般支持 512 到 8192 个 token但并不是越大越好。chunk 太大检索时噪声多一个 chunk 里混了好几个主题chunk 太小语义不完整模型拿到的上下文不够。我的经验值是中文内容按字符数算chunk_size 在 300 到 800 之间。技术文档可以偏大因为一段完整的技术说明通常需要这么多字对话体或问答类内容偏小因为每个问答对本身就是独立语义单元。overlap 的作用是防止语义在边界处被切断。如果两个相邻 chunk 完全没有重叠切分点正好落在一个句子的中间那这个句子的前半段在 chunk A后半段在 chunk B两边都不完整。overlap 一般设为 chunk_size 的 10% 到 20%。def calculate_overlap(chunk_size, content_type): 根据内容类型计算合适的 overlap ratios { technical: 0.15, # 技术文档上下文依赖强 narrative: 0.10, # 叙述类边界相对清晰 qa: 0.05, # 问答类几乎不需要重叠 code: 0.20, # 代码上下文依赖最强 } ratio ratios.get(content_type, 0.10) return int(chunk_size * ratio)4.2 递归切分的实现与边界处理递归切分Recursive Character Text Splitter是目前最通用的切分策略。它的思路是先按最粗的边界如段落切如果某段还是太大再按次级边界如句子切依次递归直到所有 chunk 都小于目标大小。分隔符的优先级顺序很关键。我一般用这个顺序\n\n段落→\n行→。中文句号→→→→ 空格 → 字符。def recursive_split(text, chunk_size, overlap, separatorsNone): if separators is None: separators [\n\n, \n, 。, , , , , , ] if len(text) chunk_size: return [text] # 找到第一个能切分的分隔符 for sep in separators: if sep : # 最后兜底按字符硬切 chunks [] for i in range(0, len(text), chunk_size - overlap): chunks.append(text[i:i chunk_size]) return chunks if sep in text: parts text.split(sep) chunks [] current for part in parts: candidate current sep part if current else part if len(candidate) chunk_size: current candidate else: if current: chunks.append(current) # 如果单个 part 就超长递归处理 if len(part) chunk_size: sub_chunks recursive_split( part, chunk_size, overlap, separators[separators.index(sep)1:] ) chunks.extend(sub_chunks[:-1]) current sub_chunks[-1] if sub_chunks else else: current part if current: chunks.append(current) # 添加 overlap return add_overlap(chunks, overlap) return [text]这段代码的核心是“尽量在高级边界切分”。段落边界优于句子边界句子边界优于字符边界。这样切出来的 chunk 语义完整性最好。4.3 元数据注入让每个 chunk 自带上下文切分完之后每个 chunk 不能是光秃秃的文本必须带上元数据。元数据的作用是在检索时提供额外信息帮助排序和过滤。我一般会注入这些字段字段名说明示例source来源文件名产品手册.mdsection_path标题路径第三章 3.2 安装步骤chunk_index在文档中的序号15content_type内容类型paragraph / code / tablechar_count字符数420这些元数据在存入向量库时一起存进去。检索时可以按section_path过滤也可以按content_type筛选。比如用户问的是代码相关的问题就可以优先检索content_typecode的 chunk。def build_chunk_with_metadata(chunks, source, structure_info): result [] for i, chunk in enumerate(chunks): # 找到这个 chunk 对应的标题路径 path find_section_path(chunk, structure_info) result.append({ text: chunk, metadata: { source: source, section_path: path, chunk_index: i, char_count: len(chunk), content_type: detect_content_type(chunk) } }) return result实操心得元数据不要贪多。我一开始把能想到的字段全塞进去结果向量库的存储成本翻倍检索速度也受影响。后来精简到 5 个核心字段效果没差别成本降了不少。5. 常见问题排查与避坑指南5.1 编码乱码问题的快速定位乱码是 txt 处理最高频的问题。表现是读出来的文本里出现文档这种字符或者中文变成问号。定位思路是先确认原始文件的真实编码再确认读取时用的编码。快速判断方法用十六进制编辑器打开文件看开头几个字节。UTF-8 BOM 是EF BB BFUTF-16 LE 是FF FEUTF-16 BE 是FE FF。没有 BOM 的话看中文字符的字节模式GBK 的中文是两个字节且高位为 1UTF-8 的中文是三个字节。def diagnose_encoding(filepath): with open(filepath, rb) as f: raw f.read(1000) # 检查 BOM if raw.startswith(b\xef\xbb\xbf): return utf-8-sig if raw.startswith(b\xff\xfe): return utf-16-le if raw.startswith(b\xfe\xff): return utf-16-be # 尝试各种编码 for enc in [utf-8, gbk, gb2312, big5]: try: raw.decode(enc) return enc except UnicodeDecodeError: continue return unknown5.2 切分粒度过大或过小的症状与调整切分粒度不对症状很明显。粒度过大时检索出来的 chunk 里混了好几个主题模型回答时容易跑偏或者只用了其中一部分信息。粒度过小时chunk 语义不完整检索命中后模型拿到的上下文不够回答不完整。调整方法是拿一批典型问题做测试看检索出来的 top-3 chunk 是否包含回答所需的信息。如果 chunk 里信息太多太杂就调小 chunk_size如果 chunk 里信息残缺就调大 chunk_size 或增加 overlap。我一般会做一个简单的评估人工标注 20 个问题每个问题标记哪些 chunk 是相关的然后计算检索的召回率和精确率。chunk_size 从 300 到 800 扫一遍看哪个值效果最好。5.3 特殊字符导致检索失效的排查有些字符在文本里看不见但会影响 Embedding 的质量。除了前面提到的零宽字符还有这些软连字符\u00ad从网页复制时常见方向标记\u200e\u200f从某些编辑器复制时带入私用区字符\ue000-\uf8ff某些字体或图标字体产生排查方法是把文本转成 Unicode 码点序列看有没有异常码点。清洗时统一过滤掉这些字符。def remove_invisible_chars(text): # 保留常规字符去掉控制字符和格式字符 return .join( ch for ch in text if unicodedata.category(ch)[0] ! C or ch in \n\t )5.4 常见问题速查表问题现象可能原因排查方法解决方案读出乱码编码不匹配十六进制查看文件头用 chardet 检测或手动指定编码检索命中率低零宽字符干扰打印 Unicode 码点过滤不可见字符chunk 语义不完整切分粒度过小检查 chunk 长度分布增大 chunk_size 或 overlapchunk 主题混杂切分粒度过大人工检查 chunk 内容减小 chunk_size优化分隔符代码块被切碎未保护特殊块检查代码块是否完整用占位符保护后再切分表格行列错乱表格被当普通文本检查表格 chunk整表保留转结构化存储避坑提醒不要等到全部数据导入完才做验证。我习惯每导入 100 条就抽检 5 条看 chunk 质量、元数据是否正确。早期发现问题调整成本低等几万条都导完了再发现切分策略有问题重跑的时间成本很高。6. 从解析到入库的完整链路串联6.1 一个可复用的解析管道实现把前面所有环节串起来形成一个完整的解析管道。输入是文件路径输出是带元数据的 chunk 列表可以直接送入 Embedding 和向量库。class DocumentParser: def __init__(self, chunk_size500, overlapNone): self.chunk_size chunk_size self.overlap overlap or int(chunk_size * 0.1) def parse(self, filepath): # 1. 读取 text read_text_file(filepath) # 2. 清洗 text normalize_whitespace(text) text remove_invisible_chars(text) text remove_repeated_lines(text) # 3. 结构解析 if filepath.endswith(.md): structure parse_markdown_structure(text) text, placeholders protect_special_blocks(text) else: structure [] placeholders {} # 4. 切分 chunks recursive_split(text, self.chunk_size, self.overlap) # 5. 还原特殊块 chunks restore_placeholders(chunks, placeholders) # 6. 注入元数据 result build_chunk_with_metadata( chunks, filepath, structure ) return result这个管道的好处是每一层都可以单独替换。比如你想换一种切分算法只需要改recursive_split这一处其他层不受影响。6.2 批量导入时的性能与稳定性考量单文件解析没问题但批量导入几千个文件时性能和稳定性就是另一回事了。我遇到过的问题包括内存溢出、某个文件解析卡死导致整个批次中断、编码检测耗时过长。解决方案是分批处理每批 50 到 100 个文件加超时机制单个文件解析超过 30 秒就跳过并记录编码检测结果缓存相同来源的文件编码通常一致。import signal from concurrent.futures import ProcessPoolExecutor def parse_with_timeout(filepath, timeout30): def handler(signum, frame): raise TimeoutError(f解析超时: {filepath}) signal.signal(signal.SIGALRM, handler) signal.alarm(timeout) try: parser DocumentParser() return parser.parse(filepath) except TimeoutError as e: print(f跳过: {e}) return [] finally: signal.alarm(0)用多进程而不是多线程因为文本解析是 CPU 密集型任务Python 的 GIL 会限制多线程效果。进程池的大小设为 CPU 核心数即可不要开太多否则上下文切换开销反而拖慢速度。6.3 解析质量的自检清单导入完成后做一轮自检。我一般检查这几项随机抽 10 个 chunk人工阅读看语义是否完整、有没有乱码统计 chunk 长度分布看有没有异常短小于 50 字符或异常长超过 2 倍 chunk_size的检查元数据字段是否都有值特别是section_path有没有为空的情况用几个典型问题做检索测试看 top-3 结果是否相关这套自检做完基本能保证导入的数据质量。如果发现问题回到对应的层去调整重新跑一遍管道即可。个人体会数据导入这件事投入的时间永远值得。我做过对比同样一套 RAG 系统数据解析做得好和做得差回答准确率能差 30% 以上。与其在检索策略上反复调参不如先把数据质量这一关把住。后面几篇我会继续讲 PDF、Word、HTML 这些格式的解析思路和这篇是一脉相承的只是多了格式特有的处理环节。