RAG数据导入解析：txt与Markdown的编码检测、段落还原与语义分块实战

RAG 系统落地时，很多人把八成精力砸在向量库选型和检索算法调优上，结果上线后回答质量一塌糊涂。排查半天才发现，问题根本不在检索端，而是最上游的数据导入环节就烂了——PDF 里的表格被拍成一坨乱码，Markdown 的标题层级全丢，txt 文件连段落边界都分不清。我前后经手过七八个 RAG 项目，踩得最深的坑几乎都集中在数据导入与解析这一段。这篇就从最基础的 txt 和 Markdown 两种格式切入，把通用文本与结构化解析这件事讲透。

RAG 数据导入与解析的核心目标只有一个：把各种格式的原始文档，转成检索端能高效利用的、语义完整的文本块。txt 和 Markdown 看起来简单，但恰恰因为简单，很多人直接read()一把梭，结果 chunk 切得稀碎，元数据全丢，后面检索时连"这段话出自哪份文档的哪个章节"都答不上来。这篇文章适合正在搭建 RAG 知识库的工程师、需要批量处理文档的数据从业者，以及想搞清楚"为什么我的 RAG 效果差"的开发者。读完你至少能掌握：txt 编码与段落还原的完整方案、Markdown 结构解析与语义分块的具体做法、以及一套可直接复用的解析流水线设计思路。

1. 为什么 txt 和 Markdown 的解析反而最容易翻车

1.1 简单格式的"隐性复杂度"

大多数人觉得 txt 就是纯文本，读进来就能用。但真实场景里的 txt 文件来源极其杂乱：从 Windows 记事本导出的 GBK 编码、从网页复制粘贴带一堆不可见字符、从 PDF 转换工具生成的硬换行、从数据库导出的定宽字段。这些文件用 UTF-8 直接读，轻则乱码，重则整个文件读取失败。

Markdown 的问题更隐蔽。它表面上是纯文本，实际上承载了完整的结构信息——标题层级、列表嵌套、代码块、表格、引用块。如果你把它当普通 txt 处理，这些结构信息全部丢失，chunk 切分时就会把"某个二级标题下的三段论述"硬生生切成互不相关的碎片。检索时用户问一个需要跨段综合的问题，系统只能召回其中一段，答案自然残缺。

我见过一个典型案例：某团队的知识库有 3000 多份 Markdown 技术文档，他们用固定 500 字符切分，结果一份文档的"配置说明"章节被切成了 7 个 chunk，其中 3 个 chunk 的开头是半句话。用户提问"如何配置 XX 参数"，检索召回的 chunk 里参数名和参数值分属不同块，模型只能瞎猜。

1.2 解析质量如何决定 RAG 的上限

RAG 的链路是"解析 → 分块 → 向量化 → 检索 → 生成"。解析是第一步，也是唯一一个"信息只减不增"的环节。后面所有环节再优秀，也无法恢复解析阶段丢失的信息。这就是所谓的"垃圾进，垃圾出"。

具体来说，解析质量从三个维度影响 RAG 效果：

• 语义完整性：chunk 是否包含完整的语义单元。一个被拦腰截断的段落，向量化后语义是扭曲的，检索时既召不回也排不准。
• 结构可追溯性：每个 chunk 是否携带来源元数据（文件名、标题路径、章节层级）。这决定了生成答案时能否给出准确引用，也决定了能否做基于结构的过滤检索。
• 噪声控制：页眉页脚、乱码字符、重复空行这些噪声如果不清除，会稀释向量语义，让相似度计算失准。

提示：在优化检索算法之前，先花时间审计你的解析产物。把 chunk 随机抽 20 个打印出来看，如果有一半读起来语义不完整，那检索端怎么调都是白费功夫。

1.3 通用文本与结构化文本的分野

从解析策略上，文档可以粗分为两类。一类是通用文本，以 txt 为代表，结构信息弱，主要靠段落、换行、标点来推断语义边界。另一类是结构化文本，以 Markdown 为代表，显式携带层级和语义标记，解析时要充分利用这些标记。

