☰
RAG数据导入实战:txt与Markdown解析、编码处理与语义分块
2026/10/6 5:16:19 网站建设 项目流程

1. 为什么 RAG 的第一道坎永远是数据导入

做 RAG 的人都有一个共识:模型选型、向量库选型、检索策略调优这些事,网上的教程一抓一大把,但真正让人在项目里卡住的,往往是看起来最不起眼的一步——把原始数据喂进去。我见过太多团队,向量数据库搭好了,Embedding 模型也调通了,结果卡在"这批 txt 文件怎么批量读进来"这种问题上,一卡就是两三天。

这个现象背后的原因其实不复杂。RAG 的数据导入和传统的数据 ETL 有本质区别。传统 ETL 处理的是结构化数据,字段对齐、类型转换、去重清洗,套路非常成熟。但 RAG 面对的是非结构化文本,而且这些文本的来源极其杂乱:有从网页复制粘贴的、有从 PDF 转出来的、有从数据库导出的、有从聊天记录里扒出来的。它们的编码格式、换行符、段落结构、特殊字符各不相同,直接扔给 Embedding 模型,出来的向量质量会差得离谱。

更关键的是,RAG 的检索质量高度依赖文本块的语义完整性。如果你把一个完整的段落按固定字数硬切,切出来的块可能前半句在讲 A 概念、后半句在讲 B 概念,检索时两个概念都匹配不上。所以数据导入这一步,不只是"读文件"那么简单,它包含了格式识别、编码处理、结构解析、语义分块四个层次的工作。

这一篇我先聚焦最基础也最通用的场景:纯文本 txt 和 Markdown 文件的导入与解析。这两类格式看起来简单,但恰恰是坑最多的。txt 没有结构信息,全靠你自己推断;Markdown 有结构但语法灵活,解析器选不对就会丢信息。把这两类吃透,后面处理 PDF、Word、HTML 就有底子了。

提示:本文所有代码基于 Python 3.10+ 编写,核心依赖是chardet、markdown-it-py、langchain-text-splitters。如果你用的是其他语言栈,原理部分同样适用,只是 API 调用方式不同。

2. txt 文件导入:编码识别是第一道生死关

2.1 为什么open()直接读会翻车

很多人导入 txt 的第一反应是open('file.txt', 'r').read(),然后发现报UnicodeDecodeError,或者读出来的中文全是乱码。这不是代码写错了,而是编码问题。txt 文件本身不携带编码信息,它就是一串字节,具体怎么解释这串字节,取决于创建它的软件和系统。

Windows 上记事本默认用 GBK(更准确地说是 GB2312 或 GB18030),macOS 和 Linux 默认用 UTF-8,而从某些老系统导出的文件可能是 GBK、Big5、甚至 Latin-1。你用一个固定编码去读所有文件,必然有一批会翻车。

我踩过最坑的一次是:一批文件里混了 UTF-8 和 GBK 两种编码,用 UTF-8 读 GBK 文件时,Python 默认会抛异常;但如果加了errors='ignore',它会静默丢掉所有无法解码的字节,结果就是读出来的文本缺字少句,你还以为是原文就这样。这种问题在检索阶段才会暴露——用户问的问题明明在文档里,就是检索不到,排查半天才发现是导入时丢了内容。

2.2 用 chardet 做编码探测的正确姿势

chardet是 Python 里最常用的编码探测库,原理是基于字节序列的统计特征来猜测编码。它的用法很简单:

import chardet def detect_encoding(file_path, sample_size=100000): with open(file_path, 'rb') as f: raw = f.read(sample_size) result = chardet.detect(raw) return result['encoding'], result['confidence']

但这里有几个实操细节必须注意。第一,采样大小要合理。默认chardet.detect只读前几 KB,对于开头是英文、后面才是中文的文件,探测结果可能不准。我一般采样 100KB,兼顾准确率和速度。第二,confidence 低于 0.7 时要警惕。这种情况下探测结果不可靠,需要走降级策略。第三,chardet 对短文本的探测极不靠谱。如果文件只有几百字节,它可能把 UTF-8 猜成 Windows-1252,这时候需要结合文件来源做人工判断。

