RAG 系统落地时最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。我见过太多团队在 Demo 阶段用几个干净的 PDF 跑通了全流程一到真实业务场景就傻眼格式五花八门、编码混乱、表格错位、层级丢失检索出来的内容驴唇不对马嘴。这个系列我打算把 RAG 数据导入与解析这条链路完整拆一遍第一篇先聚焦最基础但也最容易被轻视的部分——从纯文本 txt 到结构化 Markdown 的通用文本处理与结构化解析。你可能会想txt 有什么好讲的读进来不就完了恰恰是这个读进来不就完了的心态埋了后面无数的坑。编码识别错误导致乱码、换行符处理不当导致段落粘连、层级信息完全丢失导致检索时无法定位上下文——这些问题在数据导入阶段不解决后面检索和生成阶段花再多功夫都是白搭。这篇内容适合正在搭建 RAG 知识库的工程师、需要处理大量文档的数据从业者以及任何想把非结构化文本变成可用知识资产的人。我会从实际项目经验出发把 txt 和 Markdown 这两种最基础格式的解析逻辑、结构化策略、踩坑记录完整讲清楚。1. 为什么 txt 和 Markdown 值得单独拿出来讲1.1 纯文本是 RAG 数据源的最大公约数做 RAG 项目的人都有一个体会不管你的数据源最终是什么格式PDF 也好、Word 也好、网页也好解析到最后往往都会落到纯文本这个形态上。PDF 提取出来的是文本流Word 解析出来的是段落文本网页抓取下来的是 HTML 去标签后的内容。txt 不是一个低级格式它是所有格式解析后的公共中间态。这意味着 txt 解析的质量直接决定了后续所有环节的上限。你在 txt 阶段丢掉的结构信息在后面的分块、嵌入、检索阶段是找不回来的。我见过一个项目把技术文档的 PDF 转成 txt 时没有保留标题层级结果所有章节标题和正文混在一起检索某个 API 的参数说明时返回的片段里标题和正文搅成一团大模型根本分不清哪是标题哪是内容生成的回答质量惨不忍睹。另一个现实是很多业务系统的导出功能默认就是 txt 或 csv。日志系统导出 txt、数据库导出 csv、老旧系统只能导出纯文本。这些数据量往往很大动辄几十万行手工处理不现实必须有一套自动化的解析和结构化流程。1.2 Markdown 是结构化文本的最佳中间格式Markdown 在 RAG 流程里的地位很特殊。它既是数据源格式很多技术文档、笔记本身就是 Markdown又是理想的中间表示格式。为什么因为 Markdown 用极简的语法表达了丰富的结构信息标题层级用#数量表示列表用-或数字表示代码块用反引号包裹表格用管道符分隔。这些结构信息对 RAG 的分块策略至关重要。举个例子一个 Markdown 文档里的二级标题##天然就是一个语义边界。你可以按标题层级来分块保证每个块内的内容语义完整而不是机械地按字符数截断。代码块的存在让你知道这段内容不应该被随意切分表格的存在让你知道这里需要特殊处理。这些信息如果丢失了分块质量会断崖式下降。而且 Markdown 的可读性和可编辑性都很好。解析过程中如果发现结构有问题人工介入修正的成本很低。相比之下直接操作 PDF 的解析结果或者 HTML 的 DOM 树调试起来痛苦得多。所以我的建议是不管原始格式是什么尽量先转成 Markdown在 Markdown 层面做结构化和清洗最后再转成适合嵌入的纯文本或带元数据的结构化数据。1.3 两种格式在 RAG 流程中的定位差异虽然都是文本txt 和 Markdown 在 RAG 流程里的处理策略完全不同。txt 的核心问题是结构缺失你需要从纯文本里推断出结构Markdown 的核心问题是结构已有但需要正确解析你需要把已有的标记转换成程序可用的元数据。这个差异决定了处理思路的不同。处理 txt 时你更像一个考古学家要从蛛丝马迹里还原出文档的原始结构——空行可能是段落分隔缩进可能是层级关系特定的符号模式可能是标题。处理 Markdown 时你更像一个翻译官要把 Markdown 语法准确地转换成结构化的数据模型同时处理好各种边界情况比如嵌套列表、代码块内的特殊字符、表格的对齐方式等。理解了这个定位差异后面的技术方案选择就顺理成章了。2. txt 解析的第一道坎编码与换行符2.1 编码识别不是猜而是验处理 txt 文件第一个拦路虎就是编码。UTF-8、GBK、GB2312、GB18030、Latin-1、UTF-16甚至还有带 BOM 的 UTF-8。你永远不知道用户扔过来的文件是什么编码。很多人的做法是用chardet或charset-normalizer这类库去检测然后直接按检测结果解码。这个做法在大多数情况下能用但有一个致命问题检测库给出的是概率判断不是确定结论。我踩过的一个坑是一个 GBK 编码的文件被检测成了 GB18030大部分内容能正常解码但少数生僻字变成了乱码。因为 GB18030 是 GBK 的超集用 GB18030 解码 GBK 内容通常不会报错但某些字节序列在两个编码下的解释不同。这种看起来正常但局部乱码的问题最隐蔽往往到检索阶段才发现某些关键词死活搜不到。更稳妥的做法是验证优先先用检测结果尝试解码然后对解码后的文本做合理性校验。校验规则可以包括是否包含大量不可打印字符、是否包含常见的乱码模式比如连续的\ufffd替换字符、中文字符占比是否合理。如果校验不通过就换一个编码重试。下面是一个我实际在用的解码策略def decode_text(raw_bytes): # 优先尝试 UTF-8因为它是互联网事实标准 encodings [utf-8, gb18030, gbk, big5, latin-1] for enc in encodings: try: text raw_bytes.decode(enc) # 校验替换字符比例不能过高 if text.count(\ufffd) / max(len(text), 1) 0.001: return text, enc except (UnicodeDecodeError, LookupError): continue # 兜底用 errorsreplace 强行解码但记录警告 return raw_bytes.decode(utf-8, errorsreplace), utf-8-replace注意latin-1放在最后因为它能解码任何字节序列永远不会抛异常但解出来的可能是完全错误的文本。所以它只能作为最后的兜底而且必须配合人工检查。提示如果数据源可控最好在数据导入规范里强制要求 UTF-8 编码。这一条规则能省掉后面 80% 的编码问题。对于不可控的数据源建议在解析阶段就把编码检测结果记录下来方便后续排查。2.2 换行符的三种形态与统一策略换行符看起来简单实际上有\nUnix、\r\nWindows、\r老 Mac三种。Python 的open()在文本模式下默认会做通用换行处理把\r\n和\r都转成\n。但这个便利有时候会帮倒忙。比如处理某些从老旧系统导出的文件时\r可能不是换行符而是内容的一部分比如某些终端控制字符。如果你无脑统一成\n可能会破坏原始内容。我的做法是先用二进制模式读取自己检测换行符类型统计各种换行符的出现频率然后决定统一策略。def normalize_line_endings(raw_bytes): crlf_count raw_bytes.count(b\r\n) lf_count raw_bytes.count(b\n) - crlf_count cr_count raw_bytes.count(b\r) - crlf_count # 如果 \r 单独出现且数量很少可能是内容而非换行 if cr_count 0 and cr_count lf_count * 0.01: # 保留 \r只统一 \r\n 为 \n return raw_bytes.replace(b\r\n, b\n) # 否则全部统一为 \n return raw_bytes.replace(b\r\n, b\n).replace(b\r, b\n)这个逻辑的核心判断是如果单独的\r数量极少它更可能是内容中的特殊字符而不是换行符。这个经验来自处理日志文件的经历——某些日志里会嵌入终端控制序列里面就包含\r。2.3 BOM 头的处理与陷阱BOMByte Order Mark是 UTF-8、UTF-16 等编码在文件开头可能存在的标记字节。UTF-8 的 BOM 是EF BB BF解码后是\ufeff。这个字符不可见但会带来实际问题如果你用utf-8解码带 BOM 的文件文本开头会多一个\ufeff字符导致第一个词的匹配失败。处理方式很简单用utf-8-sig编码解码即可自动去除 BOM。但要注意有些文件可能包含多个 BOM比如拼接过的文件或者 BOM 出现在文件中间。所以除了用utf-8-sig还应该在解码后做一次全局的\ufeff清理text text.replace(\ufeff, )这个清理要放在所有文本处理的最前面因为 BOM 字符会干扰后续所有的正则匹配和字符串操作。3. 从纯文本推断结构标题、段落与列表的识别3.1 空行是段落边界的第一信号纯文本没有显式的段落标记空行是最可靠的段落分隔信号。但空行的定义需要明确是零个字符的行还是只包含空白字符的行我的经验是两者都算但处理时要区分。连续多个空行通常表示更大的语义间隔可能对应章节分隔。处理策略上我一般先把连续空行压缩成一个然后以空行为界切分段落。但这里有个坑有些文档用空行来分隔列表项有些用空行来分隔段落还有些用空行来做视觉留白。如果不加区分地按空行切分列表会被切得七零八落。我的做法是先做一轮结构探测统计空行前后内容的特征。如果空行前后的行都以列表标记开头如-、*、数字加点那这个空行大概率是列表项之间的分隔不应该作为段落边界。如果空行前后的行长度差异很大或者其中一行明显是标题特征短、无标点、可能全大写或包含特定符号那这个空行就是结构边界。3.2 标题识别的启发式规则从纯文本里识别标题本质上是一个模式匹配问题。我总结了几条在实际项目中命中率比较高的规则第一独立成行且长度较短通常少于 50 个字符。标题一般不会太长太长的行更可能是正文。第二行首或行尾没有标点符号。中文标题很少以句号结尾英文标题也很少以句点结尾。如果一行以句号、逗号、分号结尾基本可以排除标题。第三可能包含编号模式。比如第一章、1.、1.1、一、、一等。这些编号模式是强信号命中后基本可以确定是标题。第四前后有空行包围。标题通常独立成段前后会有空行。第五字体或格式信息如果有的话。纯文本没有字体信息但如果是从富文本转换来的可能会保留一些标记比如全角空格缩进、特殊符号前缀等。把这些规则组合起来可以写一个打分函数给每一行算一个标题可能性分数超过阈值的就标记为标题。下面是一个简化版的实现思路import re HEADING_PATTERNS [ (r^第[一二三四五六七八九十百千][章节篇部分], 10), # 第X章 (r^\d\.\d\.\d\s, 9), # 1.1.1 三级编号 (r^\d\.\d\s, 8), # 1.1 二级编号 (r^\d[\.、]\s, 7), # 1. 一级编号 (r^[一二三四五六七八九十][、.], 7), # 一、 (r^[(][一二三四五六七八九十][)], 6), # 一 ] def score_heading(line, prev_blank, next_blank): score 0 stripped line.strip() if not stripped: return 0 # 长度惩罚 if len(stripped) 50: return 0 # 编号模式加分 for pattern, weight in HEADING_PATTERNS: if re.match(pattern, stripped): score weight break # 前后空行加分 if prev_blank: score 3 if next_blank: score 3 # 无结尾标点加分 if stripped and stripped[-1] not in 。、.,;:!?: score 2 return score这套规则不是万能的不同类型的文档需要调整权重。比如技术文档里1.1这种编号很常见但在小说文本里就很少见。所以实际使用时最好先拿几份代表性文档做一轮人工标注根据标注结果调整规则和阈值。3.3 列表项的识别与层级还原列表的识别相对直接难点在于层级还原。纯文本里的列表层级通常靠缩进来表示但缩进可能是空格也可能是制表符空格的数量也不统一。有的文档用 2 个空格表示一级缩进有的用 4 个还有的混用。我的处理策略是先统计所有列表项的缩进量做聚类分析把相近的缩进量归为一组每组对应一个层级。这样即使缩进量不标准也能正确还原层级关系。def detect_list_levels(indents): # indents 是所有列表项的缩进量列表 unique sorted(set(indents)) if not unique: return {} # 聚类差距小于 2 个字符的归为一组 groups [[unique[0]]] for ind in unique[1:]: if ind - groups[-1][-1] 2: groups[-1].append(ind) else: groups.append([ind]) # 每组取中位数作为该层级的代表缩进 level_map {} for level, group in enumerate(groups): median group[len(group) // 2] for ind in group: level_map[ind] level return level_map这个聚类思路的关键是差距小于 2 归为一组因为实际文档里同一层级的缩进可能有 1-2 个字符的浮动。这个阈值可以根据文档情况调整但一般 2 是个比较稳妥的值。列表标记本身也需要归一化。-、*、都表示无序列表1.、1)、(1)都表示有序列表。归一化后统一转成 Markdown 的列表语法方便后续处理。4. Markdown 解析语法树构建与边界情况处理4.1 为什么不用正则而是用解析器很多人处理 Markdown 的第一反应是写正则。标题用^#\s匹配代码块用匹配看起来很简单。但 Markdown 的语法比表面看起来复杂得多正则很快就会遇到搞不定的情况。比如代码块里的#不是标题行内代码里的*不是强调嵌套列表的缩进规则和代码块冲突表格里的管道符需要转义。这些边界情况用正则处理会越来越复杂最后变成一堆难以维护的补丁。正确的做法是用成熟的 Markdown 解析器比如 Python 的markdown-it-py、mistune或者markdown库。这些库会把 Markdown 解析成 AST抽象语法树你可以遍历 AST 来提取结构信息。AST 的好处是结构明确每个节点有类型和属性不用自己处理语法歧义。from markdown_it import MarkdownIt md MarkdownIt() tokens md.parse(markdown_text) def walk_tokens(tokens, depth0): for token in tokens: if token.type heading_open: level int(token.tag[1]) # h1 - 1, h2 - 2 print( * depth fHeading level {level}) elif token.type inline: print( * depth fText: {token.content[:50]}) elif token.type fence: print( * depth fCode block: {token.info}) elif token.type table_open: print( * depth Table start)用解析器的另一个好处是它能正确处理嵌套结构。比如列表里嵌套代码块、引用块里嵌套列表这些用正则几乎不可能正确处理但 AST 天然支持。4.2 标题层级与分块策略的联动Markdown 的标题层级是分块的金矿。一个设计良好的 Markdown 文档标题层级就是天然的语义树。你可以按标题来分块保证每个块是一个完整的语义单元。具体策略是以二级标题##为主要的块边界一级标题#作为更大的分组三级及以下标题作为块内的子结构。这样分出来的块大小适中语义完整。如果某个二级标题下的内容太长再按三级标题细分如果太短就向上合并到一级标题。def split_by_headings(tokens): chunks [] current_chunk {heading: None, level: 0, content: []} for token in tokens: if token.type heading_open: level int(token.tag[1]) if level 2 and current_chunk[content]: chunks.append(current_chunk) current_chunk {heading: None, level: level, content: []} current_chunk[level] level elif token.type inline: if current_chunk[heading] is None and current_chunk[level] 0: current_chunk[heading] token.content else: current_chunk[content].append(token.content) if current_chunk[content]: chunks.append(current_chunk) return chunks这个分块逻辑的核心思想是标题作为块的锚点块的内容归属于最近的上级标题。这样每个块都带有标题信息检索时可以把标题作为上下文一起返回提升检索质量。注意分块时要把标题文本也包含进块内容里或者作为元数据存储。我见过一些实现只把正文放进块里标题丢了结果检索时完全不知道这段内容属于哪个章节上下文严重缺失。4.3 代码块、表格、引用块的特殊处理代码块在 RAG 里是个特殊存在。一方面代码块里的内容往往包含关键信息比如 API 用法、配置示例不应该被丢弃另一方面代码块的格式和正文差异很大直接混在一起嵌入可能影响检索效果。我的做法是给代码块打标签在元数据里标记content_type: code并且记录代码语言从python这样的标记里提取。检索时可以根据查询类型决定是否优先返回代码块。比如用户问怎么配置代码块和配置说明都应该返回用户问这个函数做什么可能正文说明更相关。表格的处理更麻烦。Markdown 表格在解析成 AST 后是一系列tr和td节点需要重组成结构化的数据。我的建议是把表格转成两种形式一种是保留 Markdown 原文作为整体嵌入另一种是转成表头: 值的键值对形式方便检索。比如一个参数表转成参数名: timeout, 类型: int, 默认值: 30, 说明: 请求超时时间这样的文本检索timeout 默认值时就能命中。引用块通常表示补充说明或注意事项在分块时应该和它所属的正文块放在一起不要单独切分。因为引用块单独拿出来往往语义不完整需要上下文才能理解。5. 结构化输出的数据模型设计5.1 块级数据的字段设计解析完文本后需要把结果组织成结构化的数据模型。这个模型的设计直接影响后续的检索和生成效果。我经过多个项目迭代总结出一套比较通用的字段设计字段名类型说明chunk_idstring块的唯一标识建议用文档ID_序号格式doc_idstring所属文档的标识contentstring块的文本内容content_typeenum内容类型text/code/table/listheading_pathlist标题路径如[第一章, 1.1 概述]levelint块所属的标题层级positionint块在文档中的顺序位置char_countint字符数用于分块质量监控metadatadict扩展元数据如代码语言、表格列名等heading_path这个字段特别重要。它记录了块从根标题到当前块的完整路径检索时可以把路径作为上下文一起返回。比如检索到一段关于超时配置的内容返回时带上[API 文档, 请求配置, 超时设置]这样的路径大模型就能准确理解这段内容的语境。position字段用于保持块的顺序。有些检索场景需要按原文顺序返回多个块比如总结这一章的内容就需要按 position 排序后拼接。5.2 标题路径的构建与维护标题路径的构建需要在解析过程中维护一个标题栈。遇到新标题时根据层级弹出栈中层级大于等于当前层级的标题然后压入当前标题。这样栈里的内容就是当前块的完整标题路径。class HeadingTracker: def __init__(self): self.stack [] # [(level, text), ...] def push(self, level, text): # 弹出所有层级 当前层级的标题 while self.stack and self.stack[-1][0] level: self.stack.pop() self.stack.append((level, text)) def get_path(self): return [text for _, text in self.stack]这个逻辑看起来简单但实际处理时要注意几个边界情况。第一文档可能不以一级标题开头直接从二级标题开始这时栈是空的路径就只有二级标题。第二可能出现层级跳跃比如从一级直接跳到三级中间没有二级。这时路径里就只有一级和三级中间缺失的层级不用补。第三有些文档的标题层级不规范比如用一级标题做章节又用一级标题做小节这时需要根据上下文判断或者干脆按出现顺序处理。5.3 元数据的扩展与检索增强除了基础字段元数据的设计要考虑到检索增强的需求。我一般会预留几个扩展字段keywords从块内容里提取的关键词可以用 TF-IDF 或 TextRank 算法自动提取也可以人工标注。检索时可以做关键词匹配的加权。summary块的摘要对于长块特别有用。可以用抽取式摘要算法生成或者直接取块的前 N 个字符。检索时如果块太长可以先返回摘要需要详情时再返回全文。references块内引用的其他块或文档的 ID。比如详见第 3 章这样的引用可以解析出来建立块之间的关联。检索时可以做关联扩展返回相关块。source_info数据来源信息如原始文件名、导入时间、解析版本等。用于问题追溯和数据治理。这些元数据不是每个块都必须有但设计数据模型时要预留位置。等到需要时再补比后期改数据结构要省事得多。6. 实战中的踩坑记录与排查思路6.1 乱码问题的完整排查链路乱码是 txt 处理最常见的问题但排查起来往往没有头绪。我整理了一套排查链路按顺序走基本能定位问题。第一步确认原始字节。用十六进制查看器打开文件看开头的字节序列。如果开头是EF BB BF说明有 UTF-8 BOM如果是FF FE或FE FF说明是 UTF-16。这一步能排除编码类型判断错误。第二步用不同编码解码同一段字节对比结果。如果某个编码解出来的文本可读基本可以确定编码。注意要选一段包含中文或特殊字符的内容来对比纯 ASCII 内容在任何编码下都一样没有区分度。第三步检查是否有混合编码。有些文件是拼接而成的前半部分是一种编码后半部分是另一种。这种情况检测库通常会给出错误结果。排查方法是分段检测把文件切成多个小块分别检测编码。第四步检查是否有二进制内容混入。有些 txt 文件里嵌入了图片或其他二进制数据这些字节用文本编码解码会产生乱码。排查方法是查找不可打印字符的分布如果集中在某个区域很可能是二进制内容。我遇到过一个典型案例一个日志文件前面几万行都正常从某一行开始全是乱码。用分段检测发现从那一行开始编码从 UTF-8 变成了 GBK。原因是日志系统在某个时间点做了升级改变了输出编码。这种问题只能靠分段检测发现。6.2 分块过大或过小的调优经验分块大小是 RAG 里最需要调的参数之一。太大检索精度下降因为一个块里混了太多主题太小上下文丢失大模型拿到的信息不完整。我的经验值是中文文本每块 300-500 字英文文本每块 200-400 词。但这个值不是固定的要根据内容类型调整。技术文档可以小一些因为概念密集叙述性文本可以大一些因为需要上下文连贯。调优的方法是先按默认值分块然后抽样检查。看每个块是否语义完整是否有明显的主题混杂。如果发现某个块里包含多个不相关的主题说明块太大需要按标题或段落进一步切分。如果发现某个块的内容需要结合前后块才能理解说明块太小需要合并。还有一个技巧是设置块大小的上下限。比如最小 100 字最大 800 字。小于下限的块尝试和相邻块合并大于上限的块强制切分。这样能避免极端情况。def adjust_chunk_size(chunks, min_size100, max_size800): adjusted [] buffer None for chunk in chunks: if buffer is not None: # 尝试合并 if len(buffer[content]) len(chunk[content]) max_size: buffer[content] \n chunk[content] continue else: adjusted.append(buffer) buffer None if len(chunk[content]) min_size: buffer chunk elif len(chunk[content]) max_size: # 强制切分按句子边界 adjusted.extend(force_split(chunk, max_size)) else: adjusted.append(chunk) if buffer is not None: adjusted.append(buffer) return adjusted6.3 结构信息丢失的预防措施结构信息丢失往往在解析阶段就发生了但到检索阶段才暴露。预防的关键是在解析阶段就把结构信息完整提取出来并且在后续每个环节都保留。我的做法是在解析输出里强制包含heading_path和content_type字段并且在分块、嵌入、存储的每个环节都传递这些字段。嵌入时可以把标题路径拼接到内容前面一起嵌入增强语义。存储时把结构信息放在元数据里检索时可以过滤和排序。还有一个容易被忽略的点解析日志。每次解析都记录一份日志包括处理的文件、检测到的编码、识别出的标题数量、分块数量、异常情况等。这份日志在排查问题时非常有用。比如发现某个文档检索效果差可以查解析日志看是不是标题识别失败导致分块混乱。提示建议在解析流程里加一个结构完整性检查步骤。检查项包括标题层级是否连续没有从一级直接跳到三级、每个块是否都有标题路径、代码块是否正确闭合、表格行列数是否一致。这些检查能提前发现大部分结构问题。7. 一套可复用的解析流程设计7.1 流程编排与模块划分把前面讲的内容串起来一个完整的解析流程应该包含这几个模块编码检测与解码、换行符归一化、格式识别txt 还是 Markdown、结构解析、分块、元数据提取、输出序列化。每个模块的职责要单一输入输出要明确。这样方便单独测试和替换。比如编码检测模块输入是字节流输出是文本和编码类型结构解析模块输入是文本和格式类型输出是带结构的块列表。class DocumentParser: def __init__(self, config): self.config config self.decoder TextDecoder() self.normalizer LineNormalizer() self.txt_parser TxtStructureParser() self.md_parser MarkdownStructureParser() self.chunker SemanticChunker(config) def parse(self, raw_bytes, filename): text, encoding self.decoder.decode(raw_bytes) text self.normalizer.normalize(text) fmt self.detect_format(filename, text) if fmt markdown: blocks self.md_parser.parse(text) else: blocks self.txt_parser.parse(text) chunks self.chunker.chunk(blocks) return self.serialize(chunks, filename, encoding)这个设计的核心是管道式处理每一步的输出是下一步的输入中间状态清晰可见。调试时可以单独跑某一步看输出是否符合预期。7.2 格式自动识别的判断依据格式识别不能只看文件扩展名因为.txt文件里可能是 Markdown 内容.md文件里也可能是纯文本。我的判断依据是内容特征Markdown 的强特征包括行首的#标题标记、代码块标记、-或1.列表标记、[text](url)链接语法、**bold**强调语法、|表格语法。如果这些特征在文本中出现的频率超过一定阈值就判定为 Markdown。def detect_format(filename, text): if filename.lower().endswith(.md): return markdown lines text.split(\n)[:100] # 只看前100行 md_signals 0 for line in lines: stripped line.strip() if re.match(r^#{1,6}\s, stripped): md_signals 2 elif stripped.startswith(): md_signals 2 elif re.match(r^[-*]\s, stripped): md_signals 1 elif re.match(r^\d\.\s, stripped): md_signals 1 elif | in stripped and stripped.count(|) 2: md_signals 1 return markdown if md_signals 3 else txt阈值设为 3 是个经验值。太低会误判把包含少量列表的纯文本当成 Markdown太高会漏判把结构简单的 Markdown 当成纯文本。实际使用时可以根据数据源特点调整。7.3 输出格式与下游对接解析的输出格式要考虑到下游的使用。如果下游是向量数据库输出应该是 JSON 格式每个块一个对象包含内容和元数据。如果下游是搜索引擎可能需要输出成特定的文档格式。如果下游是人工检查输出成 Markdown 或 HTML 更友好。我一般会输出两种格式一种是 JSONL每行一个 JSON 对象方便程序处理一种是 Markdown 预览文件方便人工检查。JSONL 的每一行包含块的所有字段Markdown 预览则把块按顺序展示带上标题路径和类型标记。def serialize_to_jsonl(chunks, output_path): with open(output_path, w, encodingutf-8) as f: for chunk in chunks: f.write(json.dumps(chunk, ensure_asciiFalse) \n) def serialize_to_preview(chunks, output_path): with open(output_path, w, encodingutf-8) as f: for chunk in chunks: path .join(chunk[heading_path]) f.write(f### [{chunk[content_type]}] {path}\n\n) f.write(chunk[content] \n\n---\n\n)预览文件在调试时特别有用。你可以快速浏览解析结果看分块是否合理、标题路径是否正确、有没有明显的内容丢失。我每次调整解析参数后都会生成预览文件人工抽查几十个块确认没问题再跑全量。8. 一些容易被忽略的细节8.1 空文档和超短文档的处理实际数据里总有一些空文档或只有几个字符的文档。这些文档如果直接进入解析流程可能会产生空块或极短的块影响后续检索。我的做法是在解析前先做一轮过滤字符数少于 10 的文档直接跳过并记录到日志里。如果这类文档占比很高说明数据源有问题需要回头检查数据采集环节。还有一种情况是文档内容全是空白字符或特殊符号。这种也要过滤掉。判断方法是统计有效字符字母、数字、中文的占比低于 10% 的视为无效文档。8.2 重复内容的检测与去重同一个知识库里可能有重复内容比如同一份文档被导入了两次或者不同文档里有大段相同的内容。重复内容会导致检索时返回多个相似结果浪费上下文窗口。去重可以在块级别做。计算每个块的 SimHash 或 MinHash相似的块只保留一个。SimHash 的优点是计算快适合大规模数据MinHash 更准确但计算量大。对于 RAG 场景SimHash 通常够用。def simhash(text, hash_bits64): # 简化版 SimHash 实现 tokens tokenize(text) v [0] * hash_bits for token in tokens: h hash_function(token) for i in range(hash_bits): if h (1 i): v[i] 1 else: v[i] - 1 fingerprint 0 for i in range(hash_bits): if v[i] 0: fingerprint | (1 i) return fingerprint def hamming_distance(h1, h2): return bin(h1 ^ h2).count(1)两个块的 SimHash 汉明距离小于 3 就认为是重复的。这个阈值可以根据实际数据调整但 3 是个比较通用的值。8.3 解析性能的优化思路当文档数量达到几万甚至几十万时解析性能就成了问题。优化的思路有几个第一并行处理。解析是 CPU 密集型任务可以用多进程并行。Python 的multiprocessing或者concurrent.futures都可以。注意要按文件粒度并行不要按块并行因为块之间有依赖关系。第二缓存中间结果。编码检测、格式识别这些步骤的结果可以缓存避免重复计算。如果同一批文件需要多次解析比如调整了分块参数缓存能省很多时间。第三惰性解析。如果只需要文档的元数据标题、字数等不需要全文内容可以只解析到结构层不生成完整的块。这在做数据摸底时很有用。第四选择合适的解析库。Markdown 解析库的性能差异很大mistune比markdown-it-py快但功能少一些。如果对性能要求高可以先用快速库做初步解析遇到复杂情况再切换到功能全的库。我在一个项目里处理 10 万份文档用多进程并行加上缓存把解析时间从 8 小时压到了 40 分钟。关键优化点是编码检测结果缓存、Markdown 解析用mistune、分块用 C 扩展加速的字符串操作。8.4 解析质量的可视化监控解析质量不能靠感觉要有数据支撑。我一般会统计几个指标平均块大小、块大小分布、标题识别率、代码块占比、表格占比、异常块比例。这些指标画成图表能直观看出解析质量的变化。比如块大小分布如果出现双峰分布说明分块策略可能有问题一部分块太小一部分块太大。标题识别率突然下降说明遇到了新类型的文档需要调整识别规则。异常块比例上升说明数据源质量在下降需要检查上游。这些监控数据最好能按文档类型、来源、时间维度拆分这样能快速定位问题。我在实际项目里用 Grafana 做了一个解析质量看板每次数据导入后自动更新有问题能第一时间发现。9. 从解析到入库的衔接要点9.1 块与向量的对应关系解析产出的块最终要转成向量存入向量数据库。这里有个设计决策一个块对应一个向量还是一个块拆成多个向量我的建议是一个块一个向量保持块和向量的对应关系简单清晰。如果块太大应该在解析阶段就切小而不是在嵌入阶段拆成多个向量。嵌入时要注意把标题路径拼接到内容前面。比如块内容是超时时间默认为 30 秒标题路径是[API 文档, 请求配置]嵌入的文本应该是API 文档 请求配置 超时时间默认为 30 秒。这样嵌入向量里就包含了上下文信息检索请求配置相关的超时设置时更容易命中。9.2 元数据的存储与索引元数据要存两份一份和向量一起存在向量数据库里用于检索时过滤和返回一份存在关系数据库或文档数据库里用于管理和分析。向量数据库的元数据存储能力有限复杂的查询和分析还是要靠传统数据库。需要建索引的元数据字段包括doc_id按文档查块、content_type按类型过滤、heading_path按章节查块、position按顺序返回。这些索引能显著提升检索和管理的效率。9.3 增量更新与版本管理知识库不是一次建好就完事了需要持续更新。增量更新的关键是识别哪些文档变了、哪些块变了。我的做法是给每个文档算一个内容哈希导入时对比哈希只有哈希变了才重新解析。解析后对比新旧块的哈希只更新变化的块。版本管理要记录每次导入的批次信息导入时间、文档数量、块数量、变更数量。这样出问题时可以回滚到上一个版本。向量数据库一般不支持事务回滚所以要在应用层实现版本管理比如用不同的 collection 或 namespace 来隔离不同版本。10. 个人实操体会这套解析流程我在多个项目里迭代过最大的体会是解析阶段多花一小时后面能省十小时。很多团队急于跑通全流程解析做得很粗糙结果检索效果差回头排查发现是解析阶段丢了结构信息又得重新解析、重新嵌入浪费的时间远超当初认真做解析的时间。另一个体会是不要追求一步到位。解析规则不可能一开始就完美先覆盖 80% 的常见情况剩下的 20% 通过监控发现后再逐步补充。我一般会先做一个基础版解析器跑一批数据看解析质量指标然后针对问题最多的文档类型做专项优化。这样迭代几轮解析质量就能达到可用水平。还有一点解析结果一定要人工抽查。自动化指标能发现大部分问题但有些问题只有人眼才能看出来。比如标题识别错了、段落合并错了、代码块被截断了这些在指标上可能不明显但人工一看就知道。我每个项目都会抽查至少 100 个块确认解析质量符合预期。最后解析日志和监控要尽早建。不要等到出问题了才想起来加日志。解析过程中的编码检测结果、格式识别结果、异常情况都要记录下来。这些日志在排查问题时是无价之宝。我在一个项目里靠解析日志发现了一个隐藏很久的问题某个数据源的文档编码在特定时间点会变化导致那段时间导入的文档全部乱码。如果没有日志这个问题可能永远发现不了。