这两类的处理思路完全不同。txt 的重点是"还原"——从混乱的原始字节里还原出段落和章节。Markdown 的重点是"提取"——把标记语言里的结构信息抽取成机器可用的元数据。把这两类混为一谈，是很多解析流水线设计失败的根源。下面两章分别拆解。

2. txt 文件解析：编码识别与段落还原的完整链路

2.1 编码检测：别让第一步就崩掉

txt 文件没有自描述编码信息，这是它最大的坑。同一个文件，用 UTF-8 读可能是乱码，用 GBK 读可能正常，反过来也一样。硬编码一种编码去读，迟早出事。

我的做法是引入编码检测库做自动识别，Python 生态里charset-normalizer是目前维护最活跃、准确率最高的选择（chardet已经多年不更新，不建议再用）。核心逻辑是：先读文件的前若干 KB 字节，让检测器给出候选编码及置信度，再按置信度从高到低尝试解码，第一个成功且不含替换字符的编码即为最终编码。

from charset_normalizer import from_path def detect_encoding(file_path, sample_size=100_000): with open(file_path, 'rb') as f: raw = f.read(sample_size) results = from_bytes(raw) best = results.best() if best is None: return 'utf-8' # 兜底 return best.encoding def read_text_safely(file_path): encoding = detect_encoding(file_path) try: with open(file_path, 'r', encoding=encoding, errors='strict') as f: return f.read(), encoding except UnicodeDecodeError: # 严格模式失败，退化为替换模式，保证不中断流水线 with open(file_path, 'r', encoding=encoding, errors='replace') as f: return f.read(), encoding

这里有个经验点：采样大小很关键。如果只读前 1KB，遇到文件开头是英文、后面全是中文的情况，检测器会误判成 ASCII 或 Latin-1。我一般采样 100KB 起步，对于超大文件可以分段采样后投票。另外，检测结果要记录到元数据里，方便后续排查。

注意：errors='replace'是保命手段，不是常规手段。它会把无法解码的字节替换成 U+FFFD 字符，虽然不中断流程，但会引入噪声。如果某个文件的替换字符比例超过 1%，我会把它单独标记出来人工检查，而不是直接放进知识库。

2.2 不可见字符与换行符的清洗

编码问题解决后，下一个坑是看不见的字符。从网页复制的文本常带零宽空格（U+200B）、零宽连字符（U+200D）、不换行空格（U+00A0）；从 Windows 来的文件用\r\n换行，从老 Mac 来的用\r。这些字符如果不清洗，会污染向量语义，还会让后续的分块逻辑判断失误。

清洗要分层次做。第一层是统一换行符，把所有\r\n和\r归一成\n。第二层是替换特殊空白字符，把不换行空格、各种宽度的空格统一成普通空格。第三层是移除零宽字符。第四层是压缩连续空行，但要注意保留段落边界——我一般把 3 个以上连续换行压缩成 2 个，这样既去掉了冗余，又保留了段落分隔。

import re import unicodedata def clean_text(text): # 统一换行符 text = text.replace('\r\n', '\n').replace('\r', '\n') # 归一化 Unicode，把各种兼容字符转成标准形式 text = unicodedata.normalize('NFKC', text) # 移除零宽字符 text = re.sub(r'[\u200b\u200c\u200d\ufeff]', '', text) # 替换特殊空白为普通空格 text = re.sub(r'[\u00a0\u2000-\u200a\u202f\u205f\u3000]', ' ', text) # 压缩连续空行，保留段落边界 text = re.sub(r'\n{3,}', '\n\n', text) # 去除行尾空白 text = '\n'.join(line.rstrip() for line in text.split('\n')) return text

unicodedata.normalize('NFKC', text)这一步很多人会忽略，但它很重要。NFKC 会把全角字符转半角、把各种花式引号转成标准引号、把连字符合并。这对后续的关键词匹配和向量化都有好处，因为"（"和"("在语义上是一回事，但字符层面不同，不归一化会造成检索时的匹配遗漏。