一个更稳妥的做法是维护一个编码优先级列表,按顺序尝试解码,哪个成功用哪个:

ENCODING_PRIORITY = ['utf-8', 'gb18030', 'big5', 'latin-1'] def robust_read(file_path): with open(file_path, 'rb') as f: raw = f.read() # 先用 chardet 探测 detected = chardet.detect(raw[:100000]) if detected['confidence'] > 0.7 and detected['encoding']: try: return raw.decode(detected['encoding']) except UnicodeDecodeError: pass # 降级:按优先级尝试 for enc in ENCODING_PRIORITY: try: return raw.decode(enc) except UnicodeDecodeError: continue # 最后兜底:忽略错误 return raw.decode('utf-8', errors='replace')

注意gb18030而不是gbk。gb18030 是 gbk 的超集,能覆盖更多生僻字,用 gb18030 解码 gbk 文件完全没问题,反过来则可能失败。至于latin-1,它能解码任何字节序列(因为它是单字节映射),所以放在最后兜底,但用它解出来的中文肯定是乱码,只适合纯英文文件。

2.3 换行符与不可见字符的清洗

编码搞定之后,下一个坑是换行符。Windows 用\r\n,Unix 用\n,老 Mac 用\r。Python 的open()在文本模式下会自动做 universal newlines 转换,但如果你用二进制模式读再自己 decode,就得手动处理。统一替换成\n是最省事的做法:

text = text.replace('\r\n', '\n').replace('\r', '\n')

比换行符更隐蔽的是不可见字符。从网页复制的文本经常带着零宽空格(\u200b)、零宽连字符(\u200d)、不换行空格(\u00a0)、字节顺序标记(\ufeff)。这些字符肉眼看不见,但会污染 Embedding 结果。特别是\ufeff,它经常出现在 UTF-8 文件的开头(BOM),如果你不处理,第一个文本块的开头就会多一个莫名其妙的字符。

我一般用一个正则统一清洗:

import re def clean_invisible(text): # 移除零宽字符和 BOM text = re.sub(r'[\u200b\u200c\u200d\ufeff]', '', text) # 不换行空格转普通空格 text = text.replace('\u00a0', ' ') # 连续 3 个以上换行压缩为 2 个 text = re.sub(r'\n{3,}', '\n\n', text) # 行尾空白清理 text = re.sub(r'[ \t]+\n', '\n', text) return text

这里把连续 3 个以上换行压成 2 个,是因为很多 txt 文件用多个空行做视觉分隔,但对 RAG 来说,超过 2 个空行没有语义价值,反而会让分块器产生空块。

2.4 txt 的语义分块:没有结构就自己造结构

txt 最大的问题是没有结构信息。Markdown 有#标题,HTML 有标签,PDF 有版面分析,但 txt 就是纯文字。你只能靠启发式规则来推断结构。

我常用的启发式规则有这么几条。第一,空行分隔的段落是最基本的语义单元,优先按空行切。第二,短行且以特定符号开头的可能是标题,比如以"第X章"、"一、"、"1."、"【"开头的行。第三,连续多行长度相近且以标点结尾的可能是正文段落。第四,行首有大量空格的可能是代码块或引用。

基于这些规则,我写了一个简单的结构推断函数:

def infer_structure(text): lines = text.split('\n') blocks = [] current_block = [] for line in lines: stripped = line.strip() # 空行作为块分隔 if not stripped: if current_block: blocks.append('\n'.join(current_block)) current_block = [] continue # 标题特征检测 is_heading = ( len(stripped) < 50 and re.match(r'^(第[一二三四五六七八九十百]+[章节篇]|[一二三四五六七八九十]+、|\d+[\.、]|【.+】)', stripped) ) if is_heading and current_block: blocks.append('\n'.join(current_block)) current_block = [stripped] else: current_block.append(stripped) if current_block: blocks.append('\n'.join(current_block)) return blocks

