RAG 系统落地时最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。很多人一上来就搭向量库、调 embedding 模型结果灌进去的文本要么格式乱成一锅粥要么结构信息全丢检索出来的内容驴唇不对马嘴。我做过好几个知识库项目踩过的坑基本都集中在数据进库之前这一段。这篇就从最基础的 txt 和 Markdown 两种格式切入把通用文本和结构化文本的导入解析逻辑讲透后面再展开 PDF、HTML、表格类数据的处理。1. 为什么 txt 和 Markdown 是 RAG 数据管道的起点1.1 两种格式在 RAG 里的真实定位txt 是纯文本的底线格式没有任何标记语言所有内容都是平铺的字符流。Markdown 则是在纯文本基础上加了一层轻量标记用#、-、**这些符号表达层级、列表和强调。这两者放在 RAG 场景里角色完全不同。txt 通常来自日志导出、爬虫抓取、旧系统迁移特点是能用但没结构。Markdown 则大量出现在技术文档、Wiki 导出、GitHub 仓库里天然带层级信息。我处理过的知识库项目里Markdown 占比往往超过一半因为技术团队写文档默认就用它。关键区别在于txt 解析的核心任务是切分和清洗Markdown 解析的核心任务是保留结构的同时切分。前者丢了结构不可惜后者丢了结构等于把文档的骨架拆了。1.2 通用文本与结构化文本的分野所谓通用文本指的是没有明确层级标记、段落靠空行或换行区分的内容。结构化文本则有显式的标题层级、列表、代码块、表格等元素。这个分野直接决定了 chunk 策略。我见过太多人用同一套RecursiveCharacterTextSplitter处理所有格式结果 Markdown 里的代码块被从中间切断标题和正文被分到不同 chunk检索时模型拿到半截代码完全无法理解。这不是 splitter 的问题是没做格式识别。提示在数据导入管道的最前端就应该做格式分流而不是等到切分阶段才处理。格式识别做在前面后面每一步都能针对性地优化。1.3 一个容易被忽略的事实编码问题txt 文件的编码是 RAG 导入的第一道暗坑。GBK、GB2312、UTF-8、UTF-8 with BOM、Latin-1这些编码混在一起读出来就是乱码。我遇到过一批从旧 Windows 系统导出的 txt默认 GBK 编码用 UTF-8 读出来全是问号向量化之后检索结果惨不忍睹。处理原则很简单先探测编码再统一转成 UTF-8。Python 里用chardet或charset-normalizer做探测准确率在 95% 以上。剩下 5% 的疑难文件手动指定编码兜底。import chardet def detect_encoding(file_path): with open(file_path, rb) as f: raw f.read(10000) result chardet.detect(raw) return result[encoding], result[confidence]置信度低于 0.7 的时候不要盲目相信探测结果最好人工抽检几个文件确认。这个步骤看起来繁琐但能避免后面整条管道产出垃圾数据。2. txt 文件解析从原始字节到可切分文本2.1 读取阶段的编码归一化流程txt 读取不是一句open().read()就完事。我的标准流程是三步探测编码、尝试解码、失败则回退。def read_txt_safely(file_path): encodings [utf-8, gbk, gb2312, latin-1] for enc in encodings: try: with open(file_path, r, encodingenc) as f: return f.read(), enc except (UnicodeDecodeError, LookupError): continue # 最后用 errorsreplace 兜底 with open(file_path, r, encodingutf-8, errorsreplace) as f: return f.read(), utf-8-replace这个顺序有讲究。UTF-8 优先是因为现代系统默认用它GBK 和 GB2312 覆盖中文旧系统Latin-1 是万能兜底因为它能解码任何字节序列虽然可能产生乱码但不会抛异常。errorsreplace是最后手段会把无法解码的字节替换成占位符至少保证流程不中断。实测下来这套流程能覆盖 98% 以上的 txt 文件。剩下 2% 通常是文件本身损坏或者混合编码需要单独处理。2.2 清洗环节哪些字符必须去掉txt 清洗的目标是去掉对语义无贡献的噪声。常见的噪声包括连续空行超过两个换行符行首行尾的空白字符不可见控制字符\x00到\x1f重复的分隔线如、-----页码标记、页眉页脚残留但清洗要有度。我见过有人把所有制表符和多余空格都干掉结果表格类 txt 的列对齐全乱了本来能看出结构的内容变成一坨。清洗的原则是只去掉确定无意义的字符保留可能承载结构的空白。import re def clean_txt(text): # 去掉控制字符保留换行和制表符 text re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , text) # 连续空行压缩为两个换行 text re.sub(r\n{3,}, \n\n, text) # 去掉行首行尾空白 lines [line.strip() for line in text.split(\n)] return \n.join(lines)注意这里保留了制表符\t因为它在某些 txt 里是列分隔符。如果确认文件里没有表格结构再去掉也不迟。2.3 切分策略按语义边界而非固定长度txt 切分最忌讳的就是固定字符数硬切。500 字一刀切下去句子断在中间检索出来的 chunk 读起来莫名其妙。正确的做法是优先按语义边界切分。我常用的优先级是段落边界\n\n 句子边界。 逗号 固定长度。LangChain 的RecursiveCharacterTextSplitter就是按这个思路设计的但默认分隔符对中文不友好需要自定义。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , , ], length_functionlen, )chunk_overlap设 50 是为了让相邻 chunk 有上下文重叠避免边界处的信息丢失。这个值不是越大越好太大会导致检索结果重复。我的经验是 chunk_size 的 10% 到 15% 比较合适。注意中文的length_function用len是按字符数算的一个中文字符算一个。如果你的 embedding 模型对 token 数敏感需要换成 token 计数函数。2.4 元数据附加让每个 chunk 可追溯切分完的 chunk 不能是裸文本必须带上元数据。最基本的元数据包括来源文件名、chunk 序号、原始位置偏移。这些信息在检索结果展示和问题排查时非常关键。chunks splitter.split_text(text) for i, chunk in enumerate(chunks): metadata { source: file_path, chunk_index: i, total_chunks: len(chunks), format: txt }我还会额外记录 chunk 的字符长度和是否包含特殊字符。长度异常的 chunk比如超过 2000 字或少于 20 字往往是切分出了问题需要回头检查。3. Markdown 解析保住结构才能保住语义3.1 Markdown 的 AST 解析思路Markdown 不是纯文本它有语法结构。直接当 txt 处理等于把标题、列表、代码块的语义全扔了。正确做法是先解析成 AST抽象语法树再基于 AST 做切分。Python 里常用的 Markdown 解析库有markdown-it-py、mistune、markdown。我推荐markdown-it-py因为它严格遵循 CommonMark 规范AST 结构清晰扩展性好。from markdown_it import MarkdownIt md MarkdownIt() tokens md.parse(markdown_text)解析出来的 tokens 是一个扁平列表每个 token 有type、tag、content、level等属性。标题 token 的type是heading_open后面跟着inline和heading_close。列表、代码块、表格都有对应的 token 类型。基于 AST 切分的好处是你能准确知道每个内容块属于哪个标题层级切分时可以把标题作为上下文附加到 chunk 里。3.2 标题层级与 chunk 的绑定关系Markdown 的标题层级是天然的语义分界。一个##标题下的内容语义上属于同一个主题。切分时应该以标题为锚点把标题路径附加到每个 chunk 的元数据里。比如这样一段 Markdown## 安装指南 ### 环境要求 需要 Python 3.8 以上版本。 ### 安装步骤 执行 pip install xxx。切分后需要 Python 3.8 以上版本。这个 chunk 的元数据应该包含heading_path: [安装指南, 环境要求]。检索时如果命中这个 chunk模型能知道它属于哪个章节回答时更准确。def extract_heading_path(tokens, index): path [] for i in range(index, -1, -1): if tokens[i].type heading_open: level int(tokens[i].tag[1]) content tokens[i1].content path.insert(0, (level, content)) # 按 level 排序构建层级路径 return [c for _, c in sorted(path)]这个逻辑需要遍历 token 列表找到当前内容之前的所有标题。实际实现时可以用栈来维护当前标题路径效率更高。3.3 代码块、表格、列表的特殊处理Markdown 里的代码块、表格、列表是结构最脆弱的部分切分时最容易出问题。代码块必须整体保留。一个代码块被从中间切断前后两半都失去意义。处理原则是代码块作为一个不可分割的单元如果超过 chunk_size要么整体保留允许超长要么按代码逻辑切分比如按函数边界。表格同理切断了列对齐就没了。表格要么整体保留要么按行切分但保留表头。列表的处理稍微灵活一些。短列表整体保留长列表可以按项切分但每个 chunk 要带上列表的上下文比如列表前的引导句。def is_code_block(token): return token.type in (fence, code_block) def is_table(token): return token.type table_open在切分循环里遇到这些 token就切换到特殊处理逻辑不走常规的按长度切分。3.4 数学公式与特殊语法的兼容Markdown 里的数学公式$...$和$$...$$是另一个坑。标准 Markdown 解析器不认这些语法会把它们当普通文本处理。如果你的知识库涉及技术文档公式处理不好会严重影响检索质量。处理方案有两种一是用支持数学公式的解析器如markdown-it-py配合mdit-py-plugins的dollarmath插件二是预处理阶段把公式提取出来单独处理。from mdit_py_plugins.dollarmath import dollarmath_plugin md MarkdownIt().use(dollarmath_plugin)公式在 RAG 里的处理比较特殊因为 embedding 模型对公式的语义理解有限。我的做法是把公式转成 LaTeX 文本保留同时在元数据里标记这是公式内容检索时可以针对性处理。4. 通用切分器与结构化切分器的选型对比4.1 两种切分器的适用边界通用切分器如RecursiveCharacterTextSplitter适合纯文本、无结构或结构不规则的内容。它的优势是简单、鲁棒对任何文本都能产出合理结果。劣势是丢失结构信息无法感知标题、代码块等元素。结构化切分器如MarkdownHeaderTextSplitter适合有明确层级标记的内容。它能保留标题路径、代码块完整性但前提是输入必须是规范的 Markdown。如果 Markdown 语法不规范比如标题层级跳跃、代码块未闭合解析会出问题。我的选型原则是格式明确用结构化切分器格式混乱用通用切分器混合内容先分流再分别处理。对比维度通用切分器结构化切分器输入要求任意文本规范 Markdown结构保留无标题路径、代码块切分粒度字符数控制语义单元控制鲁棒性高中实现复杂度低中高适用场景txt、日志、爬虫文本技术文档、Wiki4.2 混合管道的设计思路实际项目里纯 txt 或纯 Markdown 的情况很少更多是混合。我的做法是在管道入口做格式识别然后分流到不同的处理分支。def detect_format(file_path, content): if file_path.endswith(.md) or file_path.endswith(.markdown): return markdown # 内容里如果有大量 Markdown 标记也判定为 markdown md_markers [## , ### , , - [, **] marker_count sum(content.count(m) for m in md_markers) if marker_count 5: return markdown return txt这个识别逻辑不完美但覆盖了大部分情况。识别错误的文件会在后续处理中暴露问题比如 Markdown 解析器报错可以加日志监控。4.3 切分参数的调优经验chunk_size 和 chunk_overlap 这两个参数没有万能值需要根据内容特点和检索需求调。我的经验值技术文档chunk_size 400-600overlap 50-80小说、长文chunk_size 800-1200overlap 100-150问答对chunk_size 200-400overlap 20-50代码文档chunk_size 按函数长度overlap 按需调参的验证方法是切分后抽样检查 chunk 的完整性看是否有断句、断代码、断表格的情况。同时用几个典型 query 做检索测试看召回的内容是否完整。提示chunk_size 不是越大越好。太大的 chunk 会稀释语义检索时匹配度下降。太小的 chunk 会丢失上下文模型理解困难。找到平衡点是关键。5. 导入管道的工程化落地5.1 批量处理的并发与限流数据量大时单线程处理太慢。我的做法是用concurrent.futures做并发但要注意限流避免把磁盘 IO 打满。from concurrent.futures import ThreadPoolExecutor, as_completed def process_files(file_paths, max_workers4): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(process_single, fp): fp for fp in file_paths} for future in as_completed(futures): fp futures[future] try: results.append(future.result()) except Exception as e: print(f处理失败: {fp}, 错误: {e}) return resultsmax_workers设 4 是个保守值机械硬盘上再高会因寻道竞争反而变慢。SSD 上可以设到 8 或 16。关键是监控 IO 等待时间如果 IO wait 超过 30%说明并发过高了。5.2 失败重试与断点续传批量处理必然有失败的文件。我的管道里每个文件处理完会写一条状态记录失败的文件记录下来支持单独重试。import json import os def save_progress(progress_file, file_path, status, errorNone): record {file: file_path, status: status, error: error} with open(progress_file, a, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n)断点续传的逻辑是启动时读取进度文件跳过已成功的文件。这个机制在数据量大、处理时间长的时候特别有用避免中途失败后从头再来。5.3 质量校验怎么知道解析结果是对的解析完不能直接灌库要先做质量校验。我的校验清单包括chunk 数量是否合理和文件大小成正比是否有空 chunk 或超短 chunk是否有超长 chunk超过 chunk_size 的 2 倍编码是否正确抽样检查是否有乱码元数据是否完整def validate_chunks(chunks, chunk_size): issues [] for i, chunk in enumerate(chunks): if len(chunk) 20: issues.append(fchunk {i} 过短: {len(chunk)} 字符) if len(chunk) chunk_size * 2: issues.append(fchunk {i} 过长: {len(chunk)} 字符) if not chunk.strip(): issues.append(fchunk {i} 为空) return issues校验发现的问题要分类处理空 chunk 直接丢弃超短 chunk 合并到相邻 chunk超长 chunk 重新切分。这个环节能拦住 80% 的解析质量问题。5.4 与向量库对接的注意事项解析完的 chunk 要灌入向量库。对接时有两个坑一是 embedding 的批量大小二是元数据的存储格式。embedding 批量大小取决于模型和硬件。OpenAI 的 embedding API 单次最多 2048 个输入但实际用的时候建议每批 100-500 个避免单次请求过大导致超时。本地模型则看显存一般 32-128 一批。元数据存储要注意向量库的限制。有些向量库对元数据的字段类型和长度有限制比如 Pinecone 的 metadata 单个值不能超过 40KB。超长的元数据比如完整的标题路径需要截断或哈希处理。def prepare_for_upsert(chunks, metadatas, batch_size100): for i in range(0, len(chunks), batch_size): batch_chunks chunks[i:ibatch_size] batch_metas metadatas[i:ibatch_size] yield batch_chunks, batch_metas分批灌入的时候要记录进度避免中途失败后重复灌入。向量库一般支持 upsert存在则更新用这个语义可以安全重试。6. 实操中踩过的坑与排查链路6.1 乱码问题的完整排查过程有一次导入一批 txt检索结果里频繁出现乱码。排查链路是这样的第一步确认乱码出现在哪个环节。从向量库捞回原始 chunk发现乱码在 chunk 里说明是导入阶段的问题不是检索或展示的问题。第二步检查原始文件。用十六进制编辑器打开发现文件头有 BOM 标记但编码探测返回的是 UTF-8。BOM 是\xef\xbb\xbfUTF-8 解码后会在文本开头产生一个不可见字符导致后续处理异常。第三步修复。读取时用utf-8-sig编码它会自动去掉 BOM。with open(file_path, r, encodingutf-8-sig) as f: content f.read()这个坑的教训是编码探测不能只看返回的编码名还要检查 BOM。utf-8和utf-8-sig的区别就在 BOM 处理上。6.2 Markdown 标题层级跳跃导致的切分错乱另一个坑是 Markdown 标题层级不规范。有的文档从#直接跳到###中间没有##。用MarkdownHeaderTextSplitter处理时标题路径构建会出错chunk 的元数据里标题路径缺失或错位。排查方法是打印每个 chunk 的heading_path看是否符合预期。修复方案是在解析前做一次标题层级规范化把跳跃的层级补齐。def normalize_heading_levels(tokens): prev_level 0 for token in tokens: if token.type heading_open: level int(token.tag[1]) if level prev_level 1: # 层级跳跃需要补齐 token.tag fh{prev_level 1} prev_level int(token.tag[1]) return tokens这个规范化逻辑比较粗暴实际用的时候要结合文档结构判断。更稳妥的做法是在文档编写规范里要求标题层级连续从源头避免问题。6.3 代码块未闭合引发的解析异常Markdown 代码块用三个反引号包裹如果文档里少了一个反引号解析器会把后面的所有内容都当成代码。这种问题在人工编写的文档里很常见。排查方法是统计代码块标记的数量奇数说明有未闭合的。修复方案是自动补全或人工修正。def check_code_blocks(content): fence_count content.count() if fence_count % 2 ! 0: return False, f代码块标记数量为奇数: {fence_count} return True, OK这个检查应该放在解析之前作为数据质量校验的一部分。发现问题的文件标记出来人工确认后再处理。6.4 超长单行文本的切分困境有些 txt 文件是一整行超长文本没有任何换行。这种文件用常规切分器处理会因为找不到分隔符而退化成按字符硬切效果很差。处理方案是先做一次软换行预处理在句子边界插入换行符。def soft_wrap(text, max_line_length200): sentences re.split(r([。]), text) lines [] current for i in range(0, len(sentences) - 1, 2): sentence sentences[i] sentences[i1] if len(current) len(sentence) max_line_length: lines.append(current) current sentence else: current sentence if current: lines.append(current) return \n.join(lines)这个预处理能把超长行拆成合理长度的段落后续切分就正常了。max_line_length设 200 是个经验值可以根据内容调整。7. 从解析到入库的完整链路回顾7.1 一个可复用的管道骨架把前面的内容串起来一个完整的 txt/Markdown 导入管道包含这些阶段格式识别判断是 txt 还是 Markdown编码归一化统一转成 UTF-8内容清洗去掉噪声字符结构解析Markdown 走 ASTtxt 走纯文本切分按格式选择切分器元数据附加来源、序号、标题路径质量校验检查 chunk 合理性批量入库分批 embedding 和 upsert每个阶段都可以独立测试和优化。我的习惯是每个阶段都输出中间结果方便排查问题。7.2 性能与质量的平衡点管道设计要在性能和质量之间找平衡。全量做 AST 解析和精细切分质量高但慢简单按字符切分快但质量差。我的做法是分级处理核心文档高频检索的走精细管道边缘文档走快速管道。分级标准可以是文档的访问频率、重要性标签或者简单的文件大小阈值。def choose_pipeline(file_path, file_size): if file_size 10 * 1024 * 1024: # 大于 10MB return fast if core in file_path: return fine return standard这个策略在实际项目里能显著降低处理时间同时保证核心内容的质量。7.3 后续扩展方向txt 和 Markdown 只是起点。实际 RAG 项目里还会遇到 PDF、Word、HTML、Excel、PPT 等格式。每种格式的解析逻辑都不同但管道的骨架是通用的识别、归一化、解析、切分、校验、入库。PDF 的难点是版面分析和表格提取HTML 的难点是正文抽取和噪声过滤Excel 的难点是表格结构识别和多 sheet 处理。这些我会在后续的文章里逐个展开。现在你手上有一套能跑的 txt/Markdown 导入管道了。建议先拿一批真实数据跑一遍重点看质量校验的输出把问题文件挑出来单独处理。管道跑通之后再考虑扩展格式和优化性能。数据导入这件事慢就是快前期多花时间把数据质量做扎实后面检索和生成环节能省大量调试时间。