2.3 段落边界推断：从硬换行到语义段落

清洗完的 txt 还是一个大字符串，怎么切成语义段落？这是 txt 解析最考验经验的地方。核心难点在于区分"硬换行"和"段落换行"。硬换行是排版需要（比如每行 80 字符就换行），段落换行才是语义边界。

判断依据有几条，我按可靠性排序：

1. 空行分隔：两个换行符之间夹一个空行，几乎可以确定是段落边界。这是最可靠的信号。
2. 行尾标点：如果一行以句号、问号、感叹号、分号结尾，下一行大概率是新段落或新句子。
3. 行首缩进：中文段落常有首行缩进（两个全角空格），这是强段落信号。
4. 行长度分布：如果文件里大部分行长度接近某个固定值（比如都是 78-82 字符），说明是硬换行排版，应该把连续短行合并。

我的策略是组合判断：先按空行切分成大块，对每个大块内部，如果检测到"行长度高度一致"的特征，就把块内的单换行合并成空格（还原被硬换行打断的段落）；否则保留单换行作为软边界。这样能同时处理"一段一行"和"固定宽度换行"两种常见排版。

def infer_paragraphs(text): blocks = re.split(r'\n\s*\n', text) paragraphs = [] for block in blocks: lines = [l for l in block.split('\n') if l.strip()] if not lines: continue lengths = [len(l) for l in lines] # 判断是否为固定宽度硬换行 if len(lines) > 3 and (max(lengths) - min(lengths)) < 8: # 硬换行，合并成一段 merged = ' '.join(l.strip() for l in lines) paragraphs.append(merged) else: # 保留原有换行结构，逐行作为独立段落候选 for line in lines: paragraphs.append(line.strip()) return paragraphs

这个阈值（行数大于 3、长度差小于 8）是我在大量中英文文档上调出来的经验值。中文文档因为字符宽度一致，硬换行的长度差通常更小；英文文档因为单词长度不一，长度差会大一些，可以适当放宽到 12。你可以根据自己语料的实际情况微调。

2.4 章节标题识别：给纯文本装上骨架

txt 没有显式标题标记，但很多文档其实有隐性的章节结构，比如"第一章""1. 概述""【配置说明】"这类行。识别出这些标题行，能给后续分块提供天然的边界，还能构建出章节路径元数据。

识别规则我总结了几条正则模式：以"第X章/节/部分"开头的、以数字编号开头的（如"1."、"1.1"、"一、"）、被方括号或书名号包裹的短行、全大写或加粗标记的短行。关键约束是标题行通常较短（一般不超过 40 字）且独立成行。

TITLE_PATTERNS = [ r'^第[一二三四五六七八九十百零\d]+[章节部分篇]', r'^[\d]+(\.[\d]+)*[\s、.．]', r'^[一二三四五六七八九十]+[、.．]', r'^[【\[（(].{1,30}[】\]）)]$', ] def is_title_line(line): stripped = line.strip() if not stripped or len(stripped) > 40: return False for pattern in TITLE_PATTERNS: if re.match(pattern, stripped): return True return False

识别出标题后，我会给每个段落打上"所属章节路径"的标签。比如某段落在"第三章"下的"3.2 配置"小节里，它的元数据就是{"chapter": "第三章", "section": "3.2 配置"}。这个路径在检索时非常有用——用户问配置相关的问题，可以优先召回配置章节的 chunk。

提示：标题识别一定会有误判，比如正文里出现"1. 首先做某事"这种列表项会被误判成标题。我的处理是加一层校验：如果被判定为标题的行后面紧跟的内容也是类似的编号行，那大概率是列表而非标题，降级处理。宁可漏判，不可错判，因为错判会切断本应连续的段落。

3. Markdown 解析：把标记语言的结构信息榨干

3.1 为什么不能用正则硬解 Markdown

