尧图网页设计 
RAG 系统里最不起眼、却最容易翻车的一环就是数据导入与解析。很多人把精力全砸在向量库选型、检索策略调优、重排序模型上结果上线之后发现召回质量死活上不去回头一查原始文档在解析阶段就已经被切得七零八落——表格串行、标题层级丢失、代码块被当成正文、公式变成乱码。我做过好几个知识库项目踩过的坑里至少有六成出在数据入口这一段而不是检索本身。这篇主要聊 txt 和 Markdown 这两类最基础的纯文本格式怎么把它们干净、结构化地导入到 RAG 流程里。别看这两种格式简单真要做到通用文本与结构化解析里面的门道比想象中多。适合正在搭 RAG 知识库、做文档预处理、或者被解析质量问题折磨过的同学参考。我会把解析思路、分块策略、元数据设计、踩坑经验都摊开讲代码能直接抄。1. 为什么纯文本反而最难解析1.1 txt 的无结构是个伪命题很多人觉得 txt 最好处理读进来就是一整块字符串直接按字数切分完事。这个想法在 demo 阶段没问题一上真实数据就崩。原因很简单txt 虽然没有显式标记但它隐含结构。一份会议纪要、一份产品需求、一份小说章节它们的结构信号藏在换行、空行、缩进、标点符号里只是没有用标签写出来而已。我拿一份典型的产品需求 txt 举例它的真实长相往往是这样一、背景 本季度用户反馈集中在搜索准确率上…… 二、目标 1. 提升召回率 2. 降低响应延迟 三、方案 3.1 数据层 ……如果你按固定 500 字硬切很可能把三、方案的标题切到上一块的末尾把3.1 数据层的内容切到下一块的开头。检索的时候用户问方案是什么命中的那块里标题残缺语义就断了。所以 txt 解析的核心不是切,而是先识别出它假装没有的结构。1.2 Markdown 的结构化是半成品Markdown 比 txt 友好它有#标题、-列表、 代码块、|表格。但它的坑在于语法宽松写法千奇百怪。同一个二级标题有人写## 标题有人写##标题没空格有人用下划线式列表有人用-有人用*有人用1.代码块有人用三个反引号带语言有人不带还有人用四个空格缩进。更麻烦的是Markdown 里经常混入 HTML 标签、数学公式$...$或$$...$$、图片引用、脚注、callout 引用块。这些元素如果解析器不认识要么被当纯文本塞进 chunk要么直接丢失。我见过最离谱的一次一份技术文档里的所有代码块因为解析器没配语言标识全被合并成一大段检索某个函数怎么用时召回的全是无关段落。所以结论是txt 要挖结构Markdown 要稳结构。两者都不能直接丢给默认的文本分割器。1.3 解析质量如何直接决定 RAG 上限这里给一个我实测过的对比。同一份 200 页的技术手册两种处理方式处理方式分块策略召回命中率人工评估 100 问回答准确率粗暴定长切分固定 512 字符无重叠61%48%结构化解析 语义分块按标题层级 段落边界89%82%差距接近 30 个百分点。原因不复杂定长切分破坏了语义单元检索时向量表示的是半句话和用户 query 的语义距离自然远。而结构化解析保留了标题、段落、列表的完整边界每个 chunk 是一个自洽的语义单元向量质量高得多。提示如果你的 RAG 召回一直不理想先别急着换 embedding 模型把解析和分块这一层重新做一遍收益往往比换模型大。2. txt 解析从换行与标点里还原结构2.1 先做编码探测别让乱码毁了一切txt 最阴间的坑是编码。中文 txt 可能是 UTF-8、GBK、GB18030、UTF-8 with BOM甚至混合编码。你直接open(path, r)读遇到 GBK 文件就是一堆乱码而且乱码之后所有结构识别全部失效。我的做法是用chardet或charset-normalizer先探测再按探测结果解码并且加一层兜底import chardet def read_txt_safe(path): with open(path, rb) as f: raw f.read() # 先探测编码 detected chardet.detect(raw) encoding detected.get(encoding) or utf-8 confidence detected.get(confidence, 0) # 置信度低时按优先级尝试 candidates [encoding, utf-8, gb18030, gbk, utf-16] for enc in candidates: try: text raw.decode(enc) # 简单校验中文文档里不应出现大量替换字符 if text.count(\ufffd) len(text) * 0.01: return text except (UnicodeDecodeError, LookupError): continue # 全部失败则忽略错误强行解码 return raw.decode(utf-8, errorsignore)这里有个经验chardet对短文本探测不准置信度低于 0.7 时不要信它直接走候选列表逐个试。另外 BOM 头\ufeff要记得 strip 掉否则第一个标题会带上不可见字符匹配正则时死活匹配不上。2.2 用正则识别标题、列表、空行边界编码搞定后开始挖结构。txt 里最常见的结构信号有几类我用一组正则来识别import re # 中文数字标题一、二、三、 RE_CN_NUM_TITLE re.compile(r^\s*[一二三四五六七八九十百][、.]\s*\S) # 阿拉伯数字标题1. 1.1 1.1.1 RE_ARAB_TITLE re.compile(r^\s*\d(\.\d)*[、.\s]\s*\S) # 括号编号一(1) 【1】 RE_BRACKET_TITLE re.compile(r^\s*[(【\[]\s*[\d一二三四五六七八九十]\s*[)】\]]\s*\S) # 列表项- * • 1) RE_LIST_ITEM re.compile(r^\s*([-*•·]|\d[)])\s\S) # 空行 RE_BLANK re.compile(r^\s*$) def classify_line(line): if RE_BLANK.match(line): return blank if RE_CN_NUM_TITLE.match(line) or RE_ARAB_TITLE.match(line) or RE_BRACKET_TITLE.match(line): return title if RE_LIST_ITEM.match(line): return list return body识别出类型后就能把扁平的文本流还原成带层级的块。这里的关键判断是标题行是分块的天然边界。遇到标题就开一个新块标题本身作为这个块的上下文前缀保留下来。2.3 段落合并与断行修复真实 txt 里经常有硬换行问题——一段话被编辑器在固定宽度处强制换行导致一句话被拆成好几行。如果你按行处理语义就碎了。修复逻辑是如果一行结尾不是句末标点。》】等且下一行不是标题/列表/空行就把两行合并。SENTENCE_END tuple(。》】)) def merge_paragraphs(lines): merged [] buffer for line in lines: stripped line.rstrip() if not stripped: if buffer: merged.append(buffer) buffer merged.append() continue line_type classify_line(stripped) if line_type in (title, list): if buffer: merged.append(buffer) buffer merged.append(stripped) continue # body 行判断是否续接上一行 if buffer and not buffer.endswith(SENTENCE_END): buffer stripped else: if buffer: merged.append(buffer) buffer stripped if buffer: merged.append(buffer) return merged这套逻辑我用了很久对会议纪要、需求文档、小说都适用。唯一要注意的是诗歌、代码片段这类故意断行的内容会被误合并所以我在解析前会先判断文档类型诗歌类直接跳过合并。2.4 把 txt 块转成带层级的中间结构解析的最终产物不该是字符串列表而应该是一个带层级和元数据的中间结构。我一般用一个轻量的 dataclassfrom dataclasses import dataclass, field from typing import List, Optional dataclass class Block: content: str level: int 0 # 标题层级0 表示正文 block_type: str body # title / list / body / code heading_path: List[str] field(default_factorylist) # 祖先标题链 source: str line_start: int 0 line_end: int 0heading_path是重点。比如一个块处在三、方案 3.1 数据层下面它的heading_path就是[三、方案, 3.1 数据层]。这个路径在后续分块和检索时价值巨大——它既是 chunk 的上下文前缀也是元数据过滤的维度。用户问数据层方案你可以直接用heading_path做过滤召回精度提升非常明显。3. Markdown 解析保住标题树与代码块3.1 选对解析器别自己写正则Markdown 解析千万别自己写正则语法边界情况太多。Python 生态里主流选择是markdown-it-py、mistune、markdownPython-Markdown。我推荐markdown-it-py因为它严格遵循 CommonMark 规范而且能输出 token 流方便做结构化处理。from markdown_it import MarkdownIt md MarkdownIt(commonmark, {html: True}) tokens md.parse(markdown_text)tokens是一个扁平列表每个 token 有type、tag、level、content、children等字段。标题是heading_open/inline/heading_close三件套代码块是fence列表是bullet_list_open等。基于 token 流你能精确还原文档树。3.2 标题树构建与 heading_path 生成遍历 token 流时维护一个标题栈。遇到heading_open就根据tagh1~h6决定层级弹出栈里层级大于等于当前的标题再压入新标题。这样每个内容块都能拿到完整的heading_path。def parse_markdown_structure(tokens): blocks [] heading_stack [] # [(level, text)] current_content [] current_type body def flush(): nonlocal current_content, current_type if current_content: blocks.append(Block( content\n.join(current_content).strip(), level0, block_typecurrent_type, heading_path[h[1] for h in heading_stack], )) current_content [] current_type body i 0 while i len(tokens): tok tokens[i] if tok.type heading_open: flush() level int(tok.tag[1]) # 下一个 inline token 是标题文本 title_text tokens[i1].content while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, title_text)) i 3 continue if tok.type fence: flush() blocks.append(Block( contenttok.content, block_typecode, heading_path[h[1] for h in heading_stack], )) i 1 continue if tok.type inline: current_content.append(tok.content) i 1 flush() return blocks这段代码是骨架实际用的时候要处理嵌套列表、表格、引用块。但核心思路就是标题栈 内容缓冲 遇到边界就 flush。3.3 代码块、表格、公式的特殊处理代码块必须单独成块绝不能和正文混在一起。原因有两个一是代码的语义和自然语言差异大混在一起会污染 embedding二是代码块往往很长混进正文会导致 chunk 超长。表格在 Markdown 里是table_open/tr_open/td_open这一套 token。我的处理方式是把表格转成结构化文本比如| 字段 | 类型 | 说明 | |------|------|------| | id | int | 主键 |转成表格字段说明 - 字段: id, 类型: int, 说明: 主键 - 字段: name, 类型: str, 说明: 名称这样检索id 字段是什么类型时向量能匹配上。如果直接保留 Markdown 表格原文|和-会干扰语义表示。数学公式$...$、$$...$$建议保留原文但单独标记block_typeformula因为公式转文本容易失真。检索时如果 query 里也有公式原文匹配反而更准。3.4 处理不规范 Markdown 的兜底策略真实世界的 Markdown 经常不规范。我遇到过的情况包括标题没空格##标题、代码块没闭合、列表缩进混乱、混入 HTML。markdown-it-py对大部分情况能容错但没闭合的代码块会把后面所有内容都吞进去。兜底策略是解析前先做一轮规范化清洗——给#后补空格、检查代码块配对数量、把 HTML 标签转成纯文本或保留标记。清洗完再解析成功率能到 95% 以上。剩下 5% 解析失败的记录日志人工介入别让脏数据静默流入知识库。4. 分块策略结构化之后怎么切4.1 为什么不能定长切前面提过定长切分的危害这里展开说原理。embedding 模型把一段文本映射成向量这个向量代表的是整段文本的语义中心。如果一段文本里混了三个不相关的主题向量就是这三个主题的平均和任何一个具体 query 的距离都变远。定长切分恰恰制造了大量这种多主题混合块。结构化分块的目标是让每个 chunk 只讲一件事。标题边界、段落边界、列表边界都是天然的一件事的分界。4.2 按标题层级切 超长再递归我的默认策略是优先按标题层级切块太大再按段落递归切块太小则向上合并。具体规则如果某个标题下的内容总长度小于min_chunk_size比如 200 字就把它和相邻同级块合并。如果某个标题下的内容超过max_chunk_size比如 1500 字就按段落边界递归切分每个子块继承父块的heading_path。代码块、表格块不参与合并独立成块。def chunk_blocks(blocks, min_size200, max_size1500): chunks [] buffer [] buffer_len 0 def emit(): nonlocal buffer, buffer_len if buffer: chunks.append(buffer) buffer [] buffer_len 0 for blk in blocks: blen len(blk.content) if blk.block_type in (code, formula): emit() chunks.append([blk]) continue if buffer_len blen max_size and buffer: emit() buffer.append(blk) buffer_len blen if buffer_len min_size: emit() emit() return chunks实际用的时候我会把heading_path拼成前缀加到 chunk 内容前面比如[三、方案 3.1 数据层] 具体内容...。这样即使 chunk 被单独检索出来也带着上下文。4.3 重叠窗口怎么设才不浪费chunk 之间加重叠是为了防止关键信息正好落在边界上被切断。但重叠不是越大越好重叠太多会导致检索结果重复、浪费存储和计算。我的经验值重叠 10%~15% 的 chunk 长度。比如 chunk 平均 800 字重叠 80~120 字就够了。而且重叠应该发生在语义连续的相邻块之间跨标题的重叠没意义因为标题本身就是硬边界。另外重叠部分建议用上一块的末尾 N 字而不是上一块的开头 N 字因为段落结尾往往是总结性内容对下一块有承接作用。4.4 元数据设计让检索能过滤chunk 不只是文本还要带元数据。我一般会存这些字段字段说明用途source源文件路径溯源、去重heading_path标题链上下文前缀、过滤block_typebody/code/table/formula按类型检索char_count字符数质量监控doc_type文档类型分类过滤updated_at更新时间时效性排序heading_path和block_type是最有用的两个。用户问代码示例你可以直接过滤block_typecode用户问某个章节的内容用heading_path做前缀匹配精度提升立竿见影。5. 踩坑实录那些让我加班到凌晨的解析问题5.1 编码探测失败导致的静默乱码有一次批量导入 3000 份 txt跑完之后检索质量奇差。排查半天发现其中约 400 份是 GBK 编码但chardet把它们误判成了Windows-1252解码后全是乱码而且因为没抛异常静默流入了知识库。教训是编码探测必须加校验。我的校验规则是——解码后统计中文字符占比如果一份明显是中文文档的文件里中文字符占比低于 30%就判定探测失败走候选列表重试。这个规则救了我很多次。5.2 Markdown 代码块未闭合吞掉半篇文档前面提过未闭合的代码块是灾难。我遇到过一次一份文档里有个 开了没关结果后面 2000 字全被当成代码。检索时用户问正文内容召回的全是代码块因为解析器把正文也标成了 code。修复方案是在解析前做配对检查def check_fence_balance(text): lines text.split(\n) fence_count sum(1 for l in lines if l.strip().startswith()) if fence_count % 2 ! 0: # 奇数个说明有未闭合 return False return True发现不配对就记录警告并尝试在文档末尾补一个 兜底。5.3 标题层级跳跃导致 heading_path 错乱有些文档从 h1 直接跳到 h3或者 h2 下面又出现 h1。如果标题栈处理不当heading_path就会错乱比如出现[第一章, 1.1 节, 第二章]这种不合逻辑的链。处理方式是压栈时如果新标题层级比栈顶大超过 1 级就按栈顶层级 1 处理并在日志里记录。这样至少保证路径是单调递增的不会出现层级回跳。5.4 表格转文本后语义丢失早期我把 Markdown 表格直接转成字段: 值的列表结果发现涉及多列关联的查询召回很差。比如表格里张三 | 25 | 北京转成姓名: 张三, 年龄: 25, 城市: 北京后用户问北京的人有哪些向量匹配不上因为北京和张三被拆散了。改进方案是同时保留两种表示一份是逐行的键值对适合精确查询一份是整表的自然语言描述适合语义查询。两份都入库检索时按 query 类型路由。6. 一套可复用的解析流水线6.1 流水线整体设计把前面的东西串起来一条完整的流水线是这样的文件读取 → 编码探测 → 格式识别(txt/md) → 结构解析 → 块构建 → 分块 → 元数据注入 → 质量校验 → 输出 JSONL每个环节都可以独立测试和替换。我习惯把中间结果落盘成 JSONL方便排查问题——哪一步出错直接看那一步的输出。6.2 关键代码骨架import json from pathlib import Path def process_file(path: Path): text read_txt_safe(str(path)) if path.suffix .txt else path.read_text(encodingutf-8) if path.suffix .md: blocks parse_markdown_structure(MarkdownIt(commonmark).parse(text)) else: lines merge_paragraphs(text.split(\n)) blocks build_blocks_from_lines(lines) chunks chunk_blocks(blocks) records [] for i, chunk in enumerate(chunks): content \n.join(b.content for b in chunk) heading_path chunk[0].heading_path prefix .join(heading_path) records.append({ id: f{path.stem}_{i}, content: f[{prefix}] {content} if prefix else content, metadata: { source: str(path), heading_path: heading_path, block_type: chunk[0].block_type, char_count: len(content), } }) return records def run_pipeline(input_dir, output_file): all_records [] for p in Path(input_dir).rglob(*): if p.suffix in (.txt, .md): try: all_records.extend(process_file(p)) except Exception as e: print(f[FAIL] {p}: {e}) with open(output_file, w, encodingutf-8) as f: for r in all_records: f.write(json.dumps(r, ensure_asciiFalse) \n) print(f共处理 {len(all_records)} 个 chunk)6.3 质量校验清单导入前一定要过一遍校验我常用的检查项chunk 长度分布有没有超长3000 字或超短50 字的异常块空内容块content 为空的直接丢弃乱码检测替换字符\ufffd占比超过 1% 的块标记为可疑heading_path 完整性正文块是否都有标题归属重复检测content 完全相同的块去重这些检查跑一遍能拦下 80% 的脏数据。6.4 增量导入与去重知识库是要持续更新的不能每次全量重跑。我的做法是用source content_hash作为唯一键导入前先查已存在的 hash相同则跳过不同则更新。content_hash 用hashlib.md5(content.encode()).hexdigest()就行。增量导入还要处理文件被删除的情况——如果某个 source 在本次扫描中没出现就把它的所有 chunk 标记为失效软删除而不是物理删除方便回溯。7. 几个容易被忽略的细节7.1 中文标点与全半角统一中文文档里经常混用全角和半角标点比如。和.、和()。这会影响正则匹配和检索一致性。建议在解析前做一轮统一全角转半角针对英文和数字部分中文标点保留全角。但要注意代码块里的内容不能转否则会破坏代码。7.2 空白字符的清理\u3000全角空格、\xa0不换行空格、\t这些字符在 txt 里很常见不清理会干扰分块。统一替换成普通空格或删除。但同样代码块里的缩进要保留。7.3 超长单行的处理有些 txt 是一整篇没有换行的比如从某些系统导出的这时候按行处理完全失效。兜底方案是按句号、问号、感叹号切句再按句子聚合。中文分句可以用正则(?[。])做切分。7.4 文件名的元数据价值文件名往往包含重要信息比如2024Q1_产品需求_v2.md。把文件名解析成{时间, 类型, 版本}存进元数据检索时能做时间过滤和版本过滤。这个细节很多人忽略但实际很有用。8. 从解析到入库的衔接解析产出的 JSONL 不是终点还要考虑怎么喂给向量库。这里有几个衔接点要注意。第一是embedding 的输入格式。我习惯把heading_path前缀和正文拼在一起送进 embedding因为标题往往包含核心关键词能提升向量质量。但前缀不要太长控制在 50 字以内。第二是批量与并发。embedding 调用是 IO 密集型的用asyncio或线程池并发能大幅提速。但要注意 API 的速率限制加个信号量控制并发数。第三是失败重试。embedding 调用可能因为网络或限流失败要有重试机制并且把失败的 chunk 单独落盘方便补跑。第四是入库幂等。用 chunk id 作为主键重复导入时覆盖而不是追加避免数据膨胀。这套流程我在几个项目里跑下来从原始 txt/Markdown 到可检索的知识库一份 1000 页的文档大概 10 分钟能处理完召回质量比粗暴切分高出一大截。真正花时间的不是写代码而是处理各种边界情况和脏数据——所以质量校验那一环千万别省。后续如果要做 PDF、Word、HTML 的解析思路是一样的先还原结构再结构化分块最后注入元数据。区别只在于解析器换成了 PyMuPDF、python-docx、BeautifulSoup 这些工具。把 txt 和 Markdown 这两类基础格式吃透其他格式就是换汤不换药。 RAG数据导入解析：txt与Markdown结构化分块实战
RAG 系统里最不起眼、却最容易翻车的一环就是数据导入与解析。很多人把精力全砸在向量库选型、检索策略调优、重排序模型上结果上线之后发现召回质量死活上不去回头一查原始文档在解析阶段就已经被切得七零八落——表格串行、标题层级丢失、代码块被当成正文、公式变成乱码。我做过好几个知识库项目踩过的坑里至少有六成出在数据入口这一段而不是检索本身。这篇主要聊 txt 和 Markdown 这两类最基础的纯文本格式怎么把它们干净、结构化地导入到 RAG 流程里。别看这两种格式简单真要做到通用文本与结构化解析里面的门道比想象中多。适合正在搭 RAG 知识库、做文档预处理、或者被解析质量问题折磨过的同学参考。我会把解析思路、分块策略、元数据设计、踩坑经验都摊开讲代码能直接抄。1. 为什么纯文本反而最难解析1.1 txt 的无结构是个伪命题很多人觉得 txt 最好处理读进来就是一整块字符串直接按字数切分完事。这个想法在 demo 阶段没问题一上真实数据就崩。原因很简单txt 虽然没有显式标记但它隐含结构。一份会议纪要、一份产品需求、一份小说章节它们的结构信号藏在换行、空行、缩进、标点符号里只是没有用标签写出来而已。我拿一份典型的产品需求 txt 举例它的真实长相往往是这样一、背景 本季度用户反馈集中在搜索准确率上…… 二、目标 1. 提升召回率 2. 降低响应延迟 三、方案 3.1 数据层 ……如果你按固定 500 字硬切很可能把三、方案的标题切到上一块的末尾把3.1 数据层的内容切到下一块的开头。检索的时候用户问方案是什么命中的那块里标题残缺语义就断了。所以 txt 解析的核心不是切,而是先识别出它假装没有的结构。1.2 Markdown 的结构化是半成品Markdown 比 txt 友好它有#标题、-列表、 代码块、|表格。但它的坑在于语法宽松写法千奇百怪。同一个二级标题有人写## 标题有人写##标题没空格有人用下划线式列表有人用-有人用*有人用1.代码块有人用三个反引号带语言有人不带还有人用四个空格缩进。更麻烦的是Markdown 里经常混入 HTML 标签、数学公式$...$或$$...$$、图片引用、脚注、callout 引用块。这些元素如果解析器不认识要么被当纯文本塞进 chunk要么直接丢失。我见过最离谱的一次一份技术文档里的所有代码块因为解析器没配语言标识全被合并成一大段检索某个函数怎么用时召回的全是无关段落。所以结论是txt 要挖结构Markdown 要稳结构。两者都不能直接丢给默认的文本分割器。1.3 解析质量如何直接决定 RAG 上限这里给一个我实测过的对比。同一份 200 页的技术手册两种处理方式处理方式分块策略召回命中率人工评估 100 问回答准确率粗暴定长切分固定 512 字符无重叠61%48%结构化解析 语义分块按标题层级 段落边界89%82%差距接近 30 个百分点。原因不复杂定长切分破坏了语义单元检索时向量表示的是半句话和用户 query 的语义距离自然远。而结构化解析保留了标题、段落、列表的完整边界每个 chunk 是一个自洽的语义单元向量质量高得多。提示如果你的 RAG 召回一直不理想先别急着换 embedding 模型把解析和分块这一层重新做一遍收益往往比换模型大。2. txt 解析从换行与标点里还原结构2.1 先做编码探测别让乱码毁了一切txt 最阴间的坑是编码。中文 txt 可能是 UTF-8、GBK、GB18030、UTF-8 with BOM甚至混合编码。你直接open(path, r)读遇到 GBK 文件就是一堆乱码而且乱码之后所有结构识别全部失效。我的做法是用chardet或charset-normalizer先探测再按探测结果解码并且加一层兜底import chardet def read_txt_safe(path): with open(path, rb) as f: raw f.read() # 先探测编码 detected chardet.detect(raw) encoding detected.get(encoding) or utf-8 confidence detected.get(confidence, 0) # 置信度低时按优先级尝试 candidates [encoding, utf-8, gb18030, gbk, utf-16] for enc in candidates: try: text raw.decode(enc) # 简单校验中文文档里不应出现大量替换字符 if text.count(\ufffd) len(text) * 0.01: return text except (UnicodeDecodeError, LookupError): continue # 全部失败则忽略错误强行解码 return raw.decode(utf-8, errorsignore)这里有个经验chardet对短文本探测不准置信度低于 0.7 时不要信它直接走候选列表逐个试。另外 BOM 头\ufeff要记得 strip 掉否则第一个标题会带上不可见字符匹配正则时死活匹配不上。2.2 用正则识别标题、列表、空行边界编码搞定后开始挖结构。txt 里最常见的结构信号有几类我用一组正则来识别import re # 中文数字标题一、二、三、 RE_CN_NUM_TITLE re.compile(r^\s*[一二三四五六七八九十百][、.]\s*\S) # 阿拉伯数字标题1. 1.1 1.1.1 RE_ARAB_TITLE re.compile(r^\s*\d(\.\d)*[、.\s]\s*\S) # 括号编号一(1) 【1】 RE_BRACKET_TITLE re.compile(r^\s*[(【\[]\s*[\d一二三四五六七八九十]\s*[)】\]]\s*\S) # 列表项- * • 1) RE_LIST_ITEM re.compile(r^\s*([-*•·]|\d[)])\s\S) # 空行 RE_BLANK re.compile(r^\s*$) def classify_line(line): if RE_BLANK.match(line): return blank if RE_CN_NUM_TITLE.match(line) or RE_ARAB_TITLE.match(line) or RE_BRACKET_TITLE.match(line): return title if RE_LIST_ITEM.match(line): return list return body识别出类型后就能把扁平的文本流还原成带层级的块。这里的关键判断是标题行是分块的天然边界。遇到标题就开一个新块标题本身作为这个块的上下文前缀保留下来。2.3 段落合并与断行修复真实 txt 里经常有硬换行问题——一段话被编辑器在固定宽度处强制换行导致一句话被拆成好几行。如果你按行处理语义就碎了。修复逻辑是如果一行结尾不是句末标点。》】等且下一行不是标题/列表/空行就把两行合并。SENTENCE_END tuple(。》】)) def merge_paragraphs(lines): merged [] buffer for line in lines: stripped line.rstrip() if not stripped: if buffer: merged.append(buffer) buffer merged.append() continue line_type classify_line(stripped) if line_type in (title, list): if buffer: merged.append(buffer) buffer merged.append(stripped) continue # body 行判断是否续接上一行 if buffer and not buffer.endswith(SENTENCE_END): buffer stripped else: if buffer: merged.append(buffer) buffer stripped if buffer: merged.append(buffer) return merged这套逻辑我用了很久对会议纪要、需求文档、小说都适用。唯一要注意的是诗歌、代码片段这类故意断行的内容会被误合并所以我在解析前会先判断文档类型诗歌类直接跳过合并。2.4 把 txt 块转成带层级的中间结构解析的最终产物不该是字符串列表而应该是一个带层级和元数据的中间结构。我一般用一个轻量的 dataclassfrom dataclasses import dataclass, field from typing import List, Optional dataclass class Block: content: str level: int 0 # 标题层级0 表示正文 block_type: str body # title / list / body / code heading_path: List[str] field(default_factorylist) # 祖先标题链 source: str line_start: int 0 line_end: int 0heading_path是重点。比如一个块处在三、方案 3.1 数据层下面它的heading_path就是[三、方案, 3.1 数据层]。这个路径在后续分块和检索时价值巨大——它既是 chunk 的上下文前缀也是元数据过滤的维度。用户问数据层方案你可以直接用heading_path做过滤召回精度提升非常明显。3. Markdown 解析保住标题树与代码块3.1 选对解析器别自己写正则Markdown 解析千万别自己写正则语法边界情况太多。Python 生态里主流选择是markdown-it-py、mistune、markdownPython-Markdown。我推荐markdown-it-py因为它严格遵循 CommonMark 规范而且能输出 token 流方便做结构化处理。from markdown_it import MarkdownIt md MarkdownIt(commonmark, {html: True}) tokens md.parse(markdown_text)tokens是一个扁平列表每个 token 有type、tag、level、content、children等字段。标题是heading_open/inline/heading_close三件套代码块是fence列表是bullet_list_open等。基于 token 流你能精确还原文档树。3.2 标题树构建与 heading_path 生成遍历 token 流时维护一个标题栈。遇到heading_open就根据tagh1~h6决定层级弹出栈里层级大于等于当前的标题再压入新标题。这样每个内容块都能拿到完整的heading_path。def parse_markdown_structure(tokens): blocks [] heading_stack [] # [(level, text)] current_content [] current_type body def flush(): nonlocal current_content, current_type if current_content: blocks.append(Block( content\n.join(current_content).strip(), level0, block_typecurrent_type, heading_path[h[1] for h in heading_stack], )) current_content [] current_type body i 0 while i len(tokens): tok tokens[i] if tok.type heading_open: flush() level int(tok.tag[1]) # 下一个 inline token 是标题文本 title_text tokens[i1].content while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, title_text)) i 3 continue if tok.type fence: flush() blocks.append(Block( contenttok.content, block_typecode, heading_path[h[1] for h in heading_stack], )) i 1 continue if tok.type inline: current_content.append(tok.content) i 1 flush() return blocks这段代码是骨架实际用的时候要处理嵌套列表、表格、引用块。但核心思路就是标题栈 内容缓冲 遇到边界就 flush。3.3 代码块、表格、公式的特殊处理代码块必须单独成块绝不能和正文混在一起。原因有两个一是代码的语义和自然语言差异大混在一起会污染 embedding二是代码块往往很长混进正文会导致 chunk 超长。表格在 Markdown 里是table_open/tr_open/td_open这一套 token。我的处理方式是把表格转成结构化文本比如| 字段 | 类型 | 说明 | |------|------|------| | id | int | 主键 |转成表格字段说明 - 字段: id, 类型: int, 说明: 主键 - 字段: name, 类型: str, 说明: 名称这样检索id 字段是什么类型时向量能匹配上。如果直接保留 Markdown 表格原文|和-会干扰语义表示。数学公式$...$、$$...$$建议保留原文但单独标记block_typeformula因为公式转文本容易失真。检索时如果 query 里也有公式原文匹配反而更准。3.4 处理不规范 Markdown 的兜底策略真实世界的 Markdown 经常不规范。我遇到过的情况包括标题没空格##标题、代码块没闭合、列表缩进混乱、混入 HTML。markdown-it-py对大部分情况能容错但没闭合的代码块会把后面所有内容都吞进去。兜底策略是解析前先做一轮规范化清洗——给#后补空格、检查代码块配对数量、把 HTML 标签转成纯文本或保留标记。清洗完再解析成功率能到 95% 以上。剩下 5% 解析失败的记录日志人工介入别让脏数据静默流入知识库。4. 分块策略结构化之后怎么切4.1 为什么不能定长切前面提过定长切分的危害这里展开说原理。embedding 模型把一段文本映射成向量这个向量代表的是整段文本的语义中心。如果一段文本里混了三个不相关的主题向量就是这三个主题的平均和任何一个具体 query 的距离都变远。定长切分恰恰制造了大量这种多主题混合块。结构化分块的目标是让每个 chunk 只讲一件事。标题边界、段落边界、列表边界都是天然的一件事的分界。4.2 按标题层级切 超长再递归我的默认策略是优先按标题层级切块太大再按段落递归切块太小则向上合并。具体规则如果某个标题下的内容总长度小于min_chunk_size比如 200 字就把它和相邻同级块合并。如果某个标题下的内容超过max_chunk_size比如 1500 字就按段落边界递归切分每个子块继承父块的heading_path。代码块、表格块不参与合并独立成块。def chunk_blocks(blocks, min_size200, max_size1500): chunks [] buffer [] buffer_len 0 def emit(): nonlocal buffer, buffer_len if buffer: chunks.append(buffer) buffer [] buffer_len 0 for blk in blocks: blen len(blk.content) if blk.block_type in (code, formula): emit() chunks.append([blk]) continue if buffer_len blen max_size and buffer: emit() buffer.append(blk) buffer_len blen if buffer_len min_size: emit() emit() return chunks实际用的时候我会把heading_path拼成前缀加到 chunk 内容前面比如[三、方案 3.1 数据层] 具体内容...。这样即使 chunk 被单独检索出来也带着上下文。4.3 重叠窗口怎么设才不浪费chunk 之间加重叠是为了防止关键信息正好落在边界上被切断。但重叠不是越大越好重叠太多会导致检索结果重复、浪费存储和计算。我的经验值重叠 10%~15% 的 chunk 长度。比如 chunk 平均 800 字重叠 80~120 字就够了。而且重叠应该发生在语义连续的相邻块之间跨标题的重叠没意义因为标题本身就是硬边界。另外重叠部分建议用上一块的末尾 N 字而不是上一块的开头 N 字因为段落结尾往往是总结性内容对下一块有承接作用。4.4 元数据设计让检索能过滤chunk 不只是文本还要带元数据。我一般会存这些字段字段说明用途source源文件路径溯源、去重heading_path标题链上下文前缀、过滤block_typebody/code/table/formula按类型检索char_count字符数质量监控doc_type文档类型分类过滤updated_at更新时间时效性排序heading_path和block_type是最有用的两个。用户问代码示例你可以直接过滤block_typecode用户问某个章节的内容用heading_path做前缀匹配精度提升立竿见影。5. 踩坑实录那些让我加班到凌晨的解析问题5.1 编码探测失败导致的静默乱码有一次批量导入 3000 份 txt跑完之后检索质量奇差。排查半天发现其中约 400 份是 GBK 编码但chardet把它们误判成了Windows-1252解码后全是乱码而且因为没抛异常静默流入了知识库。教训是编码探测必须加校验。我的校验规则是——解码后统计中文字符占比如果一份明显是中文文档的文件里中文字符占比低于 30%就判定探测失败走候选列表重试。这个规则救了我很多次。5.2 Markdown 代码块未闭合吞掉半篇文档前面提过未闭合的代码块是灾难。我遇到过一次一份文档里有个 开了没关结果后面 2000 字全被当成代码。检索时用户问正文内容召回的全是代码块因为解析器把正文也标成了 code。修复方案是在解析前做配对检查def check_fence_balance(text): lines text.split(\n) fence_count sum(1 for l in lines if l.strip().startswith()) if fence_count % 2 ! 0: # 奇数个说明有未闭合 return False return True发现不配对就记录警告并尝试在文档末尾补一个 兜底。5.3 标题层级跳跃导致 heading_path 错乱有些文档从 h1 直接跳到 h3或者 h2 下面又出现 h1。如果标题栈处理不当heading_path就会错乱比如出现[第一章, 1.1 节, 第二章]这种不合逻辑的链。处理方式是压栈时如果新标题层级比栈顶大超过 1 级就按栈顶层级 1 处理并在日志里记录。这样至少保证路径是单调递增的不会出现层级回跳。5.4 表格转文本后语义丢失早期我把 Markdown 表格直接转成字段: 值的列表结果发现涉及多列关联的查询召回很差。比如表格里张三 | 25 | 北京转成姓名: 张三, 年龄: 25, 城市: 北京后用户问北京的人有哪些向量匹配不上因为北京和张三被拆散了。改进方案是同时保留两种表示一份是逐行的键值对适合精确查询一份是整表的自然语言描述适合语义查询。两份都入库检索时按 query 类型路由。6. 一套可复用的解析流水线6.1 流水线整体设计把前面的东西串起来一条完整的流水线是这样的文件读取 → 编码探测 → 格式识别(txt/md) → 结构解析 → 块构建 → 分块 → 元数据注入 → 质量校验 → 输出 JSONL每个环节都可以独立测试和替换。我习惯把中间结果落盘成 JSONL方便排查问题——哪一步出错直接看那一步的输出。6.2 关键代码骨架import json from pathlib import Path def process_file(path: Path): text read_txt_safe(str(path)) if path.suffix .txt else path.read_text(encodingutf-8) if path.suffix .md: blocks parse_markdown_structure(MarkdownIt(commonmark).parse(text)) else: lines merge_paragraphs(text.split(\n)) blocks build_blocks_from_lines(lines) chunks chunk_blocks(blocks) records [] for i, chunk in enumerate(chunks): content \n.join(b.content for b in chunk) heading_path chunk[0].heading_path prefix .join(heading_path) records.append({ id: f{path.stem}_{i}, content: f[{prefix}] {content} if prefix else content, metadata: { source: str(path), heading_path: heading_path, block_type: chunk[0].block_type, char_count: len(content), } }) return records def run_pipeline(input_dir, output_file): all_records [] for p in Path(input_dir).rglob(*): if p.suffix in (.txt, .md): try: all_records.extend(process_file(p)) except Exception as e: print(f[FAIL] {p}: {e}) with open(output_file, w, encodingutf-8) as f: for r in all_records: f.write(json.dumps(r, ensure_asciiFalse) \n) print(f共处理 {len(all_records)} 个 chunk)6.3 质量校验清单导入前一定要过一遍校验我常用的检查项chunk 长度分布有没有超长3000 字或超短50 字的异常块空内容块content 为空的直接丢弃乱码检测替换字符\ufffd占比超过 1% 的块标记为可疑heading_path 完整性正文块是否都有标题归属重复检测content 完全相同的块去重这些检查跑一遍能拦下 80% 的脏数据。6.4 增量导入与去重知识库是要持续更新的不能每次全量重跑。我的做法是用source content_hash作为唯一键导入前先查已存在的 hash相同则跳过不同则更新。content_hash 用hashlib.md5(content.encode()).hexdigest()就行。增量导入还要处理文件被删除的情况——如果某个 source 在本次扫描中没出现就把它的所有 chunk 标记为失效软删除而不是物理删除方便回溯。7. 几个容易被忽略的细节7.1 中文标点与全半角统一中文文档里经常混用全角和半角标点比如。和.、和()。这会影响正则匹配和检索一致性。建议在解析前做一轮统一全角转半角针对英文和数字部分中文标点保留全角。但要注意代码块里的内容不能转否则会破坏代码。7.2 空白字符的清理\u3000全角空格、\xa0不换行空格、\t这些字符在 txt 里很常见不清理会干扰分块。统一替换成普通空格或删除。但同样代码块里的缩进要保留。7.3 超长单行的处理有些 txt 是一整篇没有换行的比如从某些系统导出的这时候按行处理完全失效。兜底方案是按句号、问号、感叹号切句再按句子聚合。中文分句可以用正则(?[。])做切分。7.4 文件名的元数据价值文件名往往包含重要信息比如2024Q1_产品需求_v2.md。把文件名解析成{时间, 类型, 版本}存进元数据检索时能做时间过滤和版本过滤。这个细节很多人忽略但实际很有用。8. 从解析到入库的衔接解析产出的 JSONL 不是终点还要考虑怎么喂给向量库。这里有几个衔接点要注意。第一是embedding 的输入格式。我习惯把heading_path前缀和正文拼在一起送进 embedding因为标题往往包含核心关键词能提升向量质量。但前缀不要太长控制在 50 字以内。第二是批量与并发。embedding 调用是 IO 密集型的用asyncio或线程池并发能大幅提速。但要注意 API 的速率限制加个信号量控制并发数。第三是失败重试。embedding 调用可能因为网络或限流失败要有重试机制并且把失败的 chunk 单独落盘方便补跑。第四是入库幂等。用 chunk id 作为主键重复导入时覆盖而不是追加避免数据膨胀。这套流程我在几个项目里跑下来从原始 txt/Markdown 到可检索的知识库一份 1000 页的文档大概 10 分钟能处理完召回质量比粗暴切分高出一大截。真正花时间的不是写代码而是处理各种边界情况和脏数据——所以质量校验那一环千万别省。后续如果要做 PDF、Word、HTML 的解析思路是一样的先还原结构再结构化分块最后注入元数据。区别只在于解析器换成了 PyMuPDF、python-docx、BeautifulSoup 这些工具。把 txt 和 Markdown 这两类基础格式吃透其他格式就是换汤不换药。