ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

RAG数据导入与解析实战:txt与Markdown文本分块指南

RAG数据导入与解析实战:txt与Markdown文本分块指南 做 RAG 项目做久了你会发现一个特别朴素的道理检索效果的上限其实在数据导入阶段就定死了。很多人花大力气调 embedding 模型、调 rerank 权重却对扔进知识库的原始文档不闻不问——结果就是分块切得稀碎、元数据一堆空的、表格变成天书检索出来一堆看似相关实则没法用的片段。我最早做知识库时也踩过这个坑当时把一堆 PDF 和 txt 直接丢进去用户问“第三章节的结论是什么”系统返回一堆目录页码场面非常尴尬。所以这个系列我想专门聊聊 RAG 的数据导入与解析。第一篇先落在最基础的文本类内容上从 txt 这种纯文本到 Markdown 这种带轻量结构的格式把“通用文本怎么洗干净、怎么切块”和“结构化文档怎么拆出语义层次”两件事一次说透。适合正在搭知识库、但还没想清楚文档预处理该怎么做的人参考也适合想把手头的一堆 md 笔记、txt 书稿转化成高质量 RAG 数据源的同学照抄作业。1. 动手前先想清楚解析到底在解决什么问题1.1 检索效果差的锅一半要甩给导入阶段很多人理解 RAG 的流程是从“文档入库”开始的但其实真正的起点是“文档解析”。这一步直接决定了后续两个关键环节的质量第一个是分块。Embedding 模型有最大 token 限制一段文本不能整个扔进去必须切成块。但怎么切、按什么边界切直接决定了每个块内部的语义是不是连贯的。如果你拿一个 500 字的 txt 文件按固定 512 个字符硬切一个完整的段落会被拦腰斩断切出来的两个块各自都不完整检索时谁也匹配不上。第二个是元数据。RAG 不是纯粹靠向量相似度还要靠元数据做过滤、做引用溯源。一份 Markdown 文档里有标题层级、有章节编号、有代码块、有表格这些结构信息如果解析时全丢了后面做“按章节检索”“只看某个表格的内容”就无从谈起。更麻烦的是回答时你可能连这段内容出自哪个文档的哪个章节都说不出来。我经常打一个比方解析是把原料做成半成品分块和向量化是把它做成成品。半成品要是切坏了、洗不干净后面厨艺再好也白搭。所以先把“导入解析”这件事做扎实省下的调试时间远比你想的多。1.2 为什么这个系列从 txt 和 Markdown 讲起不是所有知识库资料都是排版精美的 PDF。实际项目里大量数据源反而是最不起眼的 txt 和 Markdown代码库的 README、手写的技术笔记、爬下来清洗过的正文、古籍 OCR 后的纯文本、GitHub 上的 .md 文档……这些格式门槛低、生态广几乎任何系统都能读但恰恰因为“太简单”容易被随手处理反而问题最多。txt 的价值在于“没有任何结构”逼着你把文本清洗和分块策略想清楚Markdown 的价值在于“有轻量结构但不复杂”正好是练习结构化解析的最佳样本。把这两类搞定后面再碰 PDF、Word、HTML本质上都是在同一套思路上做加法。另外多说一句这个系列标题里带着“一”后续我打算继续写表格文档解析、PDF 版面还原、复杂版面多模态内容入库这些场景。但那些场景里的很多底层手段比如编码判断、清洗规则、分块策略、元数据抽取都是这篇里打好的地基。2. 通用文本导入txt 的处理比你想的麻烦2.1 编码问题是第一道坎别上来就 open()txt 最坑的地方是没有一个统一的标准去声明它的编码。同一个文件可能是 UTF-8可能是 GBK也可能是 UTF-8-BOM甚至还有各种 obscure 的本地编码。你用open(book.txt, r, encodingutf-8)一读GBK 文件直接抛UnicodeDecodeError运气好不报错的读出来一整篇乱码。我现在的做法是先用chardet做一次编码探测再根据探测结果读取最后再用几个关键特征做校验。比如中文字符占比、是否包含常见 BOM 头、有没有连续的“”替换字符。探测只给建议最终要自己校验。import chardet def detect_and_read(filepath: str) - str: with open(filepath, rb) as f: raw f.read() # 先看有没有 BOM if raw.startswith(codecs.BOM_UTF8): return raw.decode(utf-8-sig) if raw.startswith(codecs.BOM_GB18030): return raw.decode(gb18030) # chardet 探测 detected chardet.detect(raw) encoding detected.get(encoding, utf-8) confidence detected.get(confidence, 0.0) # 低置信度时按优先级尝试 for enc in [encoding, utf-8, gb18030, big5]: try: text raw.decode(enc) # 简单校验乱码字符占比不能太高 bad_chars sum(1 for ch in text if ch \ufffd) if bad_chars / max(len(text), 1) 0.01: return text except (UnicodeDecodeError, LookupError): continue raise ValueError(f无法确定编码: {filepath})小技巧如果确认是中文场景gb18030比gbk更值得作为兜底编码。它是 GBK 的超集兼容性更强。有些老式工具生成的 GBK 文件里面对生僻字或者特殊符号用的就是 gb18030 的扩展区用 gb2312 或 gbk 会解不干净。2.2 清洗规则哪些内容必须丢掉纯文本最让人头疼的不是读不进来而是读进来之后里面什么都有。从网上爬的正文带页眉页脚、带站点导航扫描版 OCR 出来的文本行尾经常多出一些虚假的“换行”还有的人会把目录、编者按、附录全塞进同一个 txt。这些噪声如果不清洗会对分块造成很大干扰——脏文本会让整个块的主题被带偏检索时匹配到一堆垃圾片段。总结下来我的清洗规则优先级是这样排的去掉页眉页脚和重复模板。规则就是跨段落重复出现的行直接删除再看内容里是否有固定的站点名/文档版本号/日期格式。去掉目录区域。很多 txt 前面几页是目录里面有大量“第X章 ....... 12”这种行。用正则识别出“第.章|^\d.\d”并且行内以连续点号、空格或 tab 结尾接页码的可以整块删掉。合并被硬换行切断的段落。OCR 和某些排版导出的 txt每个可视行后面都有一个\n但它们在语义上属于同一个段落。判断逻辑是当前行末尾没有句号、问号、感叹号、冒号等终止符且下一行不是新章节标题时就把两行拼起来。压缩多余空行。多个连续\n替换成两个保持段落间距即可。去掉特殊噪声符号。一些网页导出文本会带着nbsp;、#160;、零宽空格\u200b这类不可见字符统一清掉。清洗的目标不是把文本变短而是让“分块”拿到一个干净的段落序列。段落干净了后面的语义切块才靠谱。2.3 分块策略别迷信固定大小新手最容易踩的坑就是写死一个chunk_size500然后按字符数硬切。我一开始也这么干效果就是检索时经常返回半句话、内容断裂、引文残缺。固定大小不是不能用但必须在“保持语义完整”的前提下切。我的经验是先按文档结构切再按大小限制合并或拆分。具体顺序是先把文档拆成段落列表用\n\n分割或者用句号、问号等作为段落内句子边界。给段落分组连续的小段落如果加起来字数没有超过上限比如 800 字就合并成一个块如果单个段落超过上限再按句号做二次切分。块和块之间留一点重叠overlap比如向前重叠 50~100 字符用来缓解跨块语义丢失问题。下面是一个我试过很稳的简单实现思路def split_into_chunks(text: str, max_chars: int 800, overlap: int 80): # 先以空行切段落 paragraphs [p.strip() for p in re.split(r\n\s*\n, text) if p.strip()] chunks [] current for para in paragraphs: if len(current) len(para) 1 max_chars: current (current \n para).strip() else: # 先把 current 存起来 if current: chunks.append(current) # 如果这个段落本身就超长需要按句切 if len(para) max_chars: sentences re.split(r(?[。]), para) tmp for sent in sentences: if len(tmp) len(sent) max_chars: tmp sent else: if tmp: chunks.append(tmp) tmp sent current tmp else: current para if current: chunks.append(current) # 加 overlap每个 chunk 末尾补上一段前一个 chunk 的尾部 if overlap 0: final_chunks [] for i, chunk in enumerate(chunks): if i 0: chunk chunks[i - 1][-overlap:] \n chunk final_chunks.append(chunk) return final_chunks return chunks注意overlap 加在哪个方向是有讲究的。我一般只往前重叠不往后重叠。因为检索时一个 chunk 和前面的 chunk 相关性更强往前重叠能保留上文语境往后重叠反而容易让相邻两个 chunk 内容高度重复浪费向量库容量。3. Markdown 解析从“能读”到“懂结构”3.1 Markdown 为什么是 RAG 的好原料如果你手里有一批整理得不错的 Markdown 笔记恭喜这几乎是最适合做知识库的文本格式了。因为 Markdown 自带轻量结构#标题定义层级-列表定义枚举关系代码块隔离技术内容|表格定义行列结构[](url)标注链接**bold**标记重点。这些结构信息不是给人看的装饰而是天然的“语义边界”。标题层级能告诉我们一个章节从哪里开始、到哪里结束代码块说明这里是一段独立的技术内容表格本身就是一个结构化的知识单元。解析 Markdown 做 RAG价值恰恰在于把这些边界提取出来而不是把 Markdown 降级成纯文本再切块。我见过有人这么做把所有.md文件读进来正则把#、*、反引号全剥掉变成一个纯字符串去 embedding。等于是把一座房子的梁柱拆了当柴烧太可惜了。正确的思路是把 Markdown 当“带标注的文本”来解析保留住结构语义让每个分块知道自己在文档树里的位置。3.2 用 AST 还是正则我的选择是 AST处理 Markdown 有三种常见思路正则表达式跑一遍把标题、列表、代码块提取出来。适合非常简单的场景但很容易被嵌套列表、转义字符、行内代码里的#搞崩溃。用 markdown-it-py 或 marko 这类解析器先把 Markdown 转成 AST抽象语法树再遍历树节点做提取和分块。灵活可控是我目前的主力方案。用大模型或专门的版面分析模型做“语义结构识别”。适合复杂文档但成本高、慢不适合批量处理海量 md 文件。正则为什么容易翻车因为 Markdown 有上下文。比如一行内容是# 标题如果它在代码块内部它其实是个字符串不是标题如果前面的行是\# 标题转义后它也不是标题。正则很难把这些语境差异全考虑进去AST 就不会有这个问题——它在解析阶段已经搞清楚了哪些是代码块、哪些是文本。用markdown-it-py的时候基本套路是这样from markdown_it import MarkdownIt from markdown_it.tree import SyntaxTreeNode md MarkdownIt(commonmark, {html: False}).enable(table) def parse_md(filepath: str): with open(filepath, r, encodingutf-8) as f: source f.read() tokens md.parse(source) node SyntaxTreeNode(tokens) return node # 遍历节点 def walk(node, depth0): if node.type heading: level node.tag # h1, h2 ... text .join(child.content for child in node.children if child.type inline) print( * depth, level, text) for child in node.children: walk(child, depth 1)这只是个轮廓。实际生产时我会把这个 AST 遍历过程做成一个“结构化抽取器”输出的不是 Markdown 原文而是带元数据的文档对象列表{ id: doc-001, title: xx笔记, sections: [ { heading: 2.3 分块策略, level: 2, content: ..., code_blocks: [..., ...], tables: [|...|] } ] }3.3 代码块、表格、行内样式怎么处理不同类型的 Markdown 节点在 RAG 里的处理方式应该不一样代码块我的建议是保留但不要跟正文混在一个 chunk 里。代码块适合做“代码搜索”场景或者作为某个知识点的示例补充。实操时我会把代码块单独抽出来打上language标签从语言标注提取比如 python 里的 python然后和它所在的章节标题一起作为独立 chunk 入库。这样用户问“给我一个 python 调用示例”检索时命中的是代码块本身而不是包着代码的长篇说明。表格这是 Markdown 解析里最容易被忽略、也最容易翻车的地方。纯文本切块时表格很容易被拦腰截断表头和表体分离。我的做法是在 AST 里把整个 table 节点作为一个不可分割的单元单独提取单独入库。必要时可以把 Markdown 表格转成简化的键值对描述比如把表头当作字段名每一行转成一个对象。这样检索时如果用户问“第三列的最大值是多少”向量匹配时更容易在语义上感知到“某一列”这个维度。行内样式比如**加粗**、*斜体*、[链接](url)。这些不建议粗暴剥掉但也不建议原样保存。我通常把加粗和斜体降级为纯文本因为在大多数 embed 模型里**和*只是噪声但链接会保留成一个特殊标记比如[文本](url)转成文本(url)这样既能提取出链接作为元数据又不干扰正文语义。3.4 图片路径与本地资源Markdown 里的隐藏坑很多人的 Markdown 笔记里有大量图片路径可能是![](./images/foo.png)也可能是绝对路径![](/Users/name/img/a.jpg)甚至可能是网络 URL。RAG 做的是文本检索纯文字模型理解不了图片内容但这并不意味着图片路径可以随手丢掉。我的处理经验是先把图片路径统一抽取出来放到当前章节的元数据里比如{images: [./images/foo.png, ...]}然后在正文里把原来的图片语法替换成一个占位符[图片:foo]。这样分块和检索时正文不会因为图片语法产生噪声但需要溯源时还能通过元数据找到这张图片的位置。如果你的系统要支持“知识库能存图片吗”这类需求那就需要在导入阶段对图片做多模态解析一般是交给视觉模型生成描述文本再把描述文本和图片路径一起入库。这一步不是本篇重点但路径抽取的前置工作是通用的提前做掉不亏。4. 完整实操从 txt 到 Markdown 的解析流水线4.1 整体流程架构我自己在项目里跑的解析流水线分五步读取 - 清洗 - 结构识别 - 分块 - 元数据组装。不管是 txt 还是 Markdown都走同一套框架只是在“结构识别”这步做分流txt 走的是纯文本规则章节标题检测、段落合并Markdown 走的是 AST 遍历。整体流程大致是这样输入文件统一读成字符串编码检测并归一化。按文件类型进入不同分支txt 走纯文本清洗md 走 AST 解析。对解析结果做“文档对象”封装每个对象包含 heading 路径、正文、代码块列表等。分块器按对象的标题路径结合最大字符数和 overlap生成 chunk。给每个 chunk 补上元数据来源文件、标题路径、chunk 序号、字符数、hash 值等。把 chunk 和元数据一起送进向量库同时把原始文档全文存一份用于溯源。如果你用 LangChain 或者 LlamaIndex也可以把上面这些封装成自定义的Loader和Splitter替换掉默认的按字符切分逻辑。但我的建议是前期先把自己的解析函数跑通确认输入输出的结构没问题再往框架里套不建议一上来就依赖框架自带 loader——它们的通用解是针对英语场景优化的中文文本的边界和标点逻辑常常对不上。4.2 环境依赖准备整个流程我用的是 Python 3.10核心依赖就三个markdown-it-py解析 Markdown 成 AST支持表格插件。chardet编码探测。pandas可选表格转结构化描述时用来处理 TSV/CSV。装依赖pip install markdown-it-py chardet pandas不装也可以自己写一个轻量的表格拆分逻辑。但 pandas 在处理大表格时确实省心尤其是转置、列名提取这些操作写起来很顺手。4.3 核心实现Markdown 结构化抽取器下面是一段我实际在用的核心实现虽然做了简化但主干逻辑没变import re from markdown_it import MarkdownIt from markdown_it.tree import SyntaxTreeNode class MarkdownExtractor: def __init__(self): self.md MarkdownIt(commonmark, {html: False}).enable(table) def extract(self, filepath: str): with open(filepath, r, encodingutf-8) as f: source f.read() tokens self.md.parse(source) root SyntaxTreeNode(tokens) sections [] current_section None heading_path [] def inline_text(node): # 提取 inline 节点的纯文本 if node.type inline: return node.content return def walk(node): nonlocal current_section if node.type heading: # 更新标题路径 level int(node.tag[1]) text .join(child.content for child in node.children if child.type inline) text re.sub(r\s, , text).strip() # 截断当前 section开始新的 if current_section and current_section.get(content): sections.append(current_section) while len(heading_path) level: heading_path.pop() heading_path.append(text) current_section { heading: / .join(heading_path), level: level, content: , code_blocks: [], tables: [], images: [], } elif current_section is not None: if node.type fence: # 代码块 lang node.info code node.content current_section[code_blocks].append({lang: lang, code: code}) current_section[content] f\n[代码块: {lang}]\n elif node.type table: # 表格单独存储 rows [] for row_node in node.children: if row_node.type tr: cells [] for cell_node in row_node.children: cells.append(.join(child.content for child in cell_node.children if child.type inline)) rows.append(cells) current_section[tables].append(rows) current_section[content] \n[表格]\n elif node.type inline: text inline_text(node) # 抽图片路径 for m in re.finditer(r!\[([^\]]*)\]\(([^)])\), text): current_section[images].append({alt: m.group(1), src: m.group(2)}) # 去图片语法保留纯文本 text re.sub(r!\[([^\]]*)\]\(([^)])\), [图片], text) text re.sub(r[*_]{1,3}([^*_])[*_]{1,3}, r\1, text) # 粗体斜体降级 text re.sub(r\[([^\]])\]\(([^)])\), r\1(\2), text) # 链接转文本(url) current_section[content] text \n for child in node.children: walk(child) walk(root) if current_section and current_section.get(content): sections.append(current_section) return { filepath: filepath, title: sections[0][heading].split( / )[0] if sections else None, sections: sections, }这段代码的核心思路通过 AST 遍历把 Markdown 还原成一个一个有标题路径的 section代码块和表格单独抽取出来。正文里保留“代码块”“表格”“图片”这些占位符而不是真正的内容。这样做的好处是分块时正文很干净代码和表格又不会丢。4.4 分块与元数据组装拿到 sections 之后接下来就是把它喂给分块器。这里我会针对每个 section 独立分块而不是把整篇文档混在一起切。因为 section 本身就是一个语义单元按它切分天然比按固定字符切要稳定。def chunk_sections(sections: list, max_chars: int 800, overlap: int 80): chunks [] for sec in sections: text sec[content].strip() if not text: continue # 长 section 二次切分 if len(text) max_chars: chunks.append({ text: text, heading: sec[heading], meta: { level: sec[level], code_blocks: len(sec[code_blocks]), tables: len(sec[tables]), } }) else: # 按句子拆 sentences re.split(r(?[。]), text) tmp for sent in sentences: if len(tmp) len(sent) max_chars: tmp sent else: if tmp: chunks.append({ text: tmp, heading: sec[heading], meta: {level: sec[level]} }) tmp sent if tmp: chunks.append({text: tmp, heading: sec[heading], meta: {level: sec[level]}}) # 最终 chunk 序号和源文件 for i, chunk in enumerate(chunks): chunk[meta][chunk_id] i return chunks元数据设计有一个我特别想强调的点heading 一定要存成完整路径而不是只存最后一级标题。比如“3.2 用 AST 还是正则”这个 section它的完整路径是“RAG 数据导入与解析全攻略一/ 3. Markdown 解析 / 3.2 用 AST 还是正则”。检索命中后回答里引用来源时只要把这个路径展示出来用户立刻知道这段内容在全文里的位置。只存最后一级标题的话经常出现“3.2”这种光秃秃的引用非常不专业。5. 常见问题与排查技巧实录下面这些问题是这几年来做 txt 和 Markdown 导入时反复遇到的直接整理成速查表。现象可能原因解决方案读取 txt 报编码错误文件是 GBK/GB18030 或混合编码先用 chardet 探测尝试 utf-8 - gb18030 - big5 逐级解码读取成功但全是乱码chardet 探测置信度低选错了编码增加乱码字符占比校验特别是\ufffd、生僻字比例分块后检索匹配不到完整语义固定字符数硬切切断句子或段落改为按段落/句子切块段落长时再二次切分Markdown 的标题识别错乱代码块内部的#被当作标题改用 AST 遍历不能只靠正则匹配行首表格内容被切成碎片表格行在分块时被截断把每个 table 节点作为一个整体 chunk不参与正文分块代码块被当成正文反引号语法解析不完整使用fence节点类型单独抽取加上语言标签图片引用导致正文很脏图片路径和 alt 文本混入正文抽取图片路径进元数据正文替换为[图片]占位符回答引用来源没有章节信息只存了标题没存标题路径保存完整的 heading_path逐级拼接一个大 section 超过模型 token单个章节内容过长按句号/分号二次切分同时保留 heading 作为前缀除了速查表还有两个值得单独说的经验第一个是 Markdown 里嵌套列表的处理。很多解析器对嵌套列表支持不友好。我在 AST 遍历时如果遇到bullet_list和ordered_list会把它当成一个整体块处理提取每一行的内容但保留层级缩进信息用 两个空格表示一级。这样既不会破坏列表的内部关系也可以当成文本入库。千万不要把列表的每一行单独切出去那样很容易把“步骤 1、步骤 2、步骤 3”这套递进关系拆散。第二个是重复文档去重。同一个 Markdown 文件可能在你的数据源中出现多份比如不同目录下的副本。我入库前会对每个 section 拼接后的文本算一个 md5用 section 的 heading_path content_hash 作为唯一键向量库里已经存在就直接跳过。这个操作简单但非常有用我见过很多团队检索结果里出现两条一模一样的答案就是没做去重导致的。写在最后的一点体会这几年的经验让我形成了一个习惯每接手一批新数据先花半小时人工看两三个典型文件再决定解析策略。txt 要看它是什么编码、有没有页眉页脚、段落是不是被硬换行切断Markdown 要看它标题层级是否规范、代码块是单行还是多行、表格是不是真的用管道符对齐的。看一遍再写解析规则基本能躲掉 80% 的坑。解析不是一个“跑通就算完”的环节它是整个 RAG 项目里最需要反复迭代的部分。你永远不知道下一批数据里会遇到什么奇怪的格式。所以我的建议是从第一篇文档就开始做全链路验证——解析、分块、入库、检索、问答每一步都跑一遍确认能回答出“这段内容出自哪一节、原文长什么样”再批量导入。下一篇我准备聊聊带表格的文档怎么处理那种从 PDF 里提取出来、特别容易散架的表格处理起来其实比 Markdown 表格刺激多了。这篇的内容如果你在实操中碰到卡住的点拿不准怎么搞的可以先按照上面速查表对着排查一遍多数问题都能当场定位。
返回列表