拿到 Markdown 文件，很多人的第一反应是写正则匹配#开头的行来提取标题。这个思路在小规模、格式规范的文档上能跑通，但一旦遇到嵌套列表、代码块里的#注释、表格里的竖线，正则就会崩。

举个典型反例：代码块里有一行 Python 注释# 这是注释，正则会把当成一级标题。再比如引用块里的> ## 小标题，正则也会误判。Markdown 的语法是有上下文依赖的，必须用真正的解析器。

Python 生态里markdown-it-py是 CommonMark 规范兼容性最好的解析器，它能把 Markdown 解析成 token 流，每个 token 带有类型、层级、内容等完整信息。基于 token 流做结构提取，比正则可靠一个数量级。

from markdown_it import MarkdownIt md = MarkdownIt("commonmark") tokens = md.parse(markdown_text)

解析出来的 token 流是扁平的，但每个heading_open、heading_close、paragraph_open等 token 带有level和tag信息，可以据此重建树形结构。

3.2 标题层级重建与章节路径生成

Markdown 的标题层级（H1 到 H6）天然构成了文档的树形目录。解析时要做的是遍历 token 流，维护一个"当前标题栈"，遇到新标题时根据层级弹出或压入栈，从而为每个内容块生成完整的章节路径。

def extract_structure(tokens): heading_stack = [] # [(level, text), ...] blocks = [] current_content = [] i = 0 while i < len(tokens): tok = tokens[i] if tok.type == 'heading_open': # 先结算上一个内容块 if current_content: blocks.append({ 'path': [h[1] for h in heading_stack], 'content': '\n'.join(current_content).strip() }) current_content = [] level = int(tok.tag[1]) # 弹出层级 >= 当前标题的栈元素 while heading_stack and heading_stack[-1][0] >= level: heading_stack.pop() # 下一个 token 是标题文本 title_text = tokens[i+1].content heading_stack.append((level, title_text)) i += 3 # 跳过 heading_open, inline, heading_close continue elif tok.type == 'inline': current_content.append(tok.content) i += 1 if current_content: blocks.append({ 'path': [h[1] for h in heading_stack], 'content': '\n'.join(current_content).strip() }) return blocks

这段代码的核心是heading_stack的维护逻辑。当遇到一个 H2 标题时，栈里所有层级大于等于 2 的标题都要弹出，然后把新 H2 压入。这样栈里始终保存着"从根到当前节点"的完整路径。每个内容块结算时，路径就是它所属的章节链路。

这个路径的价值在于：它既是元数据（可以存进向量库做过滤），也是分块的天然边界（同一路径下的内容语义相关，适合放在一起）。我实测下来，基于标题路径分块比固定长度分块，检索准确率能提升 20% 以上。

3.3 代码块、表格、列表的特殊处理

Markdown 里的代码块、表格、列表是三类特殊结构，处理方式要区别对待。

代码块必须整体保留，绝不能从中间切断。一个被截断的代码块，语义完全失效。解析时识别fencetoken，把整个代码块作为一个不可分割的单元。同时给代码块打上语言标签（从 fence 的 info 字段取），检索时可以做语言过滤。

表格要转成结构化表示。Markdown 表格本质是二维数据，直接当文本处理会丢失行列关系。我的做法是把表格转成"表头: 值"的键值对形式，或者保留为 Markdown 原格式但整体作为一个 chunk。前者适合字段明确的配置表，后者适合内容型表格。

列表要保留层级关系。嵌套列表如果拍平，会丢失从属关系。解析时根据bullet_list_open、list_item_open等 token 的嵌套深度，给每个列表项加上缩进标记，保留层级。