这个函数把文本切成"块",每个块要么是一个标题,要么是一个段落。切完之后,再根据块的长度决定是否要进一步细分。如果某个块超过 500 字,就用递归字符分块器再切;如果小于 50 字且是标题,就把它和后面的内容合并成一个语义单元。

注意:启发式规则永远不可能 100% 准确,特别是面对格式混乱的 txt。我的经验是,宁可切得粗一点,也不要切得太碎。一个 800 字的块,只要语义完整,检索效果往往比 4 个 200 字的碎块更好。因为 Embedding 模型对长文本的语义表征能力比短文本强,而且检索时返回一个完整段落比返回四个碎片更有用。

3. Markdown 解析:别用正则,用真正的解析器

3.1 正则解析 Markdown 的三大翻车现场

我见过太多人用正则表达式解析 Markdown,比如re.findall(r'^#+\s+(.+)$', text, re.MULTILINE)来提取标题。这种做法在简单文档上能跑通,但一遇到复杂文档就崩。翻车现场主要有三个。

第一个是代码块里的#。Markdown 的代码块用三个反引号包裹,里面的内容原样保留。如果你的正则不区分代码块内外,就会把代码里的注释# 这是注释当成标题提取出来。第二个是行内代码里的特殊字符。比如`# 这不是标题`这种行内代码,正则同样会误判。第三个是嵌套结构。Markdown 支持列表嵌套、引用嵌套、列表里嵌代码块,正则根本处理不了这种层级关系。

正确的做法是用真正的 Markdown 解析器。Python 生态里主流的有markdown、markdown-it-py、mistune三个。我推荐markdown-it-py,理由是它遵循 CommonMark 规范,支持插件扩展,而且能输出结构化的 token 流,方便你做后续处理。

3.2 markdown-it-py 的 token 流处理

markdown-it-py把 Markdown 解析成一棵 token 树,每个 token 有type、tag、content、level等属性。你可以遍历这棵树,按需提取结构信息:

from markdown_it import MarkdownIt md = MarkdownIt() tokens = md.parse(markdown_text) def extract_structure(tokens): structure = [] current_heading = None current_content = [] for token in tokens: if token.type == 'heading_open': # 保存上一个章节 if current_heading or current_content: structure.append({ 'heading': current_heading, 'content': '\n'.join(current_content) }) current_heading = None current_content = [] elif token.type == 'heading_close': continue elif token.type == 'inline' and current_heading is None: # 判断是否是标题内容 prev_token = tokens[tokens.index(token) - 1] if prev_token.type == 'heading_open': current_heading = token.content else: current_content.append(token.content) elif token.type == 'fence': # 代码块,保留原始内容 current_content.append(f"```{token.info}\n{token.content}```") elif token.type == 'code_block': current_content.append(f"```\n{token.content}```") if current_heading or current_content: structure.append({ 'heading': current_heading, 'content': '\n'.join(current_content) }) return structure

这段代码把 Markdown 按标题切分成章节,每个章节包含标题和正文。代码块被特殊处理,保留了原始格式。这样切出来的章节,语义完整性比按字数硬切好得多。

3.3 标题层级与章节合并策略

