1. 为什么文本导入是 RAG 系统最容易被低估的一环做过 RAG 项目的人都有一个共同体会模型选型、向量库选型、检索策略调优这些环节网上教程一抓一大把但真正让系统效果拉胯的往往是最不起眼的数据导入与解析环节。我见过太多团队花两周时间调 embedding 模型结果发现原始文档里全是乱码、断行、页眉页脚混入正文检索出来的 chunk 根本没法用。这篇内容聚焦一个非常具体的问题如何把 txt 和 Markdown 这两类最基础的文本格式干净、结构化地导入到 RAG 知识库中。别看这两种格式简单实际操作中踩的坑一点都不少——编码问题、段落切分、标题层级识别、代码块保护、表格处理每一个细节都会直接影响后续的检索质量。适合阅读这篇内容的人包括正在搭建 RAG 知识库的工程师、需要批量处理文档的数据从业者、对文档结构化解析感兴趣的技术爱好者。哪怕你之前没接触过 RAG只要跟着思路走也能理解为什么把 txt 读进来这件事值得单独写一篇攻略。先说结论txt 和 Markdown 的解析难点不在读而在切和保。读进来只是第一步怎么切分成语义完整的 chunk、怎么保留原文的结构信息、怎么避免把代码块和表格切碎才是决定 RAG 效果的关键。下面我会从整体设计思路开始一步步拆解到具体实现。2. 整体设计思路与方案选型拆解2.1 先搞清楚RAG 对文本 chunk 到底有什么要求在动手写解析代码之前得先明确目标。RAG 系统里的 chunk 不是随便切的文本片段它需要满足几个硬性条件。第一语义完整性。一个 chunk 最好能独立表达一个完整的意思不能出现这句话的前半段在 chunk A后半段在 chunk B的情况。检索的时候用户的问题匹配到 chunk A但答案的关键信息在 chunk B这就废了。第二结构可追溯。chunk 需要携带来源信息比如它来自哪个文件的哪个章节。这样检索命中后可以给用户展示这段内容出自《XXX》的第 3 章第 2 节提升可信度。第三长度可控。chunk 太短语义不完整chunk 太长检索精度下降还会浪费 embedding 的 token 额度。业界常见的做法是 200 到 800 个 token 之间具体取决于文档类型和检索需求。第四特殊内容保护。代码块、表格、公式这些内容如果被切碎基本就失去意义了。解析器必须能识别这些结构并保证它们不被拦腰截断。理解了这四点后面的方案设计就有了判断标准。2.2 为什么选择先结构化、再切分的两阶段方案处理 txt 和 Markdown有两种常见思路。一种是直接按固定长度切分比如每 500 个字符切一刀。这种做法实现简单但问题很明显它完全无视文本的语义结构经常把一句话、一个段落、一个代码块切得七零八落。对于结构松散的纯 txt 可能勉强能用但对于有明确层级结构的 Markdown简直是暴殄天物。另一种是先解析结构、再按语义单元切分。先把文档解析成一棵结构树标题、段落、列表、代码块、表格各归其位然后按照结构边界来切分 chunk。这种做法实现复杂一些但切出来的 chunk 质量高得多。我选的是第二种方案理由很直接Markdown 本身就是结构化的不用白不用。Markdown 的#、##、###天然就是章节边界包裹的就是代码块|开头的就是表格。这些信息如果被忽略等于把文档里最有价值的结构信号扔掉了。对于纯 txt虽然没有显式标记但也可以通过空行、缩进、行首符号如数字编号、短横线来推断结构。退一步说即使推断不出来至少可以按段落切分比按字符数硬切要好。2.3 工具选型为什么不用现成的文档加载器市面上有不少现成的文档加载工具比如各种 LangChain 的 DocumentLoader。它们确实方便一行代码就能把文件读成 Document 对象。但在实际项目里我倾向于自己写解析逻辑原因有三个。第一可控性。现成工具的黑盒程度太高切分规则、元数据提取逻辑都是固定的遇到特殊需求很难定制。比如你想在 chunk 里保留标题路径第 2 章 2.1 节 具体内容很多加载器默认不支持。第二调试成本。当检索效果不好时你需要知道 chunk 到底是怎么切的。用现成工具出问题了只能去翻源码自己写的逻辑哪里不对一眼就能看出来。第三依赖负担。为了一个简单的 txt 解析引入一整套框架性价比不高。核心逻辑其实就几百行代码用标准库就能搞定。当然这不是说现成工具不能用。如果你的需求很标准用现成工具快速跑通也没问题。但如果你想深入理解 RAG 的数据流自己实现一遍是值得的。2.4 整体流程设计整个解析流程我拆成四个阶段用一张表说明每个阶段的输入输出和关键动作。阶段输入关键动作输出读取与编码归一原始文件检测编码、统一转 UTF-8、去除 BOM干净的文本字符串结构解析文本字符串识别标题、段落、代码块、表格、列表结构化节点列表语义切分结构化节点列表按标题层级和长度约束切分候选 chunk 列表元数据附加候选 chunk 列表附加来源、标题路径、序号等最终 chunk 列表这个流程的好处是每一阶段职责单一出问题容易定位。比如检索效果差可以先检查切分结果再检查结构解析逐层排查。3. 核心细节解析与实操要点3.1 编码问题txt 解析的第一道坎txt 文件最大的坑就是编码。国内环境下很多老文件是 GBK 或 GB2312 编码直接按 UTF-8 读会报错或者读出乱码。更麻烦的是有些文件是混合编码前半段 GBK 后半段 UTF-8这种情况虽然少见但确实存在。我的处理策略是分级尝试。先用 UTF-8 读失败了再试 GBK再失败试 GB18030GBK 的超集覆盖面更广最后兜底用latin-1强行读进来保证不报错但可能乱码需要人工检查。def read_text_file(file_path): encodings [utf-8, gbk, gb18030, latin-1] for enc in encodings: try: with open(file_path, r, encodingenc) as f: content f.read() # 去除 BOM if content.startswith(\ufeff): content content[1:] return content, enc except (UnicodeDecodeError, LookupError): continue raise ValueError(f无法解码文件: {file_path})这里有个细节值得说BOM 处理。Windows 记事本保存的 UTF-8 文件经常带 BOM字节顺序标记表现为文件开头多一个不可见字符\ufeff。如果不处理这个字符会混进第一个 chunk影响 embedding 质量。所以读到内容后要主动检查并去除。注意不要用errorsignore来强行读取。虽然这样不会报错但会静默丢弃无法解码的字符导致内容缺失。宁可报错让人工介入也不要悄悄丢数据。3.2 Markdown 结构解析把标题层级变成树Markdown 的核心结构就是标题层级。#是一级##是二级以此类推。解析的目标是把线性的文本变成一棵树每个节点知道自己属于哪个章节。实现思路是逐行扫描 栈维护当前路径。遇到标题行就根据级别弹出栈中比它深的节点然后压入当前标题遇到普通内容行就归属到栈顶的标题下。import re HEADING_PATTERN re.compile(r^(#{1,6})\s(.*)$) def parse_markdown_structure(text): lines text.split(\n) nodes [] heading_stack [] # 存储 (level, title) current_content [] def flush_content(): if current_content: content \n.join(current_content).strip() if content: path .join([h[1] for h in heading_stack]) nodes.append({ type: content, path: path, text: content }) current_content.clear() for line in lines: match HEADING_PATTERN.match(line) if match: flush_content() level len(match.group(1)) title match.group(2).strip() while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, title)) else: current_content.append(line) flush_content() return nodes这段代码的关键在于heading_stack的维护。当遇到一个二级标题时如果栈顶是三级标题说明三级标题的章节结束了要弹出去如果栈顶是一级标题说明二级标题是一级标题的子节点直接压入即可。这样栈里始终保存着从根到当前节点的完整路径。3.3 代码块和表格的保护Markdown 里的代码块用三个反引号包裹表格用|分隔。这两类内容在切分时必须整体保留不能被切断。识别代码块的逻辑是状态机遇到就进入代码块模式再遇到就退出。在代码块模式内所有行都归属到同一个节点不参与普通的段落切分。def extract_blocks(text): lines text.split(\n) blocks [] i 0 while i len(lines): line lines[i] if line.strip().startswith(): # 代码块开始 lang line.strip()[3:].strip() code_lines [] i 1 while i len(lines) and not lines[i].strip().startswith(): code_lines.append(lines[i]) i 1 blocks.append({ type: code, lang: lang, text: \n.join(code_lines) }) i 1 # 跳过结束的 elif line.strip().startswith(|) and | in line.strip()[1:]: # 表格开始连续读取表格行 table_lines [] while i len(lines) and lines[i].strip().startswith(|): table_lines.append(lines[i]) i 1 blocks.append({ type: table, text: \n.join(table_lines) }) else: blocks.append({type: text, text: line}) i 1 return blocks表格的识别稍微 tricky 一点。Markdown 表格的每一行都以|开头和结尾中间用|分隔单元格。判断条件是行首是|且行内至少有两个|。连续的多行表格要合并成一个节点。实操心得代码块的语言标记python里的python要保留下来。检索时如果用户问的是代码相关问题语言标记能帮助排序。另外代码块在 embedding 时可以考虑单独处理因为代码的语义和自然语言差异很大。3.4 纯 txt 的结构推断txt 没有显式标记结构推断只能靠启发式规则。我常用的几条规则如下。空行分段连续两个换行符视为段落边界。这是最可靠的信号绝大多数 txt 都遵循这个约定。行首编号识别标题像第一章、1. 概述、一、、1这类行首模式很可能是标题。可以用正则匹配。缩进识别层级行首的空格或 Tab 数量可以反映层级关系。缩进多的行通常是缩进少的行的子内容。短行 空行前后如果一个短行比如少于 30 字前后都是空行它很可能是标题。TITLE_PATTERNS [ re.compile(r^第[一二三四五六七八九十百][章节部分]), re.compile(r^\d(\.\d)*\s\S), re.compile(r^[一二三四五六七八九十][、.]\s*\S), re.compile(r^[(]\d[)]\s*\S), ] def is_likely_title(line): stripped line.strip() if not stripped or len(stripped) 50: return False for pattern in TITLE_PATTERNS: if pattern.match(stripped): return True return False这些规则不可能 100% 准确但能覆盖大部分常见情况。实际项目中我会把推断结果和原始行号一起记录下来方便人工校验。3.5 切分策略按标题切还是按长度切有了结构树之后切分策略就有了依据。我的做法是优先按标题切标题内再按长度切。具体来说如果一个章节的内容总长度在阈值内比如 800 token就整体作为一个 chunk。如果超长就在章节内部按段落进一步切分切分时尽量保持段落完整。这里有个权衡标题层级用几级。如果按一级标题切chunk 会很大按三级标题切chunk 会很碎。我的经验是动态决定——先看二级标题下的内容量如果合适就用二级太大就下沉到三级。def split_by_structure(nodes, max_tokens800, min_tokens100): chunks [] buffer [] buffer_tokens 0 for node in nodes: node_tokens estimate_tokens(node[text]) # 代码块和表格不切分直接作为独立 chunk if node[type] in (code, table): if buffer: chunks.append(merge_buffer(buffer)) buffer [] buffer_tokens 0 chunks.append(node) continue # 如果加入当前节点会超长先 flush if buffer_tokens node_tokens max_tokens and buffer_tokens min_tokens: chunks.append(merge_buffer(buffer)) buffer [] buffer_tokens 0 buffer.append(node) buffer_tokens node_tokens if buffer: chunks.append(merge_buffer(buffer)) return chunksestimate_tokens是个估算函数。中文大致按 1 个字符 0.6 个 token 算英文按 4 个字符 1 个 token 算。精确计算需要调用 tokenizer但解析阶段用估算就够了没必要引入额外依赖。4. 实操过程与核心环节实现4.1 完整解析器的代码结构把前面的模块组装起来一个完整的解析器大概长这样。我按职责拆成几个类方便单独测试和替换。class TextLoader: 负责读取文件、处理编码 def load(self, file_path): # 前面实现的 read_text_file pass class MarkdownParser: 负责解析 Markdown 结构 def parse(self, text): # 返回结构化节点列表 pass class TxtParser: 负责推断 txt 结构 def parse(self, text): pass class Chunker: 负责按结构切分 def split(self, nodes, max_tokens800): pass class RAGImporter: 串联整个流程 def __init__(self): self.loader TextLoader() self.md_parser MarkdownParser() self.txt_parser TxtParser() self.chunker Chunker() def import_file(self, file_path): text, encoding self.loader.load(file_path) if file_path.endswith(.md): nodes self.md_parser.parse(text) else: nodes self.txt_parser.parse(text) chunks self.chunker.split(nodes) # 附加元数据 for i, chunk in enumerate(chunks): chunk[source] file_path chunk[encoding] encoding chunk[chunk_index] i return chunks这种分层设计的好处是每一层都可以独立替换。比如你后来想支持 PDF只需要新增一个PDFParser其他部分不用动。4.2 元数据设计chunk 应该带哪些信息元数据看起来是小事但对检索质量影响很大。我一般会附加以下几类信息。来源信息文件路径、文件名、文件修改时间。用于展示引用来源。位置信息chunk 在原文件中的起止行号、在章节树中的路径。用于定位和展示上下文。结构信息chunk 的类型正文、代码、表格、所属标题层级。用于检索时的过滤和加权。统计信息字符数、估算 token 数、段落数。用于监控和调优。chunk_metadata { source: docs/guide.md, file_name: guide.md, modified_time: 2024-01-15, start_line: 120, end_line: 145, heading_path: 第 2 章 2.1 节 配置说明, chunk_type: text, char_count: 680, token_estimate: 420, paragraph_count: 3 }这些元数据在检索阶段能派上大用场。比如用户问配置相关的说明可以优先检索heading_path里包含配置的 chunk用户问代码问题可以过滤chunk_typecode的 chunk。4.3 一个完整的解析示例拿一段真实的 Markdown 来走一遍流程。假设文件内容如下# 项目说明 这是一个示例项目。 ## 安装步骤 ### 环境要求 - Python 3.8 - 内存 4GB 以上 ### 安装命令 bash pip install example使用方式运行以下命令启动example start解析后的结构树大致是 | 节点类型 | 标题路径 | 内容 | |----------|----------|------| | content | 项目说明 | 这是一个示例项目。 | | content | 项目说明 安装步骤 环境要求 | - Python 3.8 - 内存 4GB 以上 | | code | 项目说明 安装步骤 安装命令 | pip install example | | content | 项目说明 使用方式 | 运行以下命令启动 | | code | 项目说明 使用方式 | example start | 切分时如果总长度不超阈值可能合并成 2 到 3 个 chunk。代码块作为独立 chunk 保留正文按标题路径合并。 ### 4.4 参数调优max_tokens 到底设多少 max_tokens 是最需要调的参数。设太小chunk 碎检索时上下文不足设太大chunk 长检索精度下降embedding 成本上升。 我的经验值是这样的。**技术文档**建议 500 到 800 token因为技术内容密度高一个完整概念通常在这个范围内。**叙述性文档**如小说、报告建议 300 到 500 token因为叙述内容语义相对独立短一点反而更精准。**代码文档**建议按函数或类切分不强求 token 数。 调参的方法是**先跑一批查询看检索结果**。如果发现检索出来的 chunk 经常答非所问说明 chunk 太大混入了无关内容如果发现答案不完整说明 chunk 太小关键信息被切走了。 实操心得不要一次性把所有文档都用同一个参数。可以按文档类型分组不同类型用不同参数。这个配置可以放在元数据里检索时按类型应用。 ## 5. 常见问题与排查技巧实录 ### 5.1 乱码问题速查 乱码是 txt 解析最常见的报错。下面这张表整理了典型症状和对应处理。 | 症状 | 可能原因 | 处理方法 | |------|----------|----------| | 中文显示为方块或问号 | 编码不匹配 | 尝试 GBK、GB18030 | | 开头多一个奇怪字符 | UTF-8 BOM | 读取后去除 \ufeff | | 部分内容正常部分乱码 | 混合编码 | 分段检测编码分别解码 | | 全部是 \x 转义 | 二进制文件误当文本 | 检查文件类型排除二进制 | | 换行符异常 | Windows/Mac 换行差异 | 统一替换 \r\n 和 \r 为 \n | 换行符统一这一步经常被忽略。Windows 的 \r\n、老 Mac 的 \r、Linux 的 \n如果不统一按 \n 切分时会出问题。建议读取后立即做一次 text.replace(\r\n, \n).replace(\r, \n)。 ### 5.2 切分结果不符合预期的排查思路 切分出问题通常表现为 chunk 太碎、chunk 太大、或者内容错位。排查时按以下顺序检查。 **第一步看结构解析结果**。把解析出来的节点列表打印出来看标题识别对不对、代码块有没有被正确识别。如果结构解析就错了后面切分肯定不对。 **第二步看切分边界**。把每个 chunk 的首尾几行打印出来看切分点是否落在合理的位置。理想情况下切分点应该在段落之间而不是句子中间。 **第三步看 token 估算**。如果 chunk 长度和预期差很多可能是 token 估算函数不准。中文和英文的 token 比例差异很大估算时要区分对待。 **第四步看特殊内容**。代码块、表格、公式这些内容有没有被正确处理。如果发现代码块被切碎检查状态机逻辑。 ### 5.3 标题识别失败的常见原因 Markdown 标题识别一般不会出错但 txt 的标题推断经常翻车。常见原因有几个。 **误判**把普通句子当成标题。比如这是一个测试这种短句如果前后有空行可能被误判为标题。解决办法是收紧规则要求标题必须匹配特定模式如编号开头。 **漏判**真正的标题没被识别。比如【重要】注意事项这种用方括号标记的标题如果规则里没有覆盖就会漏掉。解决办法是持续补充模式库。 **层级错乱**标题的层级关系判断错误。比如1.1和1.2应该是同级但如果规则只看数字个数可能判断错。解决办法是解析编号的层级结构而不是简单匹配。 python def parse_numbering_level(text): 解析编号层级如 1.2.3 返回 3 match re.match(r^(\d(?:\.\d)*), text.strip()) if match: return len(match.group(1).split(.)) return None5.4 性能优化大文件怎么处理处理大文件比如几十 MB 的 txt时一次性读入内存可能有问题。这时候需要流式处理。思路是逐行读取维护一个滑动窗口。窗口内的内容达到一定量就 flush 出去避免内存堆积。对于 Markdown还要维护标题栈的状态。def stream_parse(file_path, chunk_size10000): with open(file_path, r, encodingutf-8) as f: buffer [] for line in f: buffer.append(line) if len(buffer) chunk_size: yield .join(buffer) buffer [] if buffer: yield .join(buffer)流式处理的难点在于跨块的结构。比如一个代码块可能横跨两个流式块这时候需要维护状态把未闭合的代码块暂存起来等下一块来了再合并。实操心得对于超过 10MB 的文件建议先做一次预处理把它拆成多个小文件。这样既能流式处理又方便并行。拆分点选在章节边界保证每个小文件结构完整。5.5 检索效果差的归因方法如果 RAG 检索效果不好怎么判断是不是解析环节的问题我的方法是做对照实验。准备一组测试查询每个查询都有明确的标准答案。然后分别用解析后的 chunk和人工整理的 chunk跑检索对比命中率。如果人工整理的明显更好说明解析环节有问题。进一步定位可以看检索命中的 chunk 和标准答案的 chunk 是否一致。如果不一致看差异在哪里是切分点不对还是元数据缺失还是内容被污染。这样一步步缩小范围最终定位到具体的解析逻辑。6. 从 txt 到 Markdown 的通用化设计6.1 抽象出统一的文档模型txt 和 Markdown 虽然格式不同但解析后的目标是一致的一棵带元数据的结构树。所以可以抽象出一个统一的文档模型让后续的切分、embedding、入库逻辑不用关心原始格式。class DocumentNode: def __init__(self, node_type, text, heading_path, metadataNone): self.node_type node_type # text, code, table, list self.text text self.heading_path heading_path self.metadata metadata or {} def to_dict(self): return { type: self.node_type, text: self.text, heading_path: self.heading_path, **self.metadata }有了这个统一模型Markdown 解析器和 txt 解析器都输出DocumentNode列表后面的流程完全复用。将来要支持 HTML、PDF、Word也只需要新增对应的解析器。6.2 格式检测与自动路由用户上传文件时不一定能保证扩展名正确。有的 txt 文件其实是 Markdown有的 Markdown 文件扩展名是 txt。所以需要内容检测。检测逻辑很简单看文件里有没有 Markdown 的特征标记。如果出现#开头的行、代码块、|表格就按 Markdown 处理否则按纯文本处理。def detect_format(text): lines text.split(\n)[:100] # 只看前 100 行 md_signals 0 for line in lines: if re.match(r^#{1,6}\s, line): md_signals 2 if line.strip().startswith(): md_signals 2 if line.strip().startswith(|) and line.count(|) 2: md_signals 1 return markdown if md_signals 3 else text阈值设为 3 是个经验值。单个#可能是注释单个|可能是普通字符但多个信号同时出现基本可以确定是 Markdown。6.3 扩展性考虑为后续格式预留接口虽然这篇只讲 txt 和 Markdown但设计时要为后续格式留好接口。核心是解析器注册机制。class ParserRegistry: def __init__(self): self.parsers {} def register(self, format_name, parser): self.parsers[format_name] parser def get_parser(self, format_name): return self.parsers.get(format_name) def parse(self, text, format_name): parser self.get_parser(format_name) if not parser: raise ValueError(f不支持的格式: {format_name}) return parser.parse(text) registry ParserRegistry() registry.register(markdown, MarkdownParser()) registry.register(text, TxtParser())这样新增格式时只需要实现一个parse方法然后注册进去主流程完全不用改。这种设计在项目初期可能显得过度但当格式从 2 种增加到 5 种时收益就体现出来了。6.4 测试策略怎么保证解析质量解析逻辑的测试不能只靠跑一遍看看需要构造覆盖各种边界情况的测试用例。我一般会准备这几类测试文件。正常文件结构清晰的 Markdown 和 txt验证基本功能。边界文件空文件、只有标题没有内容、超长单行、嵌套代码块。异常文件编码混乱、混合换行符、BOM 开头。特殊内容包含表格、公式、图片链接、HTML 标签的 Markdown。每个测试文件都配上期望的输出用断言验证解析结果。这样每次改代码跑一遍测试就知道有没有引入回归。def test_markdown_heading_path(): text # A\n\n## B\n\n内容 parser MarkdownParser() nodes parser.parse(text) content_node [n for n in nodes if n[type] content][0] assert content_node[path] A B测试用例不用多但要精。覆盖了主要分支和边界情况就能挡住大部分低级错误。7. 一些踩坑之后的个人体会做 RAG 数据导入这两年最大的体会是解析环节的投入产出比被严重低估了。很多团队愿意花大价钱买向量数据库、调模型参数却在解析上随便找个现成工具糊弄过去。结果就是检索效果怎么调都上不去最后发现根子在数据质量上。我的建议是在项目早期就把解析逻辑做扎实。具体来说先花时间把文档格式摸清楚统计一下有多少种格式、多少种编码、多少种结构。然后针对性地写解析规则而不是指望一个通用工具解决所有问题。另一个体会是元数据的价值被低估了。很多人只关注 chunk 的文本内容忽略了标题路径、来源、类型这些元数据。实际上这些元数据在检索时能提供额外的过滤和排序信号效果提升很明显。比如用户问安装相关的问题如果 chunk 带有标题路径就能优先检索标题里含安装的 chunk。最后说一个具体的技巧保留原始行号。解析时记录每个 chunk 对应原文件的起止行号看起来没什么用但当用户反馈这段内容不对时能快速定位到原文件的具体位置排查效率高很多。这个信息在展示引用来源时也能派上用场。解析这件事做一遍不难做好很难。但只要把基础打牢后面的检索、生成环节都会顺畅很多。