def handle_special_blocks(tokens): """识别代码块、表格等特殊块，标记为不可分割""" special_blocks = [] i = 0 while i < len(tokens): tok = tokens[i] if tok.type == 'fence': special_blocks.append({ 'type': 'code', 'lang': tok.info.strip() or 'text', 'content': tok.content, 'atomic': True # 不可分割标记 }) elif tok.type == 'table_open': # 收集整个表格 table_tokens = [] depth = 1 j = i + 1 while j < len(tokens) and depth > 0: if tokens[j].type == 'table_open': depth += 1 elif tokens[j].type == 'table_close': depth -= 1 table_tokens.append(tokens[j]) j += 1 special_blocks.append({ 'type': 'table', 'tokens': table_tokens, 'atomic': True }) i = j continue i += 1 return special_blocks

atomic这个标记很关键。它告诉后续的分块器：这个块不能被切开，如果它超过了 chunk 大小上限，要么整体保留（允许超限），要么单独成块。代码块和表格我一般允许超限，因为切开的代价远大于超限的代价。

3.4 从 Markdown 到语义 chunk 的映射策略

有了结构化的块和章节路径，最后一步是把它们映射成适合检索的 chunk。这里有几个策略选择，我按推荐度排序。

策略一：标题路径 + 内容合并。把同一标题路径下的所有段落合并成一个 chunk，如果超过大小上限，再按段落边界二次切分。这个策略的优点是 chunk 语义完整，缺点是大小不均。适合章节粒度适中的文档。

策略二：固定大小 + 标题路径元数据。按固定字符数切分，但每个 chunk 都带上所属标题路径。优点是大小均匀，缺点是可能切断语义。适合章节粒度差异大的文档。

策略三：语义分块。用句子嵌入模型计算相邻句子的相似度，在相似度骤降处切分。效果最好但成本最高，适合对质量要求极高的场景。

我一般用策略一为主、策略二兜底。具体做法是：先按标题路径聚合，如果聚合后的块小于最小阈值（比如 200 字符），就和相邻块合并；如果大于最大阈值（比如 1500 字符），就按段落边界切分，切分后的每个子块都继承父块的标题路径。

def build_chunks(blocks, min_size=200, max_size=1500): chunks = [] for block in blocks: content = block['content'] path = block['path'] if len(content) <= max_size: if len(content) < min_size and chunks: # 太小，合并到上一个 chunk chunks[-1]['content'] += '\n\n' + content else: chunks.append({ 'content': content, 'path': path, 'metadata': {'section': ' > '.join(path)} }) else: # 太大，按段落切分 paragraphs = content.split('\n\n') buffer = '' for para in paragraphs: if len(buffer) + len(para) > max_size and buffer: chunks.append({ 'content': buffer.strip(), 'path': path, 'metadata': {'section': ' > '.join(path)} }) buffer = para else: buffer += '\n\n' + para if buffer else para if buffer.strip(): chunks.append({ 'content': buffer.strip(), 'path': path, 'metadata': {'section': ' > '.join(path)} }) return chunks

min_size和max_size这两个阈值不是拍脑袋定的。max_size主要受嵌入模型的最大输入长度约束，同时要考虑检索粒度——块太大，召回的内容里噪声多；块太小，语义不完整。我实测下来，中文文档 500-1000 字符、英文文档 1000-2000 字符是比较舒服的区间。min_size则是为了避免产生大量无意义的碎片块。

4. 通用解析流水线的工程化设计

4.1 分层架构：解析器、清洗器、分块器的职责边界

把 txt 和 Markdown 的解析逻辑写成一堆散落的函数，短期能跑，长期必乱。我的做法是抽象成三层：解析器层负责把原始文件转成统一的中间表示，清洗器层负责规范化文本，分块器层负责切成语义 chunk。三层之间通过明确定义的数据结构通信。

中间表示我定义成一个Document对象，包含raw_text（原始文本）、blocks（结构化块列表）、metadata（文件级元数据）。每个 block 包含type（paragraph/heading/code/table）、content、path（章节路径）、atomic（是否不可分割）。

from dataclasses import dataclass, field from typing import List, Dict, Optional @dataclass class Block: type: str content: str path: List[str] = field(default_factory=list) atomic: bool = False extra: Dict = field(default_factory=dict) @dataclass class Document: source: str raw_text: str blocks: List[Block] = field(default_factory=list) metadata: Dict = field(default_factory=dict)

这个抽象的好处是，新增一种格式（比如 HTML、PDF）时，只需要写一个新的解析器把它转成Document，后面的清洗和分块逻辑完全复用。这就是所谓的"解析器可插拔"。

4.2 格式路由：根据扩展名和内容特征选择解析器

流水线的入口需要一个路由逻辑，判断该用哪个解析器。最直接的是看文件扩展名，但扩展名会骗人——.txt文件里可能是 Markdown，.md文件里可能是纯文本。所以要做内容特征检测作为补充。

我的路由策略是：先看扩展名给出候选解析器，再读文件头部做特征检测。如果头部有大量#开头的行、有代码块标记、有链接语法，就判定为 Markdown；如果全是纯文本无标记，就判定为 txt。两者冲突时以内容特征为准。

def route_parser(file_path, head_text): ext = file_path.lower().rsplit('.', 1)[-1] md_signals = 0 lines = head_text.split('\n')[:50] for line in lines: if re.match(r'^#{1,6}\s', line): md_signals += 2 if re.match(r'^```', line): md_signals += 2 if re.match(r'^[-*+]\s', line): md_signals += 1 if re.match(r'^\d+\.\s', line): md_signals += 1 if md_signals >= 3: return 'markdown' if ext in ('md', 'markdown'): return 'markdown' return 'plaintext'

这个打分机制是我踩过坑之后加的。之前有个项目，一批.txt文件其实是 Markdown 格式，用纯文本解析器处理后标题全丢了，检索效果很差。加了内容检测后，这类文件被正确识别。

4.3 元数据注入：让每个 chunk 都能自证来源

元数据是 RAG 系统里被严重低估的一环。很多人只存 chunk 文本和向量，检索时无法做过滤、无法给引用、无法追溯。我的原则是：每个 chunk 必须携带足够的元数据，让它脱离原始文件也能自证来源。

必存的元数据字段包括：source_file（文件名）、source_path（相对路径）、section（章节路径）、chunk_index（在文档内的序号）、char_count（字符数）、parse_time（解析时间）。可选字段包括：language（语言）、doc_type（文档类型）、tags（标签）。

def enrich_metadata(chunk, doc, index): chunk['metadata'].update({ 'source_file': doc.metadata.get('filename'), 'source_path': doc.metadata.get('rel_path'), 'chunk_index': index, 'char_count': len(chunk['content']), 'parse_time': doc.metadata.get('parse_time'), 'doc_type': doc.metadata.get('doc_type', 'unknown'), }) return chunk

这些元数据在检索时能派上大用场。比如用户问"XX 文档里怎么配置"，可以先用source_file过滤再检索；用户问"第三章讲了什么"，可以用section过滤。没有元数据，这些高级检索能力都无从谈起。

4.4 幂等性与增量更新：重复导入不产生脏数据

生产环境的 RAG 知识库不是一次性导入就完事的，文档会更新、会新增。如果每次全量重导，成本高且容易产生重复。所以流水线必须支持增量更新，且要保证幂等——同一份文档重复导入，结果应该一致，不产生重复 chunk。

我的做法是给每个 chunk 生成一个稳定的指纹（fingerprint），由source_path + section + content_hash组成。导入前先查指纹是否已存在，存在则跳过或更新，不存在则插入。文档更新时，先删除该source_path下的所有旧 chunk，再插入新的。

import hashlib def compute_fingerprint(chunk): key = f"{chunk['metadata']['source_path']}|{chunk['metadata']['section']}|{chunk['content']}" return hashlib.sha256(key.encode('utf-8')).hexdigest()[:16]

用content_hash而不是整个内容做指纹，是为了控制指纹长度。16 位十六进制（64 bit）的碰撞概率在实际规模下可以忽略。指纹要存进向量库的元数据字段，方便查询。

注意：增量更新时，删除旧 chunk 和插入新 chunk 要放在同一个事务里，否则中途失败会导致文档处于"半更新"状态。如果向量库不支持事务，就先用一个临时标记位标记旧 chunk，插入成功后再物理删除。

5. 实测中的坑与调优经验

5.1 中文文档的编码陷阱

中文 txt 文档的编码问题比英文严重得多。GBK、GB2312、GB18030、UTF-8、UTF-8 with BOM，五种编码都可能出现。其中 GB2312 是 GBK 的子集，GBK 又是 GB18030 的子集，检测器经常在它们之间摇摆。

我的经验是：优先信任 GB18030。因为它是超集，能解码 GBK 和 GB2312 的所有字符。当检测器给出 GBK 或 GB2312 时，我会尝试用 GB18030 解码，如果能成功且结果合理，就用 GB18030。这样能避免因编码范围不足导致的解码失败。

另一个坑是 UTF-8 BOM。带 BOM 的文件开头有三个字节EF BB BF，如果不去掉，第一个字符会变成不可见的 BOM 字符，影响后续处理。读取时用encoding='utf-8-sig'可以自动处理。

5.2 Markdown 表格转文本的取舍

Markdown 表格转成什么形式，我试过三种方案，各有适用场景。

我一般默认用键值对方案，因为配置类表格在技术文档里占比最高。对于内容型表格（比如对比表），保留原格式。自然语言化只在特定场景用，因为转换规则很难通用。

5.3 分块大小的动态调整

固定分块大小是个伪命题。不同类型的文档、不同的章节，合适的分块大小不一样。我的做法是让分块大小动态调整：技术文档的代码密集章节，块可以小一些（500 字符）；叙述性章节，块可以大一些（1200 字符）。

判断依据是章节内的"信息密度"。信息密度高的（代码、参数、列表多），块小一些保证检索精度；信息密度低的（叙述、说明），块大一些保证语义完整。信息密度可以用"非空白字符占比"和"特殊符号占比"来粗略估计。

def estimate_density(text): if not text: return 0 non_space = len(re.sub(r'\s', '', text)) special = len(re.findall(r'[=:{}()\[\]<>|]', text)) return (non_space / len(text)) * 0.7 + (special / len(text)) * 0.3 def dynamic_chunk_size(text, base=800): density = estimate_density(text) if density > 0.6: return int(base * 0.7) # 高密度，缩小 elif density < 0.3: return int(base * 1.5) # 低密度，放大 return base

这个公式是我在几个项目里调出来的，不一定通用，但思路可以借鉴：用可量化的特征来驱动分块决策，而不是拍脑袋定一个固定值。

5.4 解析结果的验证与回归测试

解析流水线上线后，必须有验证机制。我一般建一个小规模的"黄金测试集"——挑 20-30 份有代表性的文档，人工标注出期望的 chunk 数量和关键 chunk 的内容。每次修改解析逻辑，都跑一遍测试集，对比结果。

验证指标包括：chunk 总数是否合理、平均 chunk 大小是否在预期区间、是否有空 chunk、是否有超长 chunk、关键章节的 chunk 是否完整。这些指标能快速发现解析逻辑的退化。

def validate_chunks(chunks, expected_count_range=(10, 500)): issues = [] if not (expected_count_range[0] <= len(chunks) <= expected_count_range[1]): issues.append(f"chunk 数量异常: {len(chunks)}") empty = [c for c in chunks if not c['content'].strip()] if empty: issues.append(f"存在 {len(empty)} 个空 chunk") oversized = [c for c in chunks if len(c['content']) > 3000] if oversized: issues.append(f"存在 {len(oversized)} 个超长 chunk") no_path = [c for c in chunks if not c['metadata'].get('section')] if no_path: issues.append(f"存在 {len(no_path)} 个无章节路径的 chunk") return issues

这个验证函数我一般挂在流水线的最后，每次导入后自动跑，有问题就告警。它帮我提前发现过好几次解析逻辑的 bug，比如某次修改后所有 chunk 的章节路径都空了，一查是标题栈的维护逻辑写错了。

6. 从解析产物到检索就绪的最后一公里

6.1 chunk 去重与相似块合并

解析出来的 chunk 里，经常有高度相似甚至完全重复的内容。比如文档里的"注意事项"章节和"常见问题"章节可能重复了同样的内容，或者同一份文档被导入了两次。这些重复 chunk 会浪费向量库空间，还会在检索时挤占召回名额。

去重要分两级。精确去重用内容哈希，完全相同的 chunk 只保留一个。近似去重用 MinHash 或 SimHash 计算相似度，相似度超过阈值的合并。我一般用 SimHash，因为它对文本的微小改动不敏感，适合检测"换了个说法但意思一样"的重复。

def simhash(text, hash_bits=64): tokens = re.findall(r'\w+', text.lower()) if not tokens: return 0 v = [0] * hash_bits for token in tokens: h = int(hashlib.md5(token.encode()).hexdigest(), 16) for i in range(hash_bits): v[i] += 1 if (h >> i) & 1 else -1 fingerprint = 0 for i in range(hash_bits): if v[i] > 0: fingerprint |= (1 << i) return fingerprint def hamming_distance(a, b): return bin(a ^ b).count('1')

两个 chunk 的 SimHash 汉明距离小于 3（64 位下），就认为是近似重复。这个阈值可以调，越小越严格。

6.2 向量化前的文本预处理

chunk 在送进嵌入模型之前，还有几步预处理要做。去除 Markdown 残留标记（如果 chunk 里还有**、##这类符号），统一标点（全角转半角或反之，看模型训练语料），截断超长文本（超过模型最大长度的部分要截断，但要在句子边界截断）。

截断这一步要特别小心。嵌入模型一般有 512 或 8192 token 的上限，超过的部分会被静默丢弃。如果 chunk 超长且截断位置在句子中间，语义会受损。我的做法是在截断前先按句子切分，从后往前累加句子直到接近上限，保证截断在句子边界。

def truncate_at_sentence(text, max_chars): if len(text) <= max_chars: return text sentences = re.split(r'(?<=[。！？.!?])\s*', text) result = '' for sent in sentences: if len(result) + len(sent) > max_chars: break result += sent return result or text[:max_chars]

6.3 解析质量的可观测性建设

最后一点，也是很多团队忽略的：解析质量要可观测。不能等检索效果差了才回头查解析，要在解析阶段就埋好监控指标。

我一般监控这几个指标：解析成功率（多少文件成功解析，多少失败）、平均 chunk 大小（偏离预期说明分块逻辑有问题）、空 chunk 比例（应该接近 0）、元数据完整率（有多少 chunk 缺关键元数据）、解析耗时分布（发现异常慢的文件）。这些指标接入监控面板，异常时告警。

def collect_metrics(doc, chunks, elapsed): return { 'source': doc.source, 'chunk_count': len(chunks), 'avg_chunk_size': sum(len(c['content']) for c in chunks) / max(len(chunks), 1), 'empty_ratio': sum(1 for c in chunks if not c['content'].strip()) / max(len(chunks), 1), 'metadata_complete_ratio': sum( 1 for c in chunks if c['metadata'].get('section') and c['metadata'].get('source_file') ) / max(len(chunks), 1), 'elapsed_seconds': elapsed, }

这些指标积累下来，还能反哺解析逻辑的优化。比如发现某类文件的平均 chunk 大小异常小，一查是标题识别把太多行误判成标题，导致每个 chunk 都很短。没有监控，这种问题很难被发现。

我在实际项目里最深的一个体会是：RAG 的效果优化，七分在数据，三分在算法。而数据这一块，解析又是重中之重。txt 和 Markdown 这两种"简单"格式，恰恰因为大家觉得简单而疏于处理，成了效果的黑洞。把编码检测、段落还原、结构提取、元数据注入这几件事做扎实，检索端的很多问题会自然消失。后续如果要做 PDF、HTML、Word 的解析，思路是一样的——先转成统一的中间表示，再走同一套清洗和分块流水线。这套架构我用了好几个项目，扩展性经得起考验。