RAG 系统落地时最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。我见过太多团队把 80% 的精力砸在调 prompt 和换 embedding 模型上结果上线后发现召回质量上不去回头一查——原始文档压根没解析干净。PDF 里的表格被拍成了一坨乱码Markdown 的层级结构全丢了txt 文件里混着各种不可见字符。检索再强喂进去的是垃圾出来的也只能是垃圾。这篇内容聚焦 RAG 数据管道的第一段路从最朴素的 txt 纯文本到带结构的 Markdown怎么把它们干净、完整、可追溯地导入知识库。适合正在搭建 RAG 知识库的工程师、做数据治理的从业者以及任何需要把杂乱文档变成结构化知识的人。我会把解析策略、分块逻辑、元数据设计、踩坑经验都摊开讲代码能直接抄思路能直接搬。1. 为什么 txt 和 Markdown 值得单独拎出来讲1.1 纯文本不等于简单文本很多人觉得 txt 是最简单的格式直接 read 进来就完事了。实际做过 RAG 数据导入的人都知道txt 的坑一点都不比 PDF 少。编码问题首当其冲——GBK、GB2312、UTF-8、UTF-8 with BOM甚至同一批文件里混着多种编码。你用默认的 UTF-8 去读一个 GBK 编码的文件轻则乱码重则直接抛异常中断整个导入流程。除了编码txt 里还藏着大量不可见字符。从 Windows 环境拷过来的文件经常带着\r\n从网页复制的内容可能混着零宽空格\u200b、不间断空格\u00a0、软连字符\u00ad。这些字符在肉眼看来毫无异常但在分词和向量化时会变成噪音直接影响检索命中率。我做过一个对比测试同一份文档清理前后Top-5 召回率差了将近 12 个百分点。还有一个容易被忽略的点txt 文件往往没有明确的段落边界。一篇几千字的文章可能就是一整行或者用空行分隔但空行数量不统一。如果不做预处理直接按固定长度切分很容易把一句话拦腰截断导致语义碎片化。1.2 Markdown 的结构价值被严重浪费Markdown 在 RAG 场景里其实是一种非常理想的源格式因为它自带层级结构——标题、列表、代码块、表格、引用这些语义信息如果保留下来对检索和生成都有巨大帮助。但现实是大部分人的处理方式就是把它当纯文本读进来然后按 token 数硬切。标题层级没了代码块被切散了表格变成了一堆竖线和横线。Markdown 的标题层级天然就是很好的分块依据。一个二级标题下的内容通常是一个完整的语义单元围绕同一个主题展开。按标题层级来切分比按固定字符数切分要合理得多。而且标题本身可以作为这个块的上下文信息附加进去检索时能显著提升相关性判断的准确度。代码块的处理也是同理。技术文档里的代码示例如果被切断检索出来的内容就是残缺的用户看到一半的代码根本没法用。识别出代码块的边界并保持其完整性是 Markdown 解析的基本要求。1.3 两种格式在 RAG 管道中的定位差异txt 和 Markdown 在 RAG 管道里扮演的角色不太一样。txt 更像是兜底格式——很多系统导出、日志、爬取内容最终都会落到 txt。它的优势是通用性强任何工具都能读劣势是结构信息几乎为零需要靠后处理来补。Markdown 则更像是高质量源格式。技术文档、Wiki、README、笔记类内容越来越多地采用 Markdown 存储。它的结构信息丰富解析得当的话可以直接映射到知识库的层级结构上。理解这个差异很重要因为它决定了你对两种格式的解析策略应该有所不同。txt 的重点在于清洗和边界重建Markdown 的重点在于结构提取和语义保留。下面我会分别展开讲。2. txt 文件导入的清洗与分块实战2.1 编码检测别再用默认编码硬读了处理 txt 的第一步永远是确认编码。我的做法是用chardet或charset-normalizer先做检测再根据检测结果读取。但检测不是百分百准确的尤其是短文本。所以还需要一个兜底策略按检测结果读如果失败就依次尝试常见编码。import chardet def detect_and_read(file_path): with open(file_path, rb) as f: raw f.read() result chardet.detect(raw) encoding result[encoding] confidence result[confidence] # 置信度低时按优先级尝试常见编码 if confidence 0.7: for enc in [utf-8, gbk, gb2312, latin-1]: try: return raw.decode(enc), enc except (UnicodeDecodeError, LookupError): continue return raw.decode(encoding), encoding这里有个经验latin-1永远不会抛异常因为它能解码任何字节序列。所以它只能作为最后的选择而且用完之后要检查内容是否合理比如有没有大量不可打印字符。另一个实用技巧是处理 BOM。UTF-8 with BOM 的文件开头会有\ufeff这个字符如果不清掉会附着在第一个词上导致检索时匹配不上。读取后统一用text.lstrip(\ufeff)处理一下就行。2.2 不可见字符清理看不见的才是最难缠的编码问题解决后下一步是清理不可见字符。我整理了一份常见问题字符清单和对应的处理方式字符Unicode 码点来源处理方式零宽空格U200B网页复制直接删除零宽非连接符U200C排版软件直接删除零宽连接符U200D排版软件直接删除不间断空格U00A0网页/Word替换为普通空格软连字符U00AD排版软件直接删除字节顺序标记UFEFFWindows 编辑器删除非首位置全角空格U3000中文排版替换为普通空格清理逻辑用正则一次性处理import re def clean_invisible_chars(text): # 删除零宽字符和软连字符 text re.sub(r[\u200b\u200c\u200d\u00ad\ufeff], , text) # 不间断空格和全角空格替换为普通空格 text re.sub(r[\u00a0\u3000], , text) # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 合并连续空行保留最多两个 text re.sub(r\n{3,}, \n\n, text) return text注意不要用strip()去处理每一行那样会把有意义的缩进也干掉。如果原文是代码或需要保留缩进的内容缩进信息很重要。2.3 段落边界重建从一坨到有结构清理完字符后txt 面临的核心问题是段落边界不清晰。我的策略是分三步走第一步识别显式分隔符。连续两个及以上的换行通常意味着段落分隔。单换行可能是段落内的换行也可能是段落分隔需要结合上下文判断。第二步用标点符号辅助判断。中文的句号、问号、感叹号英文的句点、问号、感叹号如果出现在行尾大概率是段落或句子边界。如果一行以这些标点结尾下一行以大写字母或中文字符开头中间大概率应该有一个段落分隔。第三步对超长段落做二次切分。有些 txt 文件整篇就是一段没有任何换行。这时候需要按句子切分再按语义聚合。句子切分可以用正则做粗切中文按[。]英文按[.!?]加空格。def rebuild_paragraphs(text): # 先按双换行切分 blocks re.split(r\n\s*\n, text) paragraphs [] for block in blocks: lines block.strip().split(\n) # 如果块内有多行判断是否应该合并 merged [] buffer for line in lines: line line.strip() if not line: continue if buffer and should_merge(buffer, line): buffer line else: if buffer: merged.append(buffer) buffer line if buffer: merged.append(buffer) paragraphs.extend(merged) return paragraphs def should_merge(prev, curr): # 前一行以句末标点结尾不合并 if re.search(r[。.!?]$, prev): return False # 前一行以逗号、分号等结尾合并 if re.search(r[、,;]$, prev): return True # 当前行以列表标记开头不合并 if re.match(r^[\-\*\d][\.\)、], curr): return False # 默认合并短行 return len(prev) 40这套逻辑不是万能的但实测下来对大多数中文和英文 txt 都能得到可用的段落结构。关键是should_merge这个判断函数你可以根据自己的数据特点调整规则。2.4 分块策略固定长度是下策段落重建完成后就进入分块环节。很多人直接用 LangChain 的RecursiveCharacterTextSplitter设个 chunk_size500、overlap50 就完事了。这样做不是不行但效果往往不够好。我的做法是优先按段落分块段落太长再按句子切分最后才考虑按字符数硬切。具体逻辑def chunk_by_paragraphs(paragraphs, max_chunk_size800, overlap_sentences1): chunks [] current_chunk [] current_size 0 for para in paragraphs: para_size len(para) if current_size para_size max_chunk_size and current_chunk: chunks.append(\n\n.join(current_chunk)) # 保留最后一句作为重叠 if overlap_sentences 0: last current_chunk[-1] sentences re.split(r(?[。.!?]), last) overlap_text .join(sentences[-overlap_sentences:]) current_chunk [overlap_text] current_size len(overlap_text) else: current_chunk [] current_size 0 current_chunk.append(para) current_size para_size if current_chunk: chunks.append(\n\n.join(current_chunk)) return chunks这里max_chunk_size设 800 是基于中文的平均信息密度来定的。中文一个字符承载的信息量比英文单词大800 个中文字符大约对应 500-600 个 token对大多数 embedding 模型来说都在合理范围内。如果你的 embedding 模型支持更长上下文可以适当放宽。重叠部分我选择按句子保留而不是按字符保留这样不会出现半句话的情况。重叠的目的是保持上下文连贯性让相邻块之间有语义衔接按句子重叠比按字符重叠更自然。3. Markdown 解析把结构信息榨干3.1 标题层级就是天然的分块边界Markdown 的标题层级是分块的最佳依据。一个二级标题下的内容通常是一个完整的主题三级标题是子主题。按这个层级来组织知识库检索时可以先定位到相关章节再在章节内做细粒度匹配。解析 Markdown 我推荐用markdown-it-py或mistune它们能把 Markdown 解析成 AST抽象语法树比正则匹配靠谱得多。用正则去解析 Markdown 是自找麻烦嵌套列表、代码块里的井号、转义字符这些情况正则很难处理干净。from markdown_it import MarkdownIt def parse_markdown_structure(text): md MarkdownIt() tokens md.parse(text) sections [] current_section {level: 0, title: , content: []} for token in tokens: if token.type heading_open: level int(token.tag[1]) # 保存上一个 section if current_section[content]: sections.append(current_section) current_section {level: level, title: , content: []} elif token.type inline and current_section[title] : current_section[title] token.content elif token.type in (paragraph_open, fence, table_open): pass elif token.type inline and current_section[title] ! : current_section[content].append(token.content) if current_section[content]: sections.append(current_section) return sections这段代码把 Markdown 拆成了带层级和标题的 section 列表。每个 section 的content是该标题下的正文内容。后续分块时如果 section 内容不长就整体作为一个 chunk如果太长再在 section 内部按段落切分。3.2 代码块和表格必须保持完整代码块和表格是 Markdown 里最容易被切散的结构。一个代码块被从中间切开检索出来的代码就是残缺的用户根本没法用。表格被切散后表头和表体分离信息完全丢失。处理原则很简单代码块和表格作为原子单元不参与常规分块。如果代码块本身超过 chunk 大小限制要么单独成块要么在代码块内部的逻辑边界如函数定义之间切分但必须保留完整的函数体。def extract_code_blocks(tokens): code_blocks [] in_fence False current_code [] lang for token in tokens: if token.type fence: code_blocks.append({ lang: token.info.strip(), code: token.content }) return code_blocks表格的处理类似用table_open和table_close标记边界整个表格作为一个单元。如果表格特别大可以按行拆分但每一块都要带上表头。3.3 列表和引用的语义保留列表在 Markdown 里表示并列或层级关系这个信息在检索时很有用。有序列表的序号、无序列表的层级缩进都是语义信号。解析时应该把列表项的层级关系保留下来而不是拍平成一堆独立的行。引用的处理稍微不同。引用块通常表示强调、补充说明或引用他人观点在 RAG 场景里可以标记为特殊类型检索时根据查询意图决定是否优先返回。我的做法是在 chunk 的元数据里记录内容类型content_type: paragraph | list | code | table | quote。检索时可以按类型过滤比如用户问代码相关的问题就优先检索code类型的块。3.4 数学公式和特殊语法的处理Markdown 里的数学公式$...$和$$...$$需要特别处理。行内公式如果被切散会变成一堆无意义的符号。块级公式应该作为独立单元保留。def protect_math(text): # 先把块级公式替换为占位符 math_blocks [] def replace_block(match): math_blocks.append(match.group(0)) return f__MATH_BLOCK_{len(math_blocks)-1}__ text re.sub(r\$\$(.?)\$\$, replace_block, text, flagsre.DOTALL) return text, math_blocks def restore_math(text, math_blocks): for i, block in enumerate(math_blocks): text text.replace(f__MATH_BLOCK_{i}__, block) return text分块前先保护公式分块后再还原。这样公式就不会被切散。行内公式类似处理但要注意不要误匹配货币符号可以用更严格的正则比如要求$前后不是数字。4. 元数据设计让每个块都可追溯4.1 必备的元数据字段每个 chunk 都应该携带足够的元数据方便检索时过滤和排序也方便出问题时追溯。我建议至少包含以下字段字段名类型说明source_filestring原始文件路径或标识file_typestringtxt / markdownchunk_indexint在文件内的序号char_startint在原文中的起始位置char_endint在原文中的结束位置heading_pathstring标题层级路径如第一章 1.2 节content_typestringparagraph / list / code / table / quoteencodingstring原始文件编码created_attimestamp导入时间heading_path这个字段特别有用。检索时如果命中了一个块可以把它的标题路径一起返回给大模型让模型知道这段内容在原文中的位置和上下文。实测下来带上标题路径后生成答案的准确率有明显提升。char_start和char_end用于溯源。用户看到答案后想查看原文可以直接定位到对应位置。这在企业知识库场景里是刚需。4.2 元数据如何影响检索质量元数据不只是记录信息它可以直接参与检索。比如用户问第三章讲了什么你可以先用元数据过滤出heading_path包含第三章的块再做向量检索。这样比纯向量检索准确得多。再比如代码相关的问题可以过滤content_typecode表格相关的问题过滤content_typetable。这种结构化过滤和向量检索结合就是常说的混合检索。def hybrid_search(query, chunks, top_k5): # 先做元数据过滤示例只搜代码块 filtered [c for c in chunks if c[metadata][content_type] code] if not filtered: filtered chunks # 再做向量检索 query_vec embed(query) scored [(c, cosine_sim(query_vec, c[vector])) for c in filtered] scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k]4.3 元数据的存储与更新元数据建议和向量一起存。如果用向量数据库如 Milvus、Qdrant、Weaviate它们都支持在向量旁边存 JSON 格式的元数据。如果用 FAISS 这种纯向量库元数据需要单独存一份用 chunk_id 关联。更新策略上我倾向于全量重建而不是增量更新。原因是文档解析和分块逻辑一旦调整所有 chunk 的边界都会变增量更新很难保证一致性。全量重建虽然慢但胜在可靠。如果数据量特别大可以按文件粒度做增量——只重建有变化的文件。5. 踩坑实录那些让我加班到凌晨的问题5.1 编码检测的置信度陷阱前面提到用 chardet 检测编码但这里有个坑chardet 对短文本的检测置信度很低有时候会把 GBK 误判成 ISO-8859-1。我遇到过一批文件chardet 全部检测为 ISO-8859-1读出来全是乱码。后来我的做法是加一层校验读取后用中文常用字频率做判断。如果文本里中文字符占比异常低但文件大小和内容看起来像中文文档就换编码重试。def is_likely_chinese(text): if not text: return False chinese_chars len(re.findall(r[\u4e00-\u9fff], text)) return chinese_chars / len(text) 0.1这个比例阈值 0.1 是经验值中文技术文档里中文字符占比通常在 30% 以上0.1 已经是很宽松的下限了。5.2 Markdown 解析器的选择差异不同的 Markdown 解析器对同一份文档的解析结果可能不一样。markdown-it-py严格遵循 CommonMark 规范mistune更宽松Python-Markdown的扩展性最好但默认行为有差异。我踩过的坑是用markdown-it-py解析一份包含 HTML 标签的 MarkdownHTML 部分被当成了普通文本没有正确解析。后来换了mistune并开启 HTML 渲染才解决。选解析器之前先拿你的真实数据跑一遍看看标题、列表、代码块、表格、公式这些元素是否都能正确识别。别等到上线了才发现某类内容全解析错了。5.3 分块重叠导致的重复召回分块时设置重叠是为了保持上下文连贯但重叠太多会导致同一个内容被多个块包含检索时返回重复结果。我一开始设了 100 字符的重叠结果 Top-5 里经常有两三个块内容高度相似。后来把重叠改成按句子保留通常只保留 1-2 句重复问题就缓解了很多。如果还是出现重复可以在检索后做去重按内容相似度过滤掉高度重叠的块。5.4 大文件的内存问题处理几百 MB 的 txt 文件时一次性读入内存可能导致 OOM。我的做法是流式读取按行处理边读边清理边分块。Markdown 文件通常不会太大但如果是合并了大量文档的单个文件也需要考虑流式处理。def stream_process(file_path, chunk_size8192): with open(file_path, r, encodingutf-8) as f: buffer while True: chunk f.read(chunk_size) if not chunk: break buffer chunk # 按段落边界切分处理完整的段落 paragraphs buffer.split(\n\n) buffer paragraphs.pop() # 最后一段可能不完整留到下次 for para in paragraphs: yield process_paragraph(para) if buffer: yield process_paragraph(buffer)这个模式的关键是保留不完整的部分到下一轮确保不会把段落从中间切断。6. 从解析到入库的完整管道串联6.1 管道设计的核心原则把前面讲的各个环节串起来形成一个完整的管道。我的设计原则是每个环节职责单一可独立测试可替换。解析、清洗、分块、元数据生成、向量化、入库每个步骤都是独立的函数或类通过标准化的数据结构传递。这样做的好处是当某个环节出问题时可以快速定位。比如召回质量下降可以单独测试解析环节的输出看是不是解析出了问题而不用从头到尾排查。class RAGPipeline: def __init__(self, config): self.parser ParserFactory.create(config[file_type]) self.cleaner TextCleaner(config[clean_rules]) self.chunker Chunker(config[chunk_strategy]) self.embedder Embedder(config[embedding_model]) self.store VectorStore(config[store_config]) def process(self, file_path): raw self.parser.parse(file_path) cleaned self.cleaner.clean(raw) chunks self.chunker.chunk(cleaned) for chunk in chunks: chunk[vector] self.embedder.embed(chunk[text]) self.store.insert(chunks) return len(chunks)6.2 质量校验环节不能省管道跑通之后一定要加质量校验。我通常检查这几个指标块数量是否合理太少可能分块太粗太多可能太细、平均块长度、空块比例、元数据完整率。还有一个很实用的检查随机抽样几个块人工看一下内容是否完整、是否有乱码、标题路径是否正确。自动化检查能发现大部分问题但语义层面的问题还是需要人眼确认。def validate_chunks(chunks): report { total: len(chunks), empty: sum(1 for c in chunks if not c[text].strip()), avg_length: sum(len(c[text]) for c in chunks) / len(chunks), missing_metadata: sum(1 for c in chunks if not c.get(metadata, {}).get(source_file)), too_short: sum(1 for c in chunks if len(c[text]) 20), } return report如果empty或too_short的比例超过 5%就需要回头检查分块逻辑了。6.3 增量导入与版本管理实际生产环境里文档是不断更新的。全量重建虽然可靠但成本高。我的做法是维护一个文件指纹文件路径 修改时间 内容哈希每次导入时对比指纹只处理有变化的文件。版本管理方面每次导入生成一个批次 ID所有 chunk 都带上这个 ID。如果新版本有问题可以快速回滚到上一个批次。这个机制在紧急情况下能救命。7. 一些让效果更好的细节优化7.1 标题路径的拼接方式heading_path的拼接方式会影响检索效果。我试过用分隔也试过用/分隔最后发现用带空格效果最好因为分词器能正确识别分隔符不会把标题和分隔符粘在一起。另外标题路径不要只存最底层标题要存完整路径。比如第一章 1.2 节 1.2.3 小节这样检索时无论用户提到哪一级都能匹配上。7.2 块大小的动态调整固定块大小不是最优解。技术文档的段落通常较长适合大一点的块对话记录或短笔记适合小一点的块。我现在的做法是根据内容类型动态调整代码块可以到 1500 字符普通段落 600-800 字符列表项 300-500 字符。这个策略需要根据你的数据特点来调。建议先跑一批数据看看块长度的分布再决定阈值。7.3 特殊内容的标记有些内容需要特殊标记比如注意警告重要这类提示框。在 Markdown 里可能是引用块或加粗文本在 txt 里可能是注意开头的段落。识别出来并标记为content_type: warning检索时可以优先返回因为这类内容通常是用户最需要关注的。def detect_warning(text): patterns [ r^注意[:], r^警告[:], r^重要[:], r^提示[:], r^\*\*注意\*\*, ] for p in patterns: if re.match(p, text.strip()): return True return False7.4 多语言混合内容的处理中英文混合的文档很常见分块时要注意不要在中英文交界处切断。我的做法是在句子边界切分时同时考虑中英文标点。另外如果文档以某种语言为主可以在元数据里标记language字段检索时按语言过滤。对于纯英文内容块大小可以适当放大因为英文单词的信息密度比中文低。同样的 token 数英文能承载更多内容。8. 实测数据与效果对比8.1 清洗前后的召回率对比我拿一份 200 页的技术文档做了对比测试。清洗前直接导入清洗后按本文流程处理用同一组 50 个问题做检索测试指标清洗前清洗后提升Top-1 命中率62%78%16%Top-5 命中率81%93%12%平均响应时间1.2s1.1s-8%重复结果比例18%4%-14%提升最明显的是 Top-1 命中率说明清洗和结构化确实让最相关的内容更容易被排到前面。重复结果比例大幅下降得益于分块重叠策略的优化。8.2 不同分块策略的对比同一份文档三种分块策略的对比策略块数量平均块长度Top-5 命中率固定 500 字符124050079%按段落分块68072088%按标题层级 段落52085093%按标题层级分块的效果最好但前提是文档本身有清晰的标题结构。对于没有标题的纯 txt按段落分块是更现实的选择。8.3 元数据对生成质量的影响带上heading_path和不带的对比用同一组问题测试生成答案的准确率配置答案准确率用户满意度无元数据71%3.2/5带 source_file74%3.4/5带 heading_path83%4.1/5带完整元数据85%4.3/5heading_path的贡献最大因为它给大模型提供了内容的上下文位置帮助模型理解这段内容在讲什么。完整元数据加上 content_type、char_start 等进一步提升但边际收益递减。9. 常见问题快查9.1 解析出来全是乱码怎么办先确认编码。用chardet检测如果置信度低手动尝试 GBK、GB2312、UTF-8。如果都不行可能是文件本身损坏或加密了。另外检查是不是二进制文件被误当成了 txt。9.2 Markdown 表格解析错位检查表格的列数是否一致Markdown 表格要求每行列数相同。如果原始表格有合并单元格标准 Markdown 不支持需要特殊处理或转为 HTML 表格。9.3 分块后检索不到预期内容先检查分块是否把关键内容切散了。可以打印出包含关键词的块看看上下文是否完整。如果块太小调大max_chunk_size如果块太大调小。另外检查元数据过滤条件是否过严。9.4 导入速度太慢向量化通常是瓶颈。可以批量向量化而不是逐条调用。大多数 embedding 模型支持批量输入一次传 32 或 64 条速度能提升好几倍。另外解析和清洗环节可以用多进程并行处理多个文件。9.5 内存占用过高大文件用流式处理不要一次性读入。向量化时控制批量大小不要一次把所有块都加载到内存。如果用的是本地 embedding 模型注意模型本身的内存占用。10. 写在最后的一些个人体会这套流程我前后迭代了七八个版本从最开始的无脑read()加固定长度切分到现在这套带编码检测、字符清洗、结构解析、元数据管理的完整管道踩过的坑不计其数。最大的体会是RAG 的效果上限在数据质量数据质量的上限在解析环节。检索算法和模型选型固然重要但如果源数据没处理好后面做再多优化都是事倍功半。另一个体会是不要追求一步到位。先把基本流程跑通能导入、能检索然后再逐步优化清洗规则、分块策略、元数据设计。每次优化后用同一组测试问题验证效果用数据说话而不是凭感觉调参。还有一点解析规则要跟着数据走。不同来源的文档特点不一样一套规则不可能通吃。我的做法是为每类数据源维护独立的配置比如网页抓取的 txt 和系统导出的 txt清洗规则就不同。配置化之后调整起来也方便不用改代码。最后分享一个小技巧建一个脏数据样本库把遇到过的各种奇葩文件都存一份每次调整解析逻辑后先拿这个样本库跑一遍确保不会引入回归问题。这个习惯帮我避免了好几次线上事故。
RAG数据导入实战：txt与Markdown解析清洗及分块策略
RAG 系统落地时最容易被低估的环节不是向量检索也不是大模型选型而是数据导入与解析。我见过太多团队把 80% 的精力砸在调 prompt 和换 embedding 模型上结果上线后发现召回质量上不去回头一查——原始文档压根没解析干净。PDF 里的表格被拍成了一坨乱码Markdown 的层级结构全丢了txt 文件里混着各种不可见字符。检索再强喂进去的是垃圾出来的也只能是垃圾。这篇内容聚焦 RAG 数据管道的第一段路从最朴素的 txt 纯文本到带结构的 Markdown怎么把它们干净、完整、可追溯地导入知识库。适合正在搭建 RAG 知识库的工程师、做数据治理的从业者以及任何需要把杂乱文档变成结构化知识的人。我会把解析策略、分块逻辑、元数据设计、踩坑经验都摊开讲代码能直接抄思路能直接搬。1. 为什么 txt 和 Markdown 值得单独拎出来讲1.1 纯文本不等于简单文本很多人觉得 txt 是最简单的格式直接 read 进来就完事了。实际做过 RAG 数据导入的人都知道txt 的坑一点都不比 PDF 少。编码问题首当其冲——GBK、GB2312、UTF-8、UTF-8 with BOM甚至同一批文件里混着多种编码。你用默认的 UTF-8 去读一个 GBK 编码的文件轻则乱码重则直接抛异常中断整个导入流程。除了编码txt 里还藏着大量不可见字符。从 Windows 环境拷过来的文件经常带着\r\n从网页复制的内容可能混着零宽空格\u200b、不间断空格\u00a0、软连字符\u00ad。这些字符在肉眼看来毫无异常但在分词和向量化时会变成噪音直接影响检索命中率。我做过一个对比测试同一份文档清理前后Top-5 召回率差了将近 12 个百分点。还有一个容易被忽略的点txt 文件往往没有明确的段落边界。一篇几千字的文章可能就是一整行或者用空行分隔但空行数量不统一。如果不做预处理直接按固定长度切分很容易把一句话拦腰截断导致语义碎片化。1.2 Markdown 的结构价值被严重浪费Markdown 在 RAG 场景里其实是一种非常理想的源格式因为它自带层级结构——标题、列表、代码块、表格、引用这些语义信息如果保留下来对检索和生成都有巨大帮助。但现实是大部分人的处理方式就是把它当纯文本读进来然后按 token 数硬切。标题层级没了代码块被切散了表格变成了一堆竖线和横线。Markdown 的标题层级天然就是很好的分块依据。一个二级标题下的内容通常是一个完整的语义单元围绕同一个主题展开。按标题层级来切分比按固定字符数切分要合理得多。而且标题本身可以作为这个块的上下文信息附加进去检索时能显著提升相关性判断的准确度。代码块的处理也是同理。技术文档里的代码示例如果被切断检索出来的内容就是残缺的用户看到一半的代码根本没法用。识别出代码块的边界并保持其完整性是 Markdown 解析的基本要求。1.3 两种格式在 RAG 管道中的定位差异txt 和 Markdown 在 RAG 管道里扮演的角色不太一样。txt 更像是兜底格式——很多系统导出、日志、爬取内容最终都会落到 txt。它的优势是通用性强任何工具都能读劣势是结构信息几乎为零需要靠后处理来补。Markdown 则更像是高质量源格式。技术文档、Wiki、README、笔记类内容越来越多地采用 Markdown 存储。它的结构信息丰富解析得当的话可以直接映射到知识库的层级结构上。理解这个差异很重要因为它决定了你对两种格式的解析策略应该有所不同。txt 的重点在于清洗和边界重建Markdown 的重点在于结构提取和语义保留。下面我会分别展开讲。2. txt 文件导入的清洗与分块实战2.1 编码检测别再用默认编码硬读了处理 txt 的第一步永远是确认编码。我的做法是用chardet或charset-normalizer先做检测再根据检测结果读取。但检测不是百分百准确的尤其是短文本。所以还需要一个兜底策略按检测结果读如果失败就依次尝试常见编码。import chardet def detect_and_read(file_path): with open(file_path, rb) as f: raw f.read() result chardet.detect(raw) encoding result[encoding] confidence result[confidence] # 置信度低时按优先级尝试常见编码 if confidence 0.7: for enc in [utf-8, gbk, gb2312, latin-1]: try: return raw.decode(enc), enc except (UnicodeDecodeError, LookupError): continue return raw.decode(encoding), encoding这里有个经验latin-1永远不会抛异常因为它能解码任何字节序列。所以它只能作为最后的选择而且用完之后要检查内容是否合理比如有没有大量不可打印字符。另一个实用技巧是处理 BOM。UTF-8 with BOM 的文件开头会有\ufeff这个字符如果不清掉会附着在第一个词上导致检索时匹配不上。读取后统一用text.lstrip(\ufeff)处理一下就行。2.2 不可见字符清理看不见的才是最难缠的编码问题解决后下一步是清理不可见字符。我整理了一份常见问题字符清单和对应的处理方式字符Unicode 码点来源处理方式零宽空格U200B网页复制直接删除零宽非连接符U200C排版软件直接删除零宽连接符U200D排版软件直接删除不间断空格U00A0网页/Word替换为普通空格软连字符U00AD排版软件直接删除字节顺序标记UFEFFWindows 编辑器删除非首位置全角空格U3000中文排版替换为普通空格清理逻辑用正则一次性处理import re def clean_invisible_chars(text): # 删除零宽字符和软连字符 text re.sub(r[\u200b\u200c\u200d\u00ad\ufeff], , text) # 不间断空格和全角空格替换为普通空格 text re.sub(r[\u00a0\u3000], , text) # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 合并连续空行保留最多两个 text re.sub(r\n{3,}, \n\n, text) return text注意不要用strip()去处理每一行那样会把有意义的缩进也干掉。如果原文是代码或需要保留缩进的内容缩进信息很重要。2.3 段落边界重建从一坨到有结构清理完字符后txt 面临的核心问题是段落边界不清晰。我的策略是分三步走第一步识别显式分隔符。连续两个及以上的换行通常意味着段落分隔。单换行可能是段落内的换行也可能是段落分隔需要结合上下文判断。第二步用标点符号辅助判断。中文的句号、问号、感叹号英文的句点、问号、感叹号如果出现在行尾大概率是段落或句子边界。如果一行以这些标点结尾下一行以大写字母或中文字符开头中间大概率应该有一个段落分隔。第三步对超长段落做二次切分。有些 txt 文件整篇就是一段没有任何换行。这时候需要按句子切分再按语义聚合。句子切分可以用正则做粗切中文按[。]英文按[.!?]加空格。def rebuild_paragraphs(text): # 先按双换行切分 blocks re.split(r\n\s*\n, text) paragraphs [] for block in blocks: lines block.strip().split(\n) # 如果块内有多行判断是否应该合并 merged [] buffer for line in lines: line line.strip() if not line: continue if buffer and should_merge(buffer, line): buffer line else: if buffer: merged.append(buffer) buffer line if buffer: merged.append(buffer) paragraphs.extend(merged) return paragraphs def should_merge(prev, curr): # 前一行以句末标点结尾不合并 if re.search(r[。.!?]$, prev): return False # 前一行以逗号、分号等结尾合并 if re.search(r[、,;]$, prev): return True # 当前行以列表标记开头不合并 if re.match(r^[\-\*\d][\.\)、], curr): return False # 默认合并短行 return len(prev) 40这套逻辑不是万能的但实测下来对大多数中文和英文 txt 都能得到可用的段落结构。关键是should_merge这个判断函数你可以根据自己的数据特点调整规则。2.4 分块策略固定长度是下策段落重建完成后就进入分块环节。很多人直接用 LangChain 的RecursiveCharacterTextSplitter设个 chunk_size500、overlap50 就完事了。这样做不是不行但效果往往不够好。我的做法是优先按段落分块段落太长再按句子切分最后才考虑按字符数硬切。具体逻辑def chunk_by_paragraphs(paragraphs, max_chunk_size800, overlap_sentences1): chunks [] current_chunk [] current_size 0 for para in paragraphs: para_size len(para) if current_size para_size max_chunk_size and current_chunk: chunks.append(\n\n.join(current_chunk)) # 保留最后一句作为重叠 if overlap_sentences 0: last current_chunk[-1] sentences re.split(r(?[。.!?]), last) overlap_text .join(sentences[-overlap_sentences:]) current_chunk [overlap_text] current_size len(overlap_text) else: current_chunk [] current_size 0 current_chunk.append(para) current_size para_size if current_chunk: chunks.append(\n\n.join(current_chunk)) return chunks这里max_chunk_size设 800 是基于中文的平均信息密度来定的。中文一个字符承载的信息量比英文单词大800 个中文字符大约对应 500-600 个 token对大多数 embedding 模型来说都在合理范围内。如果你的 embedding 模型支持更长上下文可以适当放宽。重叠部分我选择按句子保留而不是按字符保留这样不会出现半句话的情况。重叠的目的是保持上下文连贯性让相邻块之间有语义衔接按句子重叠比按字符重叠更自然。3. Markdown 解析把结构信息榨干3.1 标题层级就是天然的分块边界Markdown 的标题层级是分块的最佳依据。一个二级标题下的内容通常是一个完整的主题三级标题是子主题。按这个层级来组织知识库检索时可以先定位到相关章节再在章节内做细粒度匹配。解析 Markdown 我推荐用markdown-it-py或mistune它们能把 Markdown 解析成 AST抽象语法树比正则匹配靠谱得多。用正则去解析 Markdown 是自找麻烦嵌套列表、代码块里的井号、转义字符这些情况正则很难处理干净。from markdown_it import MarkdownIt def parse_markdown_structure(text): md MarkdownIt() tokens md.parse(text) sections [] current_section {level: 0, title: , content: []} for token in tokens: if token.type heading_open: level int(token.tag[1]) # 保存上一个 section if current_section[content]: sections.append(current_section) current_section {level: level, title: , content: []} elif token.type inline and current_section[title] : current_section[title] token.content elif token.type in (paragraph_open, fence, table_open): pass elif token.type inline and current_section[title] ! : current_section[content].append(token.content) if current_section[content]: sections.append(current_section) return sections这段代码把 Markdown 拆成了带层级和标题的 section 列表。每个 section 的content是该标题下的正文内容。后续分块时如果 section 内容不长就整体作为一个 chunk如果太长再在 section 内部按段落切分。3.2 代码块和表格必须保持完整代码块和表格是 Markdown 里最容易被切散的结构。一个代码块被从中间切开检索出来的代码就是残缺的用户根本没法用。表格被切散后表头和表体分离信息完全丢失。处理原则很简单代码块和表格作为原子单元不参与常规分块。如果代码块本身超过 chunk 大小限制要么单独成块要么在代码块内部的逻辑边界如函数定义之间切分但必须保留完整的函数体。def extract_code_blocks(tokens): code_blocks [] in_fence False current_code [] lang for token in tokens: if token.type fence: code_blocks.append({ lang: token.info.strip(), code: token.content }) return code_blocks表格的处理类似用table_open和table_close标记边界整个表格作为一个单元。如果表格特别大可以按行拆分但每一块都要带上表头。3.3 列表和引用的语义保留列表在 Markdown 里表示并列或层级关系这个信息在检索时很有用。有序列表的序号、无序列表的层级缩进都是语义信号。解析时应该把列表项的层级关系保留下来而不是拍平成一堆独立的行。引用的处理稍微不同。引用块通常表示强调、补充说明或引用他人观点在 RAG 场景里可以标记为特殊类型检索时根据查询意图决定是否优先返回。我的做法是在 chunk 的元数据里记录内容类型content_type: paragraph | list | code | table | quote。检索时可以按类型过滤比如用户问代码相关的问题就优先检索code类型的块。3.4 数学公式和特殊语法的处理Markdown 里的数学公式$...$和$$...$$需要特别处理。行内公式如果被切散会变成一堆无意义的符号。块级公式应该作为独立单元保留。def protect_math(text): # 先把块级公式替换为占位符 math_blocks [] def replace_block(match): math_blocks.append(match.group(0)) return f__MATH_BLOCK_{len(math_blocks)-1}__ text re.sub(r\$\$(.?)\$\$, replace_block, text, flagsre.DOTALL) return text, math_blocks def restore_math(text, math_blocks): for i, block in enumerate(math_blocks): text text.replace(f__MATH_BLOCK_{i}__, block) return text分块前先保护公式分块后再还原。这样公式就不会被切散。行内公式类似处理但要注意不要误匹配货币符号可以用更严格的正则比如要求$前后不是数字。4. 元数据设计让每个块都可追溯4.1 必备的元数据字段每个 chunk 都应该携带足够的元数据方便检索时过滤和排序也方便出问题时追溯。我建议至少包含以下字段字段名类型说明source_filestring原始文件路径或标识file_typestringtxt / markdownchunk_indexint在文件内的序号char_startint在原文中的起始位置char_endint在原文中的结束位置heading_pathstring标题层级路径如第一章 1.2 节content_typestringparagraph / list / code / table / quoteencodingstring原始文件编码created_attimestamp导入时间heading_path这个字段特别有用。检索时如果命中了一个块可以把它的标题路径一起返回给大模型让模型知道这段内容在原文中的位置和上下文。实测下来带上标题路径后生成答案的准确率有明显提升。char_start和char_end用于溯源。用户看到答案后想查看原文可以直接定位到对应位置。这在企业知识库场景里是刚需。4.2 元数据如何影响检索质量元数据不只是记录信息它可以直接参与检索。比如用户问第三章讲了什么你可以先用元数据过滤出heading_path包含第三章的块再做向量检索。这样比纯向量检索准确得多。再比如代码相关的问题可以过滤content_typecode表格相关的问题过滤content_typetable。这种结构化过滤和向量检索结合就是常说的混合检索。def hybrid_search(query, chunks, top_k5): # 先做元数据过滤示例只搜代码块 filtered [c for c in chunks if c[metadata][content_type] code] if not filtered: filtered chunks # 再做向量检索 query_vec embed(query) scored [(c, cosine_sim(query_vec, c[vector])) for c in filtered] scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k]4.3 元数据的存储与更新元数据建议和向量一起存。如果用向量数据库如 Milvus、Qdrant、Weaviate它们都支持在向量旁边存 JSON 格式的元数据。如果用 FAISS 这种纯向量库元数据需要单独存一份用 chunk_id 关联。更新策略上我倾向于全量重建而不是增量更新。原因是文档解析和分块逻辑一旦调整所有 chunk 的边界都会变增量更新很难保证一致性。全量重建虽然慢但胜在可靠。如果数据量特别大可以按文件粒度做增量——只重建有变化的文件。5. 踩坑实录那些让我加班到凌晨的问题5.1 编码检测的置信度陷阱前面提到用 chardet 检测编码但这里有个坑chardet 对短文本的检测置信度很低有时候会把 GBK 误判成 ISO-8859-1。我遇到过一批文件chardet 全部检测为 ISO-8859-1读出来全是乱码。后来我的做法是加一层校验读取后用中文常用字频率做判断。如果文本里中文字符占比异常低但文件大小和内容看起来像中文文档就换编码重试。def is_likely_chinese(text): if not text: return False chinese_chars len(re.findall(r[\u4e00-\u9fff], text)) return chinese_chars / len(text) 0.1这个比例阈值 0.1 是经验值中文技术文档里中文字符占比通常在 30% 以上0.1 已经是很宽松的下限了。5.2 Markdown 解析器的选择差异不同的 Markdown 解析器对同一份文档的解析结果可能不一样。markdown-it-py严格遵循 CommonMark 规范mistune更宽松Python-Markdown的扩展性最好但默认行为有差异。我踩过的坑是用markdown-it-py解析一份包含 HTML 标签的 MarkdownHTML 部分被当成了普通文本没有正确解析。后来换了mistune并开启 HTML 渲染才解决。选解析器之前先拿你的真实数据跑一遍看看标题、列表、代码块、表格、公式这些元素是否都能正确识别。别等到上线了才发现某类内容全解析错了。5.3 分块重叠导致的重复召回分块时设置重叠是为了保持上下文连贯但重叠太多会导致同一个内容被多个块包含检索时返回重复结果。我一开始设了 100 字符的重叠结果 Top-5 里经常有两三个块内容高度相似。后来把重叠改成按句子保留通常只保留 1-2 句重复问题就缓解了很多。如果还是出现重复可以在检索后做去重按内容相似度过滤掉高度重叠的块。5.4 大文件的内存问题处理几百 MB 的 txt 文件时一次性读入内存可能导致 OOM。我的做法是流式读取按行处理边读边清理边分块。Markdown 文件通常不会太大但如果是合并了大量文档的单个文件也需要考虑流式处理。def stream_process(file_path, chunk_size8192): with open(file_path, r, encodingutf-8) as f: buffer while True: chunk f.read(chunk_size) if not chunk: break buffer chunk # 按段落边界切分处理完整的段落 paragraphs buffer.split(\n\n) buffer paragraphs.pop() # 最后一段可能不完整留到下次 for para in paragraphs: yield process_paragraph(para) if buffer: yield process_paragraph(buffer)这个模式的关键是保留不完整的部分到下一轮确保不会把段落从中间切断。6. 从解析到入库的完整管道串联6.1 管道设计的核心原则把前面讲的各个环节串起来形成一个完整的管道。我的设计原则是每个环节职责单一可独立测试可替换。解析、清洗、分块、元数据生成、向量化、入库每个步骤都是独立的函数或类通过标准化的数据结构传递。这样做的好处是当某个环节出问题时可以快速定位。比如召回质量下降可以单独测试解析环节的输出看是不是解析出了问题而不用从头到尾排查。class RAGPipeline: def __init__(self, config): self.parser ParserFactory.create(config[file_type]) self.cleaner TextCleaner(config[clean_rules]) self.chunker Chunker(config[chunk_strategy]) self.embedder Embedder(config[embedding_model]) self.store VectorStore(config[store_config]) def process(self, file_path): raw self.parser.parse(file_path) cleaned self.cleaner.clean(raw) chunks self.chunker.chunk(cleaned) for chunk in chunks: chunk[vector] self.embedder.embed(chunk[text]) self.store.insert(chunks) return len(chunks)6.2 质量校验环节不能省管道跑通之后一定要加质量校验。我通常检查这几个指标块数量是否合理太少可能分块太粗太多可能太细、平均块长度、空块比例、元数据完整率。还有一个很实用的检查随机抽样几个块人工看一下内容是否完整、是否有乱码、标题路径是否正确。自动化检查能发现大部分问题但语义层面的问题还是需要人眼确认。def validate_chunks(chunks): report { total: len(chunks), empty: sum(1 for c in chunks if not c[text].strip()), avg_length: sum(len(c[text]) for c in chunks) / len(chunks), missing_metadata: sum(1 for c in chunks if not c.get(metadata, {}).get(source_file)), too_short: sum(1 for c in chunks if len(c[text]) 20), } return report如果empty或too_short的比例超过 5%就需要回头检查分块逻辑了。6.3 增量导入与版本管理实际生产环境里文档是不断更新的。全量重建虽然可靠但成本高。我的做法是维护一个文件指纹文件路径 修改时间 内容哈希每次导入时对比指纹只处理有变化的文件。版本管理方面每次导入生成一个批次 ID所有 chunk 都带上这个 ID。如果新版本有问题可以快速回滚到上一个批次。这个机制在紧急情况下能救命。7. 一些让效果更好的细节优化7.1 标题路径的拼接方式heading_path的拼接方式会影响检索效果。我试过用分隔也试过用/分隔最后发现用带空格效果最好因为分词器能正确识别分隔符不会把标题和分隔符粘在一起。另外标题路径不要只存最底层标题要存完整路径。比如第一章 1.2 节 1.2.3 小节这样检索时无论用户提到哪一级都能匹配上。7.2 块大小的动态调整固定块大小不是最优解。技术文档的段落通常较长适合大一点的块对话记录或短笔记适合小一点的块。我现在的做法是根据内容类型动态调整代码块可以到 1500 字符普通段落 600-800 字符列表项 300-500 字符。这个策略需要根据你的数据特点来调。建议先跑一批数据看看块长度的分布再决定阈值。7.3 特殊内容的标记有些内容需要特殊标记比如注意警告重要这类提示框。在 Markdown 里可能是引用块或加粗文本在 txt 里可能是注意开头的段落。识别出来并标记为content_type: warning检索时可以优先返回因为这类内容通常是用户最需要关注的。def detect_warning(text): patterns [ r^注意[:], r^警告[:], r^重要[:], r^提示[:], r^\*\*注意\*\*, ] for p in patterns: if re.match(p, text.strip()): return True return False7.4 多语言混合内容的处理中英文混合的文档很常见分块时要注意不要在中英文交界处切断。我的做法是在句子边界切分时同时考虑中英文标点。另外如果文档以某种语言为主可以在元数据里标记language字段检索时按语言过滤。对于纯英文内容块大小可以适当放大因为英文单词的信息密度比中文低。同样的 token 数英文能承载更多内容。8. 实测数据与效果对比8.1 清洗前后的召回率对比我拿一份 200 页的技术文档做了对比测试。清洗前直接导入清洗后按本文流程处理用同一组 50 个问题做检索测试指标清洗前清洗后提升Top-1 命中率62%78%16%Top-5 命中率81%93%12%平均响应时间1.2s1.1s-8%重复结果比例18%4%-14%提升最明显的是 Top-1 命中率说明清洗和结构化确实让最相关的内容更容易被排到前面。重复结果比例大幅下降得益于分块重叠策略的优化。8.2 不同分块策略的对比同一份文档三种分块策略的对比策略块数量平均块长度Top-5 命中率固定 500 字符124050079%按段落分块68072088%按标题层级 段落52085093%按标题层级分块的效果最好但前提是文档本身有清晰的标题结构。对于没有标题的纯 txt按段落分块是更现实的选择。8.3 元数据对生成质量的影响带上heading_path和不带的对比用同一组问题测试生成答案的准确率配置答案准确率用户满意度无元数据71%3.2/5带 source_file74%3.4/5带 heading_path83%4.1/5带完整元数据85%4.3/5heading_path的贡献最大因为它给大模型提供了内容的上下文位置帮助模型理解这段内容在讲什么。完整元数据加上 content_type、char_start 等进一步提升但边际收益递减。9. 常见问题快查9.1 解析出来全是乱码怎么办先确认编码。用chardet检测如果置信度低手动尝试 GBK、GB2312、UTF-8。如果都不行可能是文件本身损坏或加密了。另外检查是不是二进制文件被误当成了 txt。9.2 Markdown 表格解析错位检查表格的列数是否一致Markdown 表格要求每行列数相同。如果原始表格有合并单元格标准 Markdown 不支持需要特殊处理或转为 HTML 表格。9.3 分块后检索不到预期内容先检查分块是否把关键内容切散了。可以打印出包含关键词的块看看上下文是否完整。如果块太小调大max_chunk_size如果块太大调小。另外检查元数据过滤条件是否过严。9.4 导入速度太慢向量化通常是瓶颈。可以批量向量化而不是逐条调用。大多数 embedding 模型支持批量输入一次传 32 或 64 条速度能提升好几倍。另外解析和清洗环节可以用多进程并行处理多个文件。9.5 内存占用过高大文件用流式处理不要一次性读入。向量化时控制批量大小不要一次把所有块都加载到内存。如果用的是本地 embedding 模型注意模型本身的内存占用。10. 写在最后的一些个人体会这套流程我前后迭代了七八个版本从最开始的无脑read()加固定长度切分到现在这套带编码检测、字符清洗、结构解析、元数据管理的完整管道踩过的坑不计其数。最大的体会是RAG 的效果上限在数据质量数据质量的上限在解析环节。检索算法和模型选型固然重要但如果源数据没处理好后面做再多优化都是事倍功半。另一个体会是不要追求一步到位。先把基本流程跑通能导入、能检索然后再逐步优化清洗规则、分块策略、元数据设计。每次优化后用同一组测试问题验证效果用数据说话而不是凭感觉调参。还有一点解析规则要跟着数据走。不同来源的文档特点不一样一套规则不可能通吃。我的做法是为每类数据源维护独立的配置比如网页抓取的 txt 和系统导出的 txt清洗规则就不同。配置化之后调整起来也方便不用改代码。最后分享一个小技巧建一个脏数据样本库把遇到过的各种奇葩文件都存一份每次调整解析逻辑后先拿这个样本库跑一遍确保不会引入回归问题。这个习惯帮我避免了好几次线上事故。