Markdown 的标题有六级(#到######),但实际文档里常用的就前三级。切分时有个关键决策:按几级标题切?

如果按一级标题切,章节可能太大,一个章节几千字,超过 Embedding 模型的上下文窗口。如果按三级标题切,章节可能太碎,一个三级标题下只有一两句话。我的经验是动态决定:先按一级标题切,如果某个章节超过 1000 字,再按二级标题切;如果还超过 1000 字,再按三级标题切。这样能保证每个块的大小在合理范围内。

还有一个细节是标题的继承。如果一个二级标题下的内容被切成了多个块,每个块都应该带上完整的标题路径,比如一级标题 > 二级标题 > 三级标题。这样检索时,即使只匹配到某个块,也能知道它在文档中的位置,方便做上下文扩展。

def build_heading_path(heading_stack): return ' > '.join([h for h in heading_stack if h]) def split_by_heading(structure, max_chars=1000): chunks = [] heading_stack = [] for section in structure: heading = section['heading'] content = section['content'] if heading: # 根据标题级别维护栈 level = heading.count('#') if heading.startswith('#') else 1 heading_stack = heading_stack[:level-1] heading_stack.append(heading.lstrip('#').strip()) path = build_heading_path(heading_stack) if len(content) <= max_chars: chunks.append({ 'heading_path': path, 'content': content }) else: # 超长内容进一步切分 sub_chunks = recursive_split(content, max_chars) for sub in sub_chunks: chunks.append({ 'heading_path': path, 'content': sub }) return chunks

3.4 表格、公式、代码块的特殊处理

Markdown 里有三类内容需要特殊对待:表格、数学公式、代码块。

表格在 Markdown 里用|分隔,解析后是一个二维结构。如果直接当普通文本处理,|和---会污染语义。我的做法是把表格转成自然语言描述,比如把| 姓名 | 年龄 |转成"姓名:张三,年龄:25"。这样 Embedding 时能更好地理解表格内容。

数学公式分两种:行内公式$...$和块级公式$$...$$。行内公式一般保留原样,因为它在句子里;块级公式建议单独成块,并在前后加上说明文字,比如"公式:..."。如果公式很重要,可以考虑用专门的工具转成 LaTeX 描述文本。

代码块的处理最讲究。代码本身对 Embedding 不友好,因为它的语义和自然语言差异太大。但如果直接丢掉代码块,又会丢失重要信息。我的策略是保留代码块但加上语言标注和上下文说明。比如:

# 原始代码块 def hello(): print("world") # 处理后 代码示例(Python): def hello(): print("world")

这样 Embedding 时,模型至少能知道这是一段 Python 代码,而不是随机字符。

4. 从 txt 到 Markdown:格式转换的取舍

4.1 为什么要把 txt 转成 Markdown

有人会问:txt 直接读进来不就行了,为什么要转成 Markdown?这个问题问得好。转 Markdown 的核心价值在于结构显式化。

txt 里的结构是隐式的,靠空行、缩进、符号来暗示。这些暗示对人类读者有效,但对程序来说很模糊。转成 Markdown 后,结构变成了显式的#、-、>,程序可以精确解析。而且 Markdown 是 RAG 生态里最友好的格式,几乎所有分块器、解析器都原生支持它。

更重要的是,Markdown 的标题层级天然对应语义层级。一个##标题下的内容,语义上就是一个完整的主题。按标题切分,切出来的块语义完整性最好。这是按字数硬切永远达不到的效果。

4.2 txt 转 Markdown 的启发式规则

txt 转 Markdown 没有标准答案,全靠启发式规则。我总结了一套在实践中比较有效的规则:

txt 特征转换规则示例
以"第X章"、"第X节"开头转为#或##"第一章 概述" →# 第一章 概述
以"一、"、"二、"开头转为##"一、背景" →## 一、背景
以"1."、"1.1"开头转为###"1.1 目标" →### 1.1 目标
以"【"开头转为##"【简介】" →## 简介
以"-"、"*"、"·"开头转为列表项"- 要点" →- 要点
连续缩进 4 空格以上转为代码块缩进内容 →```包裹
空行分隔的段落保持段落段落间加空行

这套规则不是万能的,但能覆盖 80% 的常见 txt 格式。剩下的 20% 需要根据具体文档调整。我的建议是先跑一遍自动转换,再人工抽查。抽查时重点看标题识别是否准确、段落切分是否合理、有没有把正文误判成标题。

4.3 转换后的质量校验

转换完不能直接用,必须做质量校验。我一般检查这几个指标:

标题数量与预期是否匹配。如果原文有 10 个章节,转换后应该至少有 10 个标题。如果只有 3 个,说明规则没覆盖到。段落平均长度是否合理。如果平均段落超过 2000 字,说明切分不够细;如果平均段落少于 50 字,说明切得太碎。代码块和表格是否正确识别。这两类内容如果被误判成正文,会严重影响检索质量。

校验的代码可以这样写:

def validate_markdown(md_text): lines = md_text.split('\n') headings = [l for l in lines if l.startswith('#')] paragraphs = [l for l in lines if l.strip() and not l.startswith('#')] avg_para_len = sum(len(p) for p in paragraphs) / max(len(paragraphs), 1) report = { 'heading_count': len(headings), 'paragraph_count': len(paragraphs), 'avg_paragraph_length': avg_para_len, 'has_code_block': '```' in md_text, 'has_table': '|' in md_text and '---' in md_text } # 质量判断 warnings = [] if avg_para_len > 2000: warnings.append('段落过长,建议进一步切分') if avg_para_len < 50: warnings.append('段落过短,可能切分过碎') if len(headings) == 0: warnings.append('未识别到标题,检查转换规则') report['warnings'] = warnings return report

这个报告能帮你快速判断转换质量。如果 warnings 里有内容,就回去调整规则。

5. 分块策略:语义完整性与检索效果的平衡

5.1 固定长度分块为什么不够用

最简单的分块策略是固定长度,比如每 500 字切一块。这种做法实现简单,但效果很差。原因有两个。

第一,它会在句子中间切断。一个完整的句子被切成两半,前半句在块 A,后半句在块 B,检索时两个块都匹配不上完整语义。第二,它忽略了文档结构。一个章节的内容被硬切后,标题和正文分离,检索时可能只匹配到正文,丢失了标题提供的上下文。

我做过一个对比实验:同一批文档,一组用固定长度分块,一组用语义分块,然后用相同的问题做检索。语义分块的召回率比固定长度高 30% 以上。这个差距在 RAG 项目里是致命的。

5.2 递归字符分块器的参数调优

langchain-text-splitters里的RecursiveCharacterTextSplitter是目前最常用的分块器。它的核心思想是按优先级尝试不同的分隔符,先尝试按段落切,如果块还是太大,再按句子切,再按逗号切,最后才按字符切。

关键参数有三个:chunk_size、chunk_overlap、separators。

chunk_size是每块的最大字符数。这个值取决于你的 Embedding 模型。OpenAI 的text-embedding-ada-002支持 8191 token,但实际使用时建议控制在 500-1000 字符,因为太长的块会稀释语义。中文的话,500-800 字符比较合适。

chunk_overlap是相邻块的重叠字符数。重叠的目的是避免边界信息丢失。比如一个关键句子正好在切分点上,如果没有重叠,它可能被切成两半;有重叠的话,至少有一个块包含完整句子。一般设置为chunk_size的 10%-20%。

separators是分隔符优先级列表。中文文档我一般这样设置:

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=100, separators=[ '\n## ', # 二级标题 '\n### ', # 三级标题 '\n\n', # 段落 '\n', # 换行 '。', # 中文句号 '!', # 中文感叹号 '?', # 中文问号 ';', # 中文分号 ',', # 中文逗号 ' ', # 空格 '' # 字符 ], length_function=len )

注意分隔符的顺序很重要。\n##放在最前面,意味着优先按二级标题切。这样切出来的块,每个块对应一个二级章节,语义完整性最好。

5.3 标题感知分块与父子块策略

比递归分块更高级的是标题感知分块。它的思路是:先用 Markdown 解析器提取标题结构,然后按标题切分,每个块带上完整的标题路径。这样检索时,即使只匹配到正文,也能通过标题路径知道上下文。

更进一步的是父子块策略。它的思路是:把文档切成大块(父块)和小块(子块),子块用于检索,父块用于生成。检索时用子块匹配,匹配到后返回对应的父块给 LLM。这样既保证了检索精度(子块小,语义集中),又保证了生成质量(父块大,上下文完整)。

def build_parent_child_chunks(md_text, parent_size=2000, child_size=400): # 先按标题切父块 parent_chunks = split_by_heading(md_text, max_chars=parent_size) result = [] for i, parent in enumerate(parent_chunks): # 每个父块再切成子块 child_splitter = RecursiveCharacterTextSplitter( chunk_size=child_size, chunk_overlap=50 ) children = child_splitter.split_text(parent['content']) for child in children: result.append({ 'parent_id': i, 'parent_content': parent['content'], 'heading_path': parent['heading_path'], 'child_content': child }) return result

这个策略在长文档场景下效果特别好。用户问一个具体问题,子块能精确匹配;返回给 LLM 时,父块提供了完整的上下文,生成的答案更准确。

5.4 分块效果的评估方法

分块做完了,怎么知道好不好?我一般用三个方法评估。

人工抽查。随机抽 20 个块,看它们是否语义完整、是否包含标题路径、是否有明显的切断。这个方法最直接,但费时间。

检索测试。准备 20 个问题,每个问题对应文档里的一个具体知识点,看检索能否命中正确的块。命中率低于 80% 就说明分块有问题。

块大小分布。统计所有块的长度,看分布是否合理。如果大部分块都在chunk_size附近,说明分块器在硬切;如果分布比较分散,说明它尊重了文档结构。

import statistics def analyze_chunks(chunks): lengths = [len(c['content']) for c in chunks] return { 'count': len(chunks), 'mean': statistics.mean(lengths), 'median': statistics.median(lengths), 'stdev': statistics.stdev(lengths) if len(lengths) > 1 else 0, 'min': min(lengths), 'max': max(lengths) }

如果stdev很小,说明块大小很均匀,可能是硬切;如果stdev较大,说明分块器在按结构切。理想情况下,mean应该在chunk_size的 60%-80% 之间,max不超过chunk_size的 1.5 倍。

6. 实操中那些文档不会告诉你的坑

6.1 大文件的内存问题

处理大文件时,f.read()会把整个文件加载到内存。如果文件有几百 MB,内存直接爆掉。正确的做法是流式读取:

def stream_read(file_path, chunk_size=1024*1024): with open(file_path, 'r', encoding='utf-8') as f: while True: chunk = f.read(chunk_size) if not chunk: break yield chunk

但流式读取有个问题:编码探测需要采样,而采样需要读取文件开头。所以流程是:先读前 100KB 做编码探测,然后用探测到的编码流式读取整个文件。

6.2 编码探测的性能开销

chardet的探测速度不快,处理 100KB 数据大概需要 50-100ms。如果你有几千个文件,光编码探测就要几分钟。优化方法是缓存探测结果。同一个来源的文件,编码通常是一样的。你可以按目录或按来源分组,每组只探测第一个文件,后面的复用结果。

import os from functools import lru_cache @lru_cache(maxsize=128) def detect_encoding_cached(dir_path): # 取目录下第一个文件做探测 files = [f for f in os.listdir(dir_path) if f.endswith('.txt')] if not files: return 'utf-8' return detect_encoding(os.path.join(dir_path, files[0]))[0]

6.3 Markdown 解析器的选择陷阱

markdown-it-py默认遵循 CommonMark 规范,但很多中文文档用的是扩展语法,比如 GitHub Flavored Markdown(GFM)的表格、任务列表、删除线。如果你不启用这些扩展,解析时会丢内容。

启用 GFM 扩展的方法:

from markdown_it import MarkdownIt md = MarkdownIt('gfm-like') # 或者手动启用 md = MarkdownIt().enable(['table', 'strikethrough', 'tasklist'])

另外,markdown-it-py默认不解析 HTML 标签。如果你的 Markdown 里嵌了 HTML,需要启用html选项。但启用 HTML 有安全风险(XSS),如果文档来源不可信,建议先做 sanitize。

6.4 分块边界的语义断裂

即使做了语义分块,边界处仍然可能断裂。比如一个概念的定义在块 A 的末尾,例子在块 B 的开头,检索时只匹配到块 B,就缺少了定义。解决方法是在块的开头加上上一块的末尾内容作为上下文,或者用滑动窗口的方式做重叠。

我一般用chunk_overlap来处理这个问题,但重叠不是越多越好。重叠太多会导致重复内容,检索时返回多个相似块,浪费上下文窗口。10%-20% 的重叠是比较平衡的选择。

6.5 特殊字符导致的 Embedding 异常

有些特殊字符会让 Embedding 模型报错或产生异常向量。常见的有:控制字符(\x00-\x1f)、代理对(surrogate pair)、超长连续字符(比如 1000 个连续的a)。导入前最好做一次清洗:

def sanitize_for_embedding(text): # 移除控制字符(保留 \n \t) text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text) # 压缩超长连续字符 text = re.sub(r'(.)\1{50,}', r'\1' * 10, text) # 移除代理对 text = text.encode('utf-8', errors='ignore').decode('utf-8') return text

这个清洗步骤看起来不起眼,但能避免很多莫名其妙的报错。我遇到过好几次 Embedding API 返回 400 错误,排查半天发现是文档里有个控制字符。

7. 一个完整的导入解析流程

把前面所有内容串起来,一个完整的 txt/Markdown 导入解析流程是这样的:

第一步,文件扫描。遍历目录,按扩展名分类,txt 和 md 分开处理。

第二步,编码处理。对 txt 文件做编码探测,用探测结果读取;Markdown 文件默认 UTF-8,失败则降级。

第三步,内容清洗。统一换行符,移除不可见字符,压缩连续空行,sanitize 特殊字符。

第四步,格式转换。txt 按启发式规则转 Markdown;Markdown 保持原样。

第五步,结构解析。用markdown-it-py解析 Markdown,提取标题层级和章节内容。

第六步,语义分块。按标题切分,超长章节用递归分块器细分,加上标题路径。

第七步,质量校验。检查块大小分布、标题识别率、特殊内容处理情况。

第八步,输出。把块存成 JSON 或直接写入向量库。

def full_pipeline(file_path): # 1. 读取 if file_path.endswith('.txt'): text = robust_read(file_path) text = txt_to_markdown(text) else: text = robust_read(file_path) # 2. 清洗 text = clean_invisible(text) text = sanitize_for_embedding(text) # 3. 解析 md = MarkdownIt('gfm-like') tokens = md.parse(text) structure = extract_structure(tokens) # 4. 分块 chunks = split_by_heading(structure, max_chars=800) # 5. 校验 report = validate_markdown(text) return chunks, report

这套流程我在多个项目里跑过,处理几万个文件没问题。关键是每一步都要有降级策略,不能因为一个文件出错就中断整个流程。

8. 一些个人经验

做 RAG 数据导入这几年,最大的体会是:数据质量决定 RAG 上限。模型再强、检索策略再优,如果导入的文本本身就是乱的,结果一定好不了。我见过太多团队在模型和检索上花大量时间,却对数据导入敷衍了事,最后效果上不去,还找不到原因。

另一个体会是:不要追求一步到位。数据导入是个迭代过程。第一版先把文件读进来,能跑通就行;第二版加上编码处理和清洗;第三版做语义分块;第四版优化分块参数。每迭代一次,检索效果都会提升。想一次做到完美,往往什么都做不好。

最后分享一个小技巧:建立数据质量监控。每次导入后,记录文件数、总字数、平均块大小、标题识别率等指标。如果某个指标突然异常,说明这批数据有问题。这个监控能帮你及早发现问题,避免脏数据污染整个知识库。

下一篇我会讲 PDF 和 Word 的导入解析,那又是另一个坑坑洼洼的领域。PDF 的版面分析、表格提取、扫描件 OCR,每一个都能单独写一篇。如果你正在做 RAG 项目,建议先把 txt 和 Markdown 这两类吃透,它们是基础,也是最能体现数据质量差距的地方。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询