1. 为什么文本导入是 RAG 系统最容易被低估的一环
做 RAG 的人都有一个共识:模型选型、向量库选型、检索策略,这些话题热度高、讨论多,但真正让一个 RAG 系统在演示阶段就翻车的,往往是最不起眼的数据导入环节。我见过太多团队,花了两周调 embedding 模型,结果发现原始文档里全是乱码、页眉页脚、断行错位,检索出来的内容驴唇不对马嘴。
这个系列的第一篇,我想把最基础但也最容易被跳过的一块讲透:纯文本(txt)和 Markdown 这两类文档,怎么导入、怎么解析、怎么结构化。别小看这两种格式,它们恰恰是 RAG 知识库里占比最高的数据源——技术文档、产品手册、会议纪要、个人笔记,导出成 txt 或 Markdown 的情况太常见了。
先说清楚这篇内容适合谁看。如果你刚开始搭 RAG 知识库,还在纠结"我的 txt 文件怎么切分才合理";如果你已经跑通了 demo,但发现检索质量忽高忽低,怀疑是数据导入的问题;如果你在做文档结构化解析,想找一套可复用的处理流程——那这篇就是写给你的。我会从设计思路讲到具体代码,从参数计算讲到踩坑记录,尽量做到你读完就能直接抄作业。
核心关键词先摆出来:RAG、数据导入、解析、txt、Markdown、文档结构化解析。这几个词贯穿全文,后面每个章节都会围绕它们展开。
2. 整体设计思路:txt 和 Markdown 到底难在哪
2.1 两种格式的本质差异决定了处理策略
很多人觉得 txt 最简单,没有格式,直接读进来就行。这个想法对了一半。txt 确实没有显式结构,但"没有结构"本身就是最大的问题——你拿到的 txt 可能是一整篇没有空行的长文,也可能是每行都断开的诗歌,还可能是从 PDF 复制出来带着一堆多余空格的乱码。txt 的难点在于结构全靠推断。
Markdown 则相反,它有显式的结构标记:#是标题,-是列表,```是代码块,|是表格。但 Markdown 的难点在于结构标记和语义内容混在一起,而且不同人写的 Markdown 规范程度差异极大。有人用#后面不加空格,有人用===做标题下划线,有人表格不对齐,有人代码块不闭合。这些都会让解析器翻车。
所以整体设计思路的第一条原则是:txt 走"推断结构"路线,Markdown 走"解析结构"路线,但两者最终都要归一化成同一种中间表示。这个中间表示我推荐用带层级信息的块(chunk)列表,每个块包含内容、类型、层级、来源位置四个字段。
2.2 为什么选择"先解析后切分"而不是"先切分后解析"
这是我在实际项目里踩过的一个大坑。早期我图省事,直接把整个 txt 按固定字数切分,然后丢进向量库。结果检索出来的片段经常从句子中间断开,或者把标题和正文切散。后来改成先解析出结构,再按结构边界切分,检索质量立刻上了一个台阶。
具体来说,先解析后切分的好处有三个。第一,标题、列表、代码块这些结构单元天然就是语义边界,按它们切分不会破坏语义完整性。第二,解析阶段可以顺便做清洗,比如去掉页眉页脚、合并断行、修正编码。第三,解析后的块可以携带层级信息,检索时能根据层级做加权,比如标题块的权重高于正文块。
代价是解析阶段需要写更多代码,处理更多边界情况。但这个投入绝对值得,因为数据导入是一次性的,检索质量是长期的。
2.3 归一化中间表示的设计
我用的中间表示是一个 Python 字典列表,每个字典长这样:
{ "content": "这是块的内容", "type": "heading", # heading / paragraph / list / code / table "level": 2, # 标题层级,非标题为 0 "source": "doc.md", "position": 15 # 在原文档中的行号或字符偏移 }这个设计的关键在于type和level两个字段。type决定了后续切分策略,比如代码块不切分,段落按句子切分。level决定了层级关系,可以用来构建文档树,也可以用来做检索加权。
提示:
position字段看起来不起眼,但在排查问题时极其有用。当检索结果不对劲时,你可以快速定位到原文档的哪一行,判断是解析错了还是切分错了。
3. txt 文件解析:从无结构到有结构的推断方法
3.1 编码检测:第一步就卡住的人不在少数
txt 文件最常见的翻车点就是编码。中文 txt 可能是 UTF-8、GBK、GB2312、GB18030,甚至还有 UTF-8 with BOM。你直接用open(file, 'r')读,遇到编码不对就抛异常或者读出乱码。
我的做法是用chardet库先检测编码,检测置信度低于 0.8 的再用charset-normalizer二次确认。实测下来,这两个库组合使用,中文 txt 的编码识别准确率能到 95% 以上。
import chardet from charset_normalizer import from_path def detect_encoding(file_path): with open(file_path, 'rb') as f: raw = f.read(100000) # 只读前 100KB,加快速度 result = chardet.detect(raw) if result['confidence'] > 0.8: return result['encoding'] # 置信度低,用 charset-normalizer 二次确认 best = from_path(file_path).best() return best.encoding if best else 'utf-8'这里有个细节:只读前 100KB 做检测,而不是读整个文件。因为大文件的编码检测很慢,而前 100KB 通常足够判断编码。如果文件前 100KB 全是英文,检测可能不准,这时候可以再读中间一段做二次检测。
注意:检测出编码后,读取时一定要加
errors='replace'或errors='ignore',避免个别非法字节导致整个文件读取失败。虽然会损失一点内容,但比整个文件读不进来强。
3.2 断行合并:中文 txt 最大的坑
从 PDF 或网页复制出来的中文 txt,经常出现这种情况:一个完整的句子被硬生生断成好几行,每行末尾没有标点。如果你直接按行切分,语义就碎了。
判断是否需要合并断行的逻辑是这样的:如果当前行末尾不是句号、问号、感叹号、分号、冒号等终止标点,且下一行开头不是特殊标记(如数字序号、项目符号),那么这两行应该合并。
import re def merge_lines(lines): merged = [] buffer = "" for line in lines: stripped = line.strip() if not stripped: if buffer: merged.append(buffer) buffer = "" merged.append("") continue if not buffer: buffer = stripped elif re.search(r'[。!?;:.!?;:]$', buffer): merged.append(buffer) buffer = stripped elif re.match(r'^[\d]+[.、))]', stripped) or re.match(r'^[-*•]', stripped): merged.append(buffer) buffer = stripped else: buffer += stripped if buffer: merged.append(buffer) return merged这个逻辑看起来简单,但实际调参很讲究。比如英文 txt 的断行合并规则就不一样,英文单词之间有空格,合并时要在中间加空格。再比如代码类 txt,断行是故意的,不能合并。所以我在实际项目里会先判断文档类型,再选择对应的合并策略。
3.3 标题识别:没有#怎么判断哪行是标题
txt 没有 Markdown 的#标记,标题识别全靠启发式规则。我总结了四条规则,按优先级排序:
第一条,独立成行且长度较短。标题通常不超过 30 个字,且独占一行,前后有空行。
第二条,有编号特征。比如"第一章"、"1.1 节"、"(一)"、"Part 1"这类模式。
第三条,末尾无标点。标题一般不以句号结尾,正文段落通常有句号。
第四条,字体或格式线索。如果 txt 是从 Word 或 PDF 导出的,可能保留了加粗标记(如**标题**)或全角空格缩进。
def is_heading(line, prev_line, next_line): stripped = line.strip() if not stripped or len(stripped) > 40: return False, 0 # 规则一:前后空行 + 短行 if prev_line.strip() == "" and next_line.strip() == "": if len(stripped) <= 30: return True, 2 # 规则二:编号特征 if re.match(r'^(第[一二三四五六七八九十]+[章节部分])', stripped): return True, 1 if re.match(r'^\d+(\.\d+)*[\s、.]', stripped): level = stripped.split()[0].count('.') + 1 return True, level # 规则三:末尾无标点 + 短行 if len(stripped) <= 25 and not re.search(r'[。!?,、;:]$', stripped): return True, 3 return False, 0这套规则不可能 100% 准确,但实测下来对技术文档、产品手册这类结构规整的 txt,准确率能到 85% 左右。剩下的 15% 靠人工抽检修正,或者用更复杂的模型做分类。
实操心得:不要追求 100% 的自动识别准确率。我的做法是自动识别 + 人工抽检 10% 的样本,发现系统性错误就调整规则,个别错误直接手动标注。这样投入产出比最高。
3.4 列表和代码块的识别
txt 里的列表识别相对简单,看行首是否有-、*、•、1.、(1)这类标记。但要注意区分有序列表和无序列表,以及嵌套列表的层级。
代码块识别在 txt 里比较难,因为没有```标记。我的做法是看连续多行是否满足以下特征:行首有统一缩进(4 个空格或 1 个 Tab)、包含代码特征字符(如{}、()、=、;)、行长度分布均匀。满足三条中的两条就判定为代码块。
def is_code_block(lines): if len(lines) < 3: return False indent_pattern = [len(l) - len(l.lstrip()) for l in lines] uniform_indent = len(set(indent_pattern)) <= 2 and min(indent_pattern) >= 2 code_chars = sum(1 for l in lines if re.search(r'[{}()=;<>]', l)) code_ratio = code_chars / len(lines) return uniform_indent and code_ratio > 0.5这个判断逻辑对 Python、JavaScript 这类缩进敏感的代码效果很好,对 C、Java 这类用大括号的代码也还行。但对配置文件、日志文件可能误判,需要根据实际数据调整阈值。
4. Markdown 解析:结构显式但陷阱不少
4.1 为什么不用现成的 Markdown 解析库
你可能会问,Markdown 有那么多现成的解析库,比如 Python 的markdown、mistune,JavaScript 的marked、remark,为什么还要自己写?
原因是:通用 Markdown 解析库的目标是渲染成 HTML,而 RAG 需要的是保留结构信息的块列表。渲染成 HTML 后,你还要再从 HTML 里提取结构,多了一道工序,而且 HTML 的标签嵌套会丢失一些 Markdown 特有的语义,比如标题层级、列表嵌套深度。
我的做法是用mistune的 AST 模式,直接拿到语法树,然后遍历语法树生成块列表。这样既利用了成熟库的解析能力,又能自定义输出格式。
import mistune def parse_markdown(content): md = mistune.create_markdown(renderer=None) # 返回 AST ast = md(content) blocks = [] for node in ast: if node['type'] == 'heading': blocks.append({ 'content': extract_text(node), 'type': 'heading', 'level': node['attrs']['level'], 'position': node.get('position', 0) }) elif node['type'] == 'paragraph': blocks.append({ 'content': extract_text(node), 'type': 'paragraph', 'level': 0, 'position': node.get('position', 0) }) # ... 处理 list、code_block、table 等 return blocks4.2 标题层级与文档树构建
Markdown 的标题层级是构建文档树的关键。#是一级,##是二级,以此类推。但实际文档里经常出现跳级,比如从#直接跳到###,或者多个#并列。
我的处理策略是:遇到跳级时,自动补全中间层级。比如#后面直接跟###,我会在中间插入一个空的二级标题,保证树的完整性。这样后续做层级加权检索时不会出错。
def build_tree(blocks): tree = {'level': 0, 'children': [], 'content': 'root'} stack = [tree] for block in blocks: if block['type'] == 'heading': level = block['level'] while stack[-1]['level'] >= level: stack.pop() node = {'level': level, 'content': block['content'], 'children': []} stack[-1]['children'].append(node) stack.append(node) else: stack[-1]['children'].append(block) return tree这个树结构在后续检索时非常有用。比如用户问"第三章讲了什么",你可以先定位到第三章的标题节点,然后只检索它下面的子节点,大幅缩小检索范围。
4.3 表格解析:Markdown 表格的坑最多
Markdown 表格的语法是| 列1 | 列2 |,但实际文档里表格写法五花八门。有的对齐了,有的没对齐;有的分隔行是|---|---|,有的是| --- | --- |;有的单元格里有|字符但没转义。
我的做法是先用正则匹配表格块,然后按|分割,去掉首尾空单元格,再判断第二行是否是分隔行。如果是,第一行是表头,后面是数据行;如果不是,所有行都是数据行。
def parse_table(lines): rows = [] for line in lines: cells = [c.strip() for c in line.strip().strip('|').split('|')] rows.append(cells) if len(rows) >= 2 and all(re.match(r'^:?-+:?$', c) for c in rows[1]): header = rows[0] data = rows[2:] else: header = None data = rows return {'header': header, 'data': data}表格在 RAG 里的处理比较特殊。我通常会把表格转成自然语言描述再入库,比如"下表展示了各型号参数:型号 A 的功率是 100W,型号 B 的功率是 150W"。这样检索时更容易匹配到自然语言查询。
4.4 代码块和数学公式的处理
Markdown 代码块用```包裹,解析时要注意语言标记。代码块在 RAG 里通常不切分,整块作为一个单元。但如果代码块特别长(超过 2000 字符),还是要按函数或类切分。
数学公式用$...$或$$...$$包裹。这部分在 RAG 里是个难点,因为向量模型对数学公式的编码效果普遍不好。我的做法是把公式转成 LaTeX 源码保留,同时在旁边加一段自然语言解释,检索时用解释文本匹配,返回时带上公式源码。
提示:如果你的知识库里有大量数学公式,建议单独建一个公式索引,用专门的数学检索方案,不要和普通文本混在一起。
5. 完整实操流程:从原始文件到可入库块
5.1 环境准备与依赖安装
先把环境搭起来。我用的 Python 版本是 3.10,主要依赖四个库:
pip install chardet charset-normalizer mistune beautifulsoup4chardet和charset-normalizer负责编码检测,mistune负责 Markdown 解析,beautifulsoup4用来处理从 HTML 转来的 Markdown(有些 Markdown 是网页导出的,带 HTML 标签)。
5.2 统一入口函数的设计
我设计了一个统一入口函数load_document(file_path),根据文件扩展名自动选择解析器,返回归一化的块列表。
def load_document(file_path): ext = file_path.lower().split('.')[-1] encoding = detect_encoding(file_path) with open(file_path, 'r', encoding=encoding, errors='replace') as f: content = f.read() if ext == 'md' or ext == 'markdown': blocks = parse_markdown(content) elif ext == 'txt': blocks = parse_txt(content) else: raise ValueError(f"不支持的文件格式: {ext}") # 统一后处理 blocks = clean_blocks(blocks) blocks = attach_source(blocks, file_path) return blocks这个设计的好处是扩展性强。以后要支持 PDF、Word,只需要加一个分支,后处理逻辑完全复用。
5.3 清洗后处理:去掉噪音保留信号
解析出来的块还需要清洗。常见的噪音包括:空块、纯符号块、页眉页脚、重复内容。
def clean_blocks(blocks): cleaned = [] seen = set() for block in blocks: content = block['content'].strip() # 去掉空块 if not content: continue # 去掉纯符号块 if re.match(r'^[\s\W]+$', content): continue # 去掉过短的非标题块 if block['type'] != 'heading' and len(content) < 10: continue # 去重 key = content[:100] if key in seen: continue seen.add(key) block['content'] = content cleaned.append(block) return cleaned这里有个细节:去重时用前 100 个字符做 key,而不是整个内容。因为有些块内容很长,但开头相同,可能是重复的页眉。用前 100 字符做 key 能抓住大部分重复情况,又不会因为末尾差异漏掉。
5.4 切分策略:按块类型差异化处理
清洗后的块还不能直接入库,需要切分成适合向量模型的大小。不同块类型的切分策略不一样:
| 块类型 | 切分策略 | 目标大小 | 重叠 |
|---|---|---|---|
| heading | 不切分 | - | - |
| paragraph | 按句子切分 | 300-500 字 | 50 字 |
| list | 按列表项切分 | 200-400 字 | 无 |
| code | 按函数/类切分 | 500-1000 字 | 无 |
| table | 转自然语言后切分 | 300-500 字 | 无 |
段落切分我推荐按句子边界切,而不是按固定字数切。因为句子是语义的最小完整单元,从句子中间切开会导致语义不完整。中文句子边界用。!?判断,英文用.?!判断。
def split_paragraph(text, max_len=500, overlap=50): sentences = re.split(r'(?<=[。!?.!?])\s*', text) chunks = [] current = "" for sent in sentences: if len(current) + len(sent) <= max_len: current += sent else: if current: chunks.append(current) current = sent if current: chunks.append(current) # 加重叠 if overlap > 0 and len(chunks) > 1: for i in range(1, len(chunks)): chunks[i] = chunks[i-1][-overlap:] + chunks[i] return chunks重叠的作用是防止语义在切分点丢失。比如一个概念的解释跨了两个块,有重叠的话,两个块都能检索到部分信息。
5.5 元数据附加:让每个块都可追溯
最后一步是给每个块附加元数据。除了前面说的source和position,我还建议加上heading_path,记录这个块所属的标题路径。
def attach_heading_path(blocks): path = [] for block in blocks: if block['type'] == 'heading': level = block['level'] path = path[:level-1] path.append(block['content']) block['heading_path'] = ' > '.join(path) return blocksheading_path在检索时非常有用。比如检索结果里显示"第三章 > 3.2 节 > 参数配置",用户一眼就知道这个块在文档的什么位置,信任度会高很多。
6. 常见问题与排查技巧实录
6.1 编码问题速查表
| 现象 | 可能原因 | 解决方法 |
|---|---|---|
| 中文显示为乱码 | 编码检测错误 | 手动指定 GBK 或 GB18030 |
| 文件开头有奇怪字符 | UTF-8 BOM | 用utf-8-sig编码读取 |
| 部分字符显示为问号 | 非法字节 | 加errors='replace' |
| 读取时报 UnicodeDecodeError | 混合编码 | 分段检测,逐段解码 |
6.2 解析结果不对怎么排查
第一步,打印原始内容的前 500 字符,确认读取是否正确。第二步,打印解析后的块列表,看结构是否符合预期。第三步,如果结构不对,单独测试解析函数,用最小复现样本调试。
我常用的调试代码是这样的:
def debug_parse(file_path): blocks = load_document(file_path) print(f"总块数: {len(blocks)}") for i, block in enumerate(blocks[:20]): print(f"[{i}] type={block['type']} level={block['level']}") print(f" content={block['content'][:80]}") print(f" path={block.get('heading_path', '')}")这个输出能快速定位问题。如果块数明显偏少,可能是清洗过度;如果块数偏多,可能是切分过细;如果类型全是 paragraph,可能是标题识别失败。
6.3 我踩过的三个坑
第一个坑是过度清洗。早期我写清洗规则时,把长度小于 20 的块全删了,结果把很多短标题也删了。后来改成只删非标题的短块,标题再短也保留。
第二个坑是切分重叠过大。我一开始设了 200 字重叠,结果向量库里全是重复内容,检索时返回一堆相似结果。后来把重叠降到 50 字,效果好多了。
第三个坑是忽略表格。我一开始把表格当普通段落处理,结果表格内容被切得七零八落。后来改成表格转自然语言,检索准确率明显提升。
实操心得:数据导入阶段一定要做抽样验证。我通常随机抽 20 个块,人工检查内容是否完整、结构是否正确、元数据是否齐全。这个习惯帮我提前发现了无数问题。
6.4 性能优化:大文件怎么处理
如果 txt 文件超过 10MB,一次性读入内存可能有问题。我的做法是分块读取,每次读 1MB,解析后立即处理,不保留原始内容。
def stream_read(file_path, chunk_size=1024*1024): with open(file_path, 'r', encoding='utf-8', errors='replace') as f: while True: chunk = f.read(chunk_size) if not chunk: break yield chunk但分块读取有个问题:断行可能跨块。所以要在块边界处保留最后一行,和下一块的第一行拼接。这个逻辑稍微复杂一点,但对大文件是必须的。
7. 结构化输出的下游衔接
解析和切分完成后,块列表要进入向量化环节。这里简单提一下衔接要点,因为这是下一篇的主题。
每个块在向量化前,我建议做一次检索文本重组:把heading_path和块内容拼在一起,作为向量化的输入。比如"第三章 > 3.2 节 > 参数配置:型号 A 的功率是 100W..."。这样向量里包含了层级信息,检索时能更好地匹配带上下文的查询。
另外,type和level这两个字段要作为元数据存进向量库,检索时可以按类型过滤。比如用户问代码相关的问题,可以只检索type=code的块。
最后再分享一个小技巧:如果你的知识库同时有 txt 和 Markdown,建议在元数据里加一个format字段,记录原始格式。这样后续做效果分析时,可以对比两种格式的检索质量,判断哪种格式的数据更有价值。
这个系列后续还会讲 PDF、Word、HTML 的解析,以及切分策略的进阶玩法。txt 和 Markdown 是基础,把这块打扎实了,后面的格式都是在这个框架上做扩展。