尧图网站建设 
1. RAG 数据导入与解析的整体设计思路1.1 为什么数据导入是 RAG 系统的第一道生死关做过 RAG 项目的人都有一个共识检索效果差八成问题出在数据导入和解析环节而不是模型本身。我见过太多团队花大价钱调 embedding 模型、换 rerank 策略结果回头一看原始文档被解析得七零八落——表格变成了一堆乱码、标题层级全丢了、代码块和正文混在一起。这种情况下后面再怎么优化都是白搭。RAG 的本质是先检索、后生成检索的质量直接决定了生成的上限。而检索质量的第一道关卡就是数据导入与解析。你可以把整个 RAG 流程想象成一条流水线原始文档是原材料解析是粗加工切分是精加工向量化是打包检索是出库。如果粗加工阶段就把原材料搞坏了后面每一步都在放大错误。这一篇我聚焦在通用文本与结构化文档的导入解析上具体覆盖从最朴素的 txt 纯文本到带层级结构的 Markdown 文档。为什么先讲这两类因为它们是一切文档格式的最大公约数——PDF 解析出来最终要转成文本或 MarkdownWord 解析出来也是HTML 抓取下来还是。把 txt 和 Markdown 这两条路走通了后面处理 PDF、Word、Excel 就是套壳的事。1.2 通用文本与结构化文档的边界划分在动手写代码之前得先把通用文本和结构化文档这两个概念掰扯清楚因为它们对应的解析策略完全不同。通用文本txt 类的特点是没有显式的结构标记段落靠换行区分标题靠人眼识别。它可能是小说文本、日志文件、导出的聊天记录、词典数据。这类文档的解析难点在于——你得靠启发式规则去猜结构。比如一行文字前后都有空行、长度较短、不含标点结尾那它大概率是个标题。结构化文档Markdown 类的特点是结构是显式声明的。#是标题-是列表是代码块|是表格。解析这类文档核心工作是把标记语言转换成带元数据的中间表示同时保留层级关系。我一般会把导入解析拆成三个阶段阶段核心任务输出物读取层把文件读进内存处理编码原始字符串解析层识别结构提取元数据结构化节点树归一化层统一成下游可消费的格式标准文档对象这个三层划分的好处是职责清晰。读取层只管编码和 IO解析层只管结构识别归一化层只管格式统一。后面要加 PDF 支持只需要在读取层加一个 PDF reader解析层和归一化层几乎不用动。1.3 技术选型为什么不用现成框架一把梭市面上有 LangChain、LlamaIndex 这类框架自带各种 DocumentLoader很多人第一反应是直接调TextLoader、UnstructuredMarkdownLoader完事。我早期也这么干过但踩了几个坑之后现在更倾向于自己写解析层只在切分和向量化环节用框架。原因有三点。第一框架的 Loader 是黑盒它把元数据塞在metadata字典里字段命名和结构各版本还不一样你想加个自定义字段比如章节路径、文档来源分类很别扭。第二通用 Loader 对中文支持一般尤其是标题识别、标点处理这些细节英文场景调好的参数直接拿来用中文会翻车。第三调试困难解析出问题时你很难定位是 Loader 的锅还是切分器的锅。自己写解析层代码量其实不大一个 txt 解析器加一个 Markdown 解析器核心逻辑加起来两三百行。但换来的是完全可控元数据想加什么加什么中文规则想怎么调怎么调出问题一眼就能定位。这笔账怎么算都划算。提示不是说框架不能用而是建议把解析这一步掌握在自己手里。切分、向量化、检索这些环节用框架提效没问题但解析层是数据质量的源头值得自己写。2. 核心细节解析与实操要点2.1 编码问题中文文本导入的第一只拦路虎处理中文 txt 文件编码问题是绕不过去的坎。我统计过自己经手的项目大概有三成的解析乱码问题根源都是编码没处理好。中文文本常见的编码有这几种UTF-8、GBK、GB2312、GB18030、Big5。其中 GB18030 是 GBK 的超集GBK 又是 GB2312 的超集。很多老系统导出的 txt 是 GBK 编码你直接用 UTF-8 读就会得到一堆锟斤拷。我的处理策略是先探测、再兜底。探测用chardet库它能给出编码的置信度兜底就是按优先级列表逐个尝试解码哪个成功用哪个。import chardet def detect_encoding(file_path, sample_size100000): with open(file_path, rb) as f: raw f.read(sample_size) result chardet.detect(raw) return result[encoding], result[confidence] def read_text_safe(file_path): # 优先用探测结果 encoding, confidence detect_encoding(file_path) candidates [] if encoding and confidence 0.7: candidates.append(encoding) # 兜底列表中文场景按这个顺序试 candidates.extend([utf-8, gb18030, gbk, big5, latin-1]) for enc in candidates: 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这里有个细节值得说为什么把latin-1放在兜底列表里因为 latin-1 能解码任意字节序列永远不会抛异常。它放在最后是作为绝对不崩的保险但实际项目中如果走到这一步说明前面全失败了得人工介入检查。还有一个坑BOM 头。Windows 记事本保存的 UTF-8 文件常带 BOM字节序标记读出来开头会多个\ufeff。处理方法是读取后用content.lstrip(\ufeff)清掉或者直接用utf-8-sig编码读取。注意chardet对短文本的探测准确率不高如果文件小于 1KB建议直接按 UTF-8 和 GB18030 两个候选试别太信探测结果。2.2 通用文本的结构识别用启发式规则猜出层级纯 txt 没有结构标记但我们能从排版规律里猜出结构。核心思路是基于行的特征做分类把每一行判定为标题、正文、列表项还是空行。我总结了一套在中文场景下比较稳的启发式规则按优先级从高到低空行连续两个以上换行作为段落分隔符标题候选行长度小于 40 字、不以句号/逗号/分号结尾、前后有空行、可能带编号前缀如第一章、1.1、一、列表项以-、*、•、·或数字加顿号/点号开头正文其余情况标题识别是最关键的也是最容易出错的。我一般会定义一个正则集合来匹配常见的中文标题模式import re TITLE_PATTERNS [ r^第[一二三四五六七八九十百千][章节回篇部分], # 第X章 r^[一二三四五六七八九十][、.], # 一、 r^\d[、.]\s*\S, # 1. 或 1、 r^\d\.\d[、.]?\s*\S, # 1.1 r^[(]\d[)], # (1) r^[A-Z][、.]\s*\S, # A. ] def is_title_candidate(line, prev_blank, next_blank): line line.strip() if not line or len(line) 40: return False # 带编号前缀的直接判定为标题 for pat in TITLE_PATTERNS: if re.match(pat, line): return True # 无编号的要求前后有空行且不以标点结尾 if prev_blank and next_blank: if not re.search(r[。、,;:]$, line): return True return False这套规则不是万能的但对小说、技术文档、报告这类常见文本准确率能到八成以上。剩下的两成怎么办留人工校验的接口。我会在解析结果里给每个标题打一个confidence分数低分的标出来让用户能快速过一遍。2.3 Markdown 解析保留层级是核心诉求Markdown 的解析比 txt 简单因为结构是显式的。但简单不等于随便核心诉求是保留层级关系。为什么层级这么重要因为 RAG 切分时我们希望每个 chunk 都带上它所属的章节路径。比如一个 chunk 来自第三章 3.2 节 参数说明检索时用户问参数怎么配这个 chunk 的章节路径就是极强的上下文信号能显著提升召回准确率。解析 Markdown 我推荐用markdown-it-py或者mistune它们能把 Markdown 解析成 token 流每个 token 带类型和层级信息。不推荐用正则硬撸因为 Markdown 的边界情况太多嵌套列表、代码块里的#、转义的\#正则很容易翻车。from markdown_it import MarkdownIt def parse_markdown(content): md MarkdownIt(commonmark, {html: False}) tokens md.parse(content) nodes [] heading_stack [] # 维护标题层级栈 for token in tokens: if token.type heading_open: level int(token.tag[1]) # h1 - 1 # 弹出比当前层级深的 while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, )) elif token.type inline and heading_stack and heading_stack[-1][1] : # 这是标题的文本内容 heading_stack[-1] (heading_stack[-1][0], token.content) elif token.type paragraph_open: # 记录当前段落所属的章节路径 path .join(h[1] for h in heading_stack) nodes.append({type: paragraph, path: path}) return nodes这段代码的关键是heading_stack。它维护了当前所处的标题路径遇到新标题时按层级弹出旧的、压入新的。这样每个段落节点都能拿到完整的章节路径。提示markdown-it-py的html: False选项很重要它会禁用 HTML 标签解析避免 XSS 风险也避免 HTML 标签污染文本内容。2.4 元数据设计给每个 chunk 装上身份证解析出来的内容最终要变成带元数据的文档对象。元数据设计得好不好直接决定了后面检索和过滤的灵活性。我一般会设计这么几类字段字段类别具体字段用途来源信息source_path, file_name, file_type溯源、按来源过滤结构信息section_path, heading_level, node_type上下文增强、按章节过滤内容信息char_count, token_count, language切分策略、质量过滤时间信息created_at, modified_at时效性排序自定义tags, category业务标签过滤其中section_path是我认为最有价值的字段。它记录了 chunk 在文档中的位置路径检索时可以作为强上下文拼接到 chunk 前面也可以作为过滤条件。比如用户问第三章讲了什么直接按section_path前缀过滤就能精准命中。元数据的存储格式建议用扁平字典别嵌套太深。因为大多数向量数据库对元数据的支持是键值对级别嵌套结构存进去容易出问题。如果确实需要嵌套序列化成 JSON 字符串存。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.9 以上因为要用到一些新语法特性。核心依赖就三个pip install chardet markdown-it-py tiktokenchardet编码探测markdown-it-pyMarkdown 解析tiktokentoken 计数用于后续切分时控制 chunk 大小如果你还想处理更复杂的文档可以加beautifulsoup4HTML、python-docxWord、openpyxlExcel但这一篇聚焦 txt 和 Markdown这三个就够了。3.2 通用文本解析器完整实现把前面讲的编码处理、结构识别串起来形成一个完整的 txt 解析器。我把它设计成一个类方便复用和扩展。import re import chardet from dataclasses import dataclass, field from typing import List, Optional dataclass class DocNode: node_type: str # title / paragraph / list_item content: str level: int 0 # 标题层级正文为 0 section_path: str # 章节路径 metadata: dict field(default_factorydict) class TxtParser: def __init__(self, max_title_len40): self.max_title_len max_title_len self.title_patterns [ r^第[一二三四五六七八九十百千][章节回篇部分], r^[一二三四五六七八九十][、.], r^\d[、.]\s*\S, r^\d\.\d[、.]?\s*\S, r^[(]\d[)], ] def _detect_encoding(self, raw: bytes) - str: result chardet.detect(raw[:100000]) enc result.get(encoding) if enc and result.get(confidence, 0) 0.7: return enc return utf-8 def _read(self, file_path: str) - str: with open(file_path, rb) as f: raw f.read() enc self._detect_encoding(raw) for candidate in [enc, utf-8, gb18030, gbk, big5]: try: text raw.decode(candidate) return text.lstrip(\ufeff) except (UnicodeDecodeError, LookupError): continue return raw.decode(utf-8, errorsreplace) def _is_title(self, line: str, prev_blank: bool, next_blank: bool) - bool: line line.strip() if not line or len(line) self.max_title_len: return False for pat in self.title_patterns: if re.match(pat, line): return True if prev_blank and next_blank: if not re.search(r[。、,;:!?]$, line): return True return False def parse(self, file_path: str) - List[DocNode]: text self._read(file_path) lines text.split(\n) nodes [] heading_stack [] # [(level, title)] buffer [] def flush_buffer(): if buffer: content \n.join(buffer).strip() if content: path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typeparagraph, contentcontent, section_pathpath, )) buffer.clear() for i, line in enumerate(lines): prev_blank (i 0) or (not lines[i-1].strip()) next_blank (i len(lines)-1) or (not lines[i1].strip()) if not line.strip(): flush_buffer() continue if self._is_title(line, prev_blank, next_blank): flush_buffer() # 推断层级带编号的按编号深度否则默认 1 level self._infer_level(line) while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, line.strip())) nodes.append(DocNode( node_typetitle, contentline.strip(), levellevel, section_path .join(h[1] for h in heading_stack), )) else: buffer.append(line) flush_buffer() return nodes def _infer_level(self, line: str) - int: line line.strip() if re.match(r^第[一二三四五六七八九十百千][章篇], line): return 1 if re.match(r^第[一二三四五六七八九十百千][节回], line): return 2 if re.match(r^\d\.\d, line): return 2 if re.match(r^\d[、.], line): return 1 return 1这个解析器的核心逻辑是逐行扫描 缓冲区。遇到标题就 flush 缓冲区、更新标题栈遇到正文就攒进缓冲区遇到空行也 flush。这样能保证每个段落节点都带上正确的section_path。3.3 Markdown 解析器完整实现Markdown 解析器基于markdown-it-py重点是把 token 流转换成统一的DocNode列表。from markdown_it import MarkdownIt class MarkdownParser: def __init__(self): self.md MarkdownIt(commonmark, {html: False}) def parse(self, file_path: str) - List[DocNode]: with open(file_path, r, encodingutf-8) as f: content f.read() return self.parse_string(content) def parse_string(self, content: str) - List[DocNode]: tokens self.md.parse(content) nodes [] heading_stack [] # [(level, title)] i 0 while i len(tokens): token tokens[i] if token.type heading_open: level int(token.tag[1]) # 下一个 token 是 inline包含标题文本 title_text tokens[i1].content if i1 len(tokens) else while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, title_text)) nodes.append(DocNode( node_typetitle, contenttitle_text, levellevel, section_path .join(h[1] for h in heading_stack), )) i 3 # 跳过 heading_open, inline, heading_close continue if token.type paragraph_open: # 收集到 paragraph_close 之间的内容 para_content tokens[i1].content if i1 len(tokens) else path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typeparagraph, contentpara_content, section_pathpath, )) i 3 continue if token.type fence: # 代码块 path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typecode, contenttoken.content, section_pathpath, metadata{lang: token.info.strip()}, )) i 1 continue if token.type table_open: # 表格整体作为一个节点 table_lines [] j i while j len(tokens) and tokens[j].type ! table_close: if tokens[j].type inline: table_lines.append(tokens[j].content) j 1 path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typetable, content\n.join(table_lines), section_pathpath, )) i j 1 continue i 1 return nodes这里有个细节代码块和表格要单独处理。因为它们的切分策略和普通段落不一样——代码块最好整体保留不要从中间切开表格也是切开就失去意义了。所以我在解析阶段就把它们标记成独立的node_type后面切分时按类型走不同策略。3.4 归一化输出统一成下游可消费的格式两个解析器输出的都是DocNode列表但下游切分器、向量化器需要的是统一的文档对象。我一般会再包一层归一化把DocNode列表转成标准的Document对象。dataclass class Document: doc_id: str source: str nodes: List[DocNode] metadata: dict field(default_factorydict) def to_chunks(self, chunker): 交给切分器处理 return chunker.split(self) def normalize(nodes: List[DocNode], source: str) - Document: import hashlib doc_id hashlib.md5(source.encode()).hexdigest()[:16] total_chars sum(len(n.content) for n in nodes) return Document( doc_iddoc_id, sourcesource, nodesnodes, metadata{ total_nodes: len(nodes), total_chars: total_chars, has_structure: any(n.node_type title for n in nodes), } )doc_id用来源路径的 MD5 生成保证同一文件多次导入得到相同 ID方便去重和更新。has_structure字段标记文档是否有识别出标题结构后面切分时可以据此选择不同的切分策略——有结构的按章节切没结构的按固定长度切。3.5 完整调用示例与效果验证把上面几个类串起来跑一个完整流程def import_document(file_path: str) - Document: if file_path.endswith(.md) or file_path.endswith(.markdown): parser MarkdownParser() else: parser TxtParser() nodes parser.parse(file_path) return normalize(nodes, file_path) # 使用 doc import_document(sample.md) print(f文档 ID: {doc.doc_id}) print(f节点数: {len(doc.nodes)}) for node in doc.nodes[:5]: print(f[{node.node_type}] path{node.section_path}) print(f content: {node.content[:50]}...)跑完之后重点看几个指标节点总数是否合理太少说明结构没识别出来太多说明切得太碎、section_path 是否正确层级有没有错乱、内容有没有乱码。这三个指标正常解析这关就算过了。4. 常见问题与排查技巧实录4.1 编码乱码问题速查编码问题是最高频的我整理了一个速查表现象可能原因解决方法中文显示为锟斤拷UTF-8 读 GBK 文件改用 gb18030 读取中文显示为????编码不支持中文检查是否用了 latin-1开头多出\ufeffUTF-8 BOM用 utf-8-sig 或 lstrip部分字符乱码混合编码分段探测或人工确认繁体显示异常Big5 编码改用 big5 或 big5hkscs排查思路是先看乱码模式。锟斤拷是典型的 UTF-8/GBK 错配????是编码表里没有对应字符\ufeff是 BOM。根据模式反推原因比盲目试编码快得多。4.2 标题识别误判的三种典型场景启发式规则再稳也有翻车的时候。我遇到最多的三种误判第一种短句被误判为标题。比如正文里有一句今天天气不错前后恰好有空行长度也短就被当成标题了。解决办法是加一个标题不应以常见语气词结尾的规则或者要求标题候选行不含句末标点。第二种长标题被漏判。有些技术文档的标题很长超过 40 字就被我的规则排除了。解决办法是对带编号前缀的行放宽长度限制因为带编号的基本可以确定是标题。第三种代码块里的#被误判。txt 里如果有代码片段# 这是注释会被当成 Markdown 标题。这个在纯 txt 场景下很难完美解决我的做法是检测连续多行以#或//开头时判定为代码块整体跳过标题识别。提示标题识别没有 100% 准确的方案建议在解析结果里保留confidence字段低置信度的标题让用户人工确认。宁可漏判不可误判——误判的标题会污染 section_path比没有 section_path 更糟。4.3 Markdown 解析的边界情况处理Markdown 解析虽然用库但边界情况还是得自己处理嵌套列表markdown-it-py会把嵌套列表解析成多层bullet_list_open需要递归处理。我一般把整个列表作为一个节点保留原始文本不拆开。引用块blockquote里的内容我倾向于保留标记因为引用块往往有特殊语义如注意事项。HTML 内联虽然禁用了 HTML 解析但有些 Markdown 里混了br这类标签会被当纯文本。可以在解析后做一次正则清理。数学公式$...$和$$...$$在 commonmark 里不被识别需要装mdit-py-plugins的 dollarmath 插件。如果文档里有公式记得加上。4.4 大文件处理的内存优化处理几十 MB 的大 txt 文件时一次性读进内存可能爆。我的优化策略是流式读取 分批解析def parse_large_txt(file_path, batch_size10000): nodes [] buffer [] heading_stack [] with open(file_path, r, encodingutf-8) as f: for line in f: # 逐行读不一次性加载 # ... 同样的解析逻辑 if len(buffer) batch_size: # 处理一批 pass return nodes逐行读的好处是内存占用恒定跟文件大小无关。代价是没法做前后空行判断需要维护一个滑动窗口。我一般维护最近 3 行的状态足够判断空行了。4.5 解析质量的自检清单每次解析完我会跑一遍自检清单确认质量[ ] 编码是否正确有没有乱码字符[ ] 标题数量是否合理跟文档目录对比[ ] section_path 层级有没有错乱比如 h3 出现在 h1 前面[ ] 空节点有没有content 为空的节点要过滤[ ] 代码块和表格有没有被正确识别[ ] 总字符数跟原文件是否接近差太多说明丢内容了这个清单跑一遍基本能拦住九成的解析问题。剩下的疑难杂症就得靠人工抽查了。5. 从解析到切分的衔接要点5.1 结构化节点如何指导切分策略解析出来的DocNode列表直接决定了切分策略。我的原则是按节点类型走不同切分逻辑标题节点不单独成 chunk而是作为后续内容的section_path前缀段落节点按 token 数切分超过阈值就拆拆的时候保留 section_path代码块节点整体保留不拆。如果代码块本身超长按函数/类边界拆表格节点整体保留不拆。表格拆开就失去语义了这样切出来的 chunk每个都带完整的章节路径检索时上下文信息充足。5.2 chunk 元数据的继承与增强切分时chunk 的元数据从父节点继承同时做一些增强def make_chunk(node, parent_doc): return { content: node.content, doc_id: parent_doc.doc_id, source: parent_doc.source, section_path: node.section_path, node_type: node.node_type, # 增强字段 char_count: len(node.content), has_code: node.node_type code, depth: node.section_path.count( ) 1, }depth字段表示 chunk 在文档中的深度检索时可以据此调整权重——浅层节点章节标题附近往往概括性强深层节点细节多可以根据查询类型动态调整。5.3 为后续 PDF/Word 解析预留扩展点这一篇虽然只讲 txt 和 Markdown但架构上要为后续格式预留扩展点。我的做法是定义统一的 Parser 接口from abc import ABC, abstractmethod class BaseParser(ABC): abstractmethod def parse(self, file_path: str) - List[DocNode]: pass class PdfParser(BaseParser): def parse(self, file_path: str) - List[DocNode]: # 后续实现 pass所有解析器都实现parse方法返回统一的DocNode列表。这样归一化层和切分层完全不用改加新格式只需要写一个新的 Parser 类。这个扩展点在后续处理 PDF、Word、Excel 时会非常省事。6. 实操中的几点个人体会6.1 解析质量比解析速度重要得多我早期做 RAG 时追求解析速度用了各种优化结果检索效果一直上不去。后来把解析速度降下来加了编码探测、标题识别、结构保留这些慢操作检索准确率直接涨了一截。解析是离线的一次性工作慢一点没关系质量才是关键。一个文档解析花 10 秒还是 1 秒对用户体验没影响但解析质量差 10%检索效果可能差一半。6.2 保留原始文本永远是对的解析过程中我坚持保留原始文本。即使做了归一化、做了结构提取原始字符串也存一份。因为后面调试时你永远不知道会需要回看原文的哪个细节。我踩过的坑是早期为了省空间解析后只存结构化结果结果发现某个 chunk 内容异常想对比原文却找不到了只能重新解析。从那以后原始文本必存。6.3 中文场景的规则要自己调网上开源的解析规则大多是英文场景调好的直接拿来用中文会翻车。比如英文标题识别靠首字母大写中文没这个概念英文句子靠句号分句中文还有分号、顿号、省略号。中文场景的解析规则必须基于中文语料自己调。我一般会准备一个 100 篇左右的中文测试集每次改规则都跑一遍看准确率和召回率的变化。6.4 给解析结果加一个人工校验入口再好的自动解析也有出错的时候。我会在解析流程最后加一个人工校验入口把解析结果渲染成可视化界面让用户能快速浏览、标记错误、手动修正。这个入口看起来增加了工作量但实际用下来它极大提升了数据质量也降低了后续排查问题的成本。用户花 5 分钟校验可能省下后面 5 小时的调试。最后分享一个小技巧解析日志要详细。每次解析记录用了什么编码、识别了多少标题、有多少节点、耗时多少。这些日志在排查问题时是金矿。我现在的解析器日志详细到每一行被判定为什么类型出问题时直接看日志就能定位。这个习惯是从无数次解析结果不对但不知道哪一步错了的痛苦中养成的强烈建议你也加上。 RAG数据导入与解析实战：从txt到Markdown的完整指南
1. RAG 数据导入与解析的整体设计思路1.1 为什么数据导入是 RAG 系统的第一道生死关做过 RAG 项目的人都有一个共识检索效果差八成问题出在数据导入和解析环节而不是模型本身。我见过太多团队花大价钱调 embedding 模型、换 rerank 策略结果回头一看原始文档被解析得七零八落——表格变成了一堆乱码、标题层级全丢了、代码块和正文混在一起。这种情况下后面再怎么优化都是白搭。RAG 的本质是先检索、后生成检索的质量直接决定了生成的上限。而检索质量的第一道关卡就是数据导入与解析。你可以把整个 RAG 流程想象成一条流水线原始文档是原材料解析是粗加工切分是精加工向量化是打包检索是出库。如果粗加工阶段就把原材料搞坏了后面每一步都在放大错误。这一篇我聚焦在通用文本与结构化文档的导入解析上具体覆盖从最朴素的 txt 纯文本到带层级结构的 Markdown 文档。为什么先讲这两类因为它们是一切文档格式的最大公约数——PDF 解析出来最终要转成文本或 MarkdownWord 解析出来也是HTML 抓取下来还是。把 txt 和 Markdown 这两条路走通了后面处理 PDF、Word、Excel 就是套壳的事。1.2 通用文本与结构化文档的边界划分在动手写代码之前得先把通用文本和结构化文档这两个概念掰扯清楚因为它们对应的解析策略完全不同。通用文本txt 类的特点是没有显式的结构标记段落靠换行区分标题靠人眼识别。它可能是小说文本、日志文件、导出的聊天记录、词典数据。这类文档的解析难点在于——你得靠启发式规则去猜结构。比如一行文字前后都有空行、长度较短、不含标点结尾那它大概率是个标题。结构化文档Markdown 类的特点是结构是显式声明的。#是标题-是列表是代码块|是表格。解析这类文档核心工作是把标记语言转换成带元数据的中间表示同时保留层级关系。我一般会把导入解析拆成三个阶段阶段核心任务输出物读取层把文件读进内存处理编码原始字符串解析层识别结构提取元数据结构化节点树归一化层统一成下游可消费的格式标准文档对象这个三层划分的好处是职责清晰。读取层只管编码和 IO解析层只管结构识别归一化层只管格式统一。后面要加 PDF 支持只需要在读取层加一个 PDF reader解析层和归一化层几乎不用动。1.3 技术选型为什么不用现成框架一把梭市面上有 LangChain、LlamaIndex 这类框架自带各种 DocumentLoader很多人第一反应是直接调TextLoader、UnstructuredMarkdownLoader完事。我早期也这么干过但踩了几个坑之后现在更倾向于自己写解析层只在切分和向量化环节用框架。原因有三点。第一框架的 Loader 是黑盒它把元数据塞在metadata字典里字段命名和结构各版本还不一样你想加个自定义字段比如章节路径、文档来源分类很别扭。第二通用 Loader 对中文支持一般尤其是标题识别、标点处理这些细节英文场景调好的参数直接拿来用中文会翻车。第三调试困难解析出问题时你很难定位是 Loader 的锅还是切分器的锅。自己写解析层代码量其实不大一个 txt 解析器加一个 Markdown 解析器核心逻辑加起来两三百行。但换来的是完全可控元数据想加什么加什么中文规则想怎么调怎么调出问题一眼就能定位。这笔账怎么算都划算。提示不是说框架不能用而是建议把解析这一步掌握在自己手里。切分、向量化、检索这些环节用框架提效没问题但解析层是数据质量的源头值得自己写。2. 核心细节解析与实操要点2.1 编码问题中文文本导入的第一只拦路虎处理中文 txt 文件编码问题是绕不过去的坎。我统计过自己经手的项目大概有三成的解析乱码问题根源都是编码没处理好。中文文本常见的编码有这几种UTF-8、GBK、GB2312、GB18030、Big5。其中 GB18030 是 GBK 的超集GBK 又是 GB2312 的超集。很多老系统导出的 txt 是 GBK 编码你直接用 UTF-8 读就会得到一堆锟斤拷。我的处理策略是先探测、再兜底。探测用chardet库它能给出编码的置信度兜底就是按优先级列表逐个尝试解码哪个成功用哪个。import chardet def detect_encoding(file_path, sample_size100000): with open(file_path, rb) as f: raw f.read(sample_size) result chardet.detect(raw) return result[encoding], result[confidence] def read_text_safe(file_path): # 优先用探测结果 encoding, confidence detect_encoding(file_path) candidates [] if encoding and confidence 0.7: candidates.append(encoding) # 兜底列表中文场景按这个顺序试 candidates.extend([utf-8, gb18030, gbk, big5, latin-1]) for enc in candidates: 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这里有个细节值得说为什么把latin-1放在兜底列表里因为 latin-1 能解码任意字节序列永远不会抛异常。它放在最后是作为绝对不崩的保险但实际项目中如果走到这一步说明前面全失败了得人工介入检查。还有一个坑BOM 头。Windows 记事本保存的 UTF-8 文件常带 BOM字节序标记读出来开头会多个\ufeff。处理方法是读取后用content.lstrip(\ufeff)清掉或者直接用utf-8-sig编码读取。注意chardet对短文本的探测准确率不高如果文件小于 1KB建议直接按 UTF-8 和 GB18030 两个候选试别太信探测结果。2.2 通用文本的结构识别用启发式规则猜出层级纯 txt 没有结构标记但我们能从排版规律里猜出结构。核心思路是基于行的特征做分类把每一行判定为标题、正文、列表项还是空行。我总结了一套在中文场景下比较稳的启发式规则按优先级从高到低空行连续两个以上换行作为段落分隔符标题候选行长度小于 40 字、不以句号/逗号/分号结尾、前后有空行、可能带编号前缀如第一章、1.1、一、列表项以-、*、•、·或数字加顿号/点号开头正文其余情况标题识别是最关键的也是最容易出错的。我一般会定义一个正则集合来匹配常见的中文标题模式import re TITLE_PATTERNS [ r^第[一二三四五六七八九十百千][章节回篇部分], # 第X章 r^[一二三四五六七八九十][、.], # 一、 r^\d[、.]\s*\S, # 1. 或 1、 r^\d\.\d[、.]?\s*\S, # 1.1 r^[(]\d[)], # (1) r^[A-Z][、.]\s*\S, # A. ] def is_title_candidate(line, prev_blank, next_blank): line line.strip() if not line or len(line) 40: return False # 带编号前缀的直接判定为标题 for pat in TITLE_PATTERNS: if re.match(pat, line): return True # 无编号的要求前后有空行且不以标点结尾 if prev_blank and next_blank: if not re.search(r[。、,;:]$, line): return True return False这套规则不是万能的但对小说、技术文档、报告这类常见文本准确率能到八成以上。剩下的两成怎么办留人工校验的接口。我会在解析结果里给每个标题打一个confidence分数低分的标出来让用户能快速过一遍。2.3 Markdown 解析保留层级是核心诉求Markdown 的解析比 txt 简单因为结构是显式的。但简单不等于随便核心诉求是保留层级关系。为什么层级这么重要因为 RAG 切分时我们希望每个 chunk 都带上它所属的章节路径。比如一个 chunk 来自第三章 3.2 节 参数说明检索时用户问参数怎么配这个 chunk 的章节路径就是极强的上下文信号能显著提升召回准确率。解析 Markdown 我推荐用markdown-it-py或者mistune它们能把 Markdown 解析成 token 流每个 token 带类型和层级信息。不推荐用正则硬撸因为 Markdown 的边界情况太多嵌套列表、代码块里的#、转义的\#正则很容易翻车。from markdown_it import MarkdownIt def parse_markdown(content): md MarkdownIt(commonmark, {html: False}) tokens md.parse(content) nodes [] heading_stack [] # 维护标题层级栈 for token in tokens: if token.type heading_open: level int(token.tag[1]) # h1 - 1 # 弹出比当前层级深的 while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, )) elif token.type inline and heading_stack and heading_stack[-1][1] : # 这是标题的文本内容 heading_stack[-1] (heading_stack[-1][0], token.content) elif token.type paragraph_open: # 记录当前段落所属的章节路径 path .join(h[1] for h in heading_stack) nodes.append({type: paragraph, path: path}) return nodes这段代码的关键是heading_stack。它维护了当前所处的标题路径遇到新标题时按层级弹出旧的、压入新的。这样每个段落节点都能拿到完整的章节路径。提示markdown-it-py的html: False选项很重要它会禁用 HTML 标签解析避免 XSS 风险也避免 HTML 标签污染文本内容。2.4 元数据设计给每个 chunk 装上身份证解析出来的内容最终要变成带元数据的文档对象。元数据设计得好不好直接决定了后面检索和过滤的灵活性。我一般会设计这么几类字段字段类别具体字段用途来源信息source_path, file_name, file_type溯源、按来源过滤结构信息section_path, heading_level, node_type上下文增强、按章节过滤内容信息char_count, token_count, language切分策略、质量过滤时间信息created_at, modified_at时效性排序自定义tags, category业务标签过滤其中section_path是我认为最有价值的字段。它记录了 chunk 在文档中的位置路径检索时可以作为强上下文拼接到 chunk 前面也可以作为过滤条件。比如用户问第三章讲了什么直接按section_path前缀过滤就能精准命中。元数据的存储格式建议用扁平字典别嵌套太深。因为大多数向量数据库对元数据的支持是键值对级别嵌套结构存进去容易出问题。如果确实需要嵌套序列化成 JSON 字符串存。3. 实操过程与核心环节实现3.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.9 以上因为要用到一些新语法特性。核心依赖就三个pip install chardet markdown-it-py tiktokenchardet编码探测markdown-it-pyMarkdown 解析tiktokentoken 计数用于后续切分时控制 chunk 大小如果你还想处理更复杂的文档可以加beautifulsoup4HTML、python-docxWord、openpyxlExcel但这一篇聚焦 txt 和 Markdown这三个就够了。3.2 通用文本解析器完整实现把前面讲的编码处理、结构识别串起来形成一个完整的 txt 解析器。我把它设计成一个类方便复用和扩展。import re import chardet from dataclasses import dataclass, field from typing import List, Optional dataclass class DocNode: node_type: str # title / paragraph / list_item content: str level: int 0 # 标题层级正文为 0 section_path: str # 章节路径 metadata: dict field(default_factorydict) class TxtParser: def __init__(self, max_title_len40): self.max_title_len max_title_len self.title_patterns [ r^第[一二三四五六七八九十百千][章节回篇部分], r^[一二三四五六七八九十][、.], r^\d[、.]\s*\S, r^\d\.\d[、.]?\s*\S, r^[(]\d[)], ] def _detect_encoding(self, raw: bytes) - str: result chardet.detect(raw[:100000]) enc result.get(encoding) if enc and result.get(confidence, 0) 0.7: return enc return utf-8 def _read(self, file_path: str) - str: with open(file_path, rb) as f: raw f.read() enc self._detect_encoding(raw) for candidate in [enc, utf-8, gb18030, gbk, big5]: try: text raw.decode(candidate) return text.lstrip(\ufeff) except (UnicodeDecodeError, LookupError): continue return raw.decode(utf-8, errorsreplace) def _is_title(self, line: str, prev_blank: bool, next_blank: bool) - bool: line line.strip() if not line or len(line) self.max_title_len: return False for pat in self.title_patterns: if re.match(pat, line): return True if prev_blank and next_blank: if not re.search(r[。、,;:!?]$, line): return True return False def parse(self, file_path: str) - List[DocNode]: text self._read(file_path) lines text.split(\n) nodes [] heading_stack [] # [(level, title)] buffer [] def flush_buffer(): if buffer: content \n.join(buffer).strip() if content: path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typeparagraph, contentcontent, section_pathpath, )) buffer.clear() for i, line in enumerate(lines): prev_blank (i 0) or (not lines[i-1].strip()) next_blank (i len(lines)-1) or (not lines[i1].strip()) if not line.strip(): flush_buffer() continue if self._is_title(line, prev_blank, next_blank): flush_buffer() # 推断层级带编号的按编号深度否则默认 1 level self._infer_level(line) while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, line.strip())) nodes.append(DocNode( node_typetitle, contentline.strip(), levellevel, section_path .join(h[1] for h in heading_stack), )) else: buffer.append(line) flush_buffer() return nodes def _infer_level(self, line: str) - int: line line.strip() if re.match(r^第[一二三四五六七八九十百千][章篇], line): return 1 if re.match(r^第[一二三四五六七八九十百千][节回], line): return 2 if re.match(r^\d\.\d, line): return 2 if re.match(r^\d[、.], line): return 1 return 1这个解析器的核心逻辑是逐行扫描 缓冲区。遇到标题就 flush 缓冲区、更新标题栈遇到正文就攒进缓冲区遇到空行也 flush。这样能保证每个段落节点都带上正确的section_path。3.3 Markdown 解析器完整实现Markdown 解析器基于markdown-it-py重点是把 token 流转换成统一的DocNode列表。from markdown_it import MarkdownIt class MarkdownParser: def __init__(self): self.md MarkdownIt(commonmark, {html: False}) def parse(self, file_path: str) - List[DocNode]: with open(file_path, r, encodingutf-8) as f: content f.read() return self.parse_string(content) def parse_string(self, content: str) - List[DocNode]: tokens self.md.parse(content) nodes [] heading_stack [] # [(level, title)] i 0 while i len(tokens): token tokens[i] if token.type heading_open: level int(token.tag[1]) # 下一个 token 是 inline包含标题文本 title_text tokens[i1].content if i1 len(tokens) else while heading_stack and heading_stack[-1][0] level: heading_stack.pop() heading_stack.append((level, title_text)) nodes.append(DocNode( node_typetitle, contenttitle_text, levellevel, section_path .join(h[1] for h in heading_stack), )) i 3 # 跳过 heading_open, inline, heading_close continue if token.type paragraph_open: # 收集到 paragraph_close 之间的内容 para_content tokens[i1].content if i1 len(tokens) else path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typeparagraph, contentpara_content, section_pathpath, )) i 3 continue if token.type fence: # 代码块 path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typecode, contenttoken.content, section_pathpath, metadata{lang: token.info.strip()}, )) i 1 continue if token.type table_open: # 表格整体作为一个节点 table_lines [] j i while j len(tokens) and tokens[j].type ! table_close: if tokens[j].type inline: table_lines.append(tokens[j].content) j 1 path .join(h[1] for h in heading_stack) nodes.append(DocNode( node_typetable, content\n.join(table_lines), section_pathpath, )) i j 1 continue i 1 return nodes这里有个细节代码块和表格要单独处理。因为它们的切分策略和普通段落不一样——代码块最好整体保留不要从中间切开表格也是切开就失去意义了。所以我在解析阶段就把它们标记成独立的node_type后面切分时按类型走不同策略。3.4 归一化输出统一成下游可消费的格式两个解析器输出的都是DocNode列表但下游切分器、向量化器需要的是统一的文档对象。我一般会再包一层归一化把DocNode列表转成标准的Document对象。dataclass class Document: doc_id: str source: str nodes: List[DocNode] metadata: dict field(default_factorydict) def to_chunks(self, chunker): 交给切分器处理 return chunker.split(self) def normalize(nodes: List[DocNode], source: str) - Document: import hashlib doc_id hashlib.md5(source.encode()).hexdigest()[:16] total_chars sum(len(n.content) for n in nodes) return Document( doc_iddoc_id, sourcesource, nodesnodes, metadata{ total_nodes: len(nodes), total_chars: total_chars, has_structure: any(n.node_type title for n in nodes), } )doc_id用来源路径的 MD5 生成保证同一文件多次导入得到相同 ID方便去重和更新。has_structure字段标记文档是否有识别出标题结构后面切分时可以据此选择不同的切分策略——有结构的按章节切没结构的按固定长度切。3.5 完整调用示例与效果验证把上面几个类串起来跑一个完整流程def import_document(file_path: str) - Document: if file_path.endswith(.md) or file_path.endswith(.markdown): parser MarkdownParser() else: parser TxtParser() nodes parser.parse(file_path) return normalize(nodes, file_path) # 使用 doc import_document(sample.md) print(f文档 ID: {doc.doc_id}) print(f节点数: {len(doc.nodes)}) for node in doc.nodes[:5]: print(f[{node.node_type}] path{node.section_path}) print(f content: {node.content[:50]}...)跑完之后重点看几个指标节点总数是否合理太少说明结构没识别出来太多说明切得太碎、section_path 是否正确层级有没有错乱、内容有没有乱码。这三个指标正常解析这关就算过了。4. 常见问题与排查技巧实录4.1 编码乱码问题速查编码问题是最高频的我整理了一个速查表现象可能原因解决方法中文显示为锟斤拷UTF-8 读 GBK 文件改用 gb18030 读取中文显示为????编码不支持中文检查是否用了 latin-1开头多出\ufeffUTF-8 BOM用 utf-8-sig 或 lstrip部分字符乱码混合编码分段探测或人工确认繁体显示异常Big5 编码改用 big5 或 big5hkscs排查思路是先看乱码模式。锟斤拷是典型的 UTF-8/GBK 错配????是编码表里没有对应字符\ufeff是 BOM。根据模式反推原因比盲目试编码快得多。4.2 标题识别误判的三种典型场景启发式规则再稳也有翻车的时候。我遇到最多的三种误判第一种短句被误判为标题。比如正文里有一句今天天气不错前后恰好有空行长度也短就被当成标题了。解决办法是加一个标题不应以常见语气词结尾的规则或者要求标题候选行不含句末标点。第二种长标题被漏判。有些技术文档的标题很长超过 40 字就被我的规则排除了。解决办法是对带编号前缀的行放宽长度限制因为带编号的基本可以确定是标题。第三种代码块里的#被误判。txt 里如果有代码片段# 这是注释会被当成 Markdown 标题。这个在纯 txt 场景下很难完美解决我的做法是检测连续多行以#或//开头时判定为代码块整体跳过标题识别。提示标题识别没有 100% 准确的方案建议在解析结果里保留confidence字段低置信度的标题让用户人工确认。宁可漏判不可误判——误判的标题会污染 section_path比没有 section_path 更糟。4.3 Markdown 解析的边界情况处理Markdown 解析虽然用库但边界情况还是得自己处理嵌套列表markdown-it-py会把嵌套列表解析成多层bullet_list_open需要递归处理。我一般把整个列表作为一个节点保留原始文本不拆开。引用块blockquote里的内容我倾向于保留标记因为引用块往往有特殊语义如注意事项。HTML 内联虽然禁用了 HTML 解析但有些 Markdown 里混了br这类标签会被当纯文本。可以在解析后做一次正则清理。数学公式$...$和$$...$$在 commonmark 里不被识别需要装mdit-py-plugins的 dollarmath 插件。如果文档里有公式记得加上。4.4 大文件处理的内存优化处理几十 MB 的大 txt 文件时一次性读进内存可能爆。我的优化策略是流式读取 分批解析def parse_large_txt(file_path, batch_size10000): nodes [] buffer [] heading_stack [] with open(file_path, r, encodingutf-8) as f: for line in f: # 逐行读不一次性加载 # ... 同样的解析逻辑 if len(buffer) batch_size: # 处理一批 pass return nodes逐行读的好处是内存占用恒定跟文件大小无关。代价是没法做前后空行判断需要维护一个滑动窗口。我一般维护最近 3 行的状态足够判断空行了。4.5 解析质量的自检清单每次解析完我会跑一遍自检清单确认质量[ ] 编码是否正确有没有乱码字符[ ] 标题数量是否合理跟文档目录对比[ ] section_path 层级有没有错乱比如 h3 出现在 h1 前面[ ] 空节点有没有content 为空的节点要过滤[ ] 代码块和表格有没有被正确识别[ ] 总字符数跟原文件是否接近差太多说明丢内容了这个清单跑一遍基本能拦住九成的解析问题。剩下的疑难杂症就得靠人工抽查了。5. 从解析到切分的衔接要点5.1 结构化节点如何指导切分策略解析出来的DocNode列表直接决定了切分策略。我的原则是按节点类型走不同切分逻辑标题节点不单独成 chunk而是作为后续内容的section_path前缀段落节点按 token 数切分超过阈值就拆拆的时候保留 section_path代码块节点整体保留不拆。如果代码块本身超长按函数/类边界拆表格节点整体保留不拆。表格拆开就失去语义了这样切出来的 chunk每个都带完整的章节路径检索时上下文信息充足。5.2 chunk 元数据的继承与增强切分时chunk 的元数据从父节点继承同时做一些增强def make_chunk(node, parent_doc): return { content: node.content, doc_id: parent_doc.doc_id, source: parent_doc.source, section_path: node.section_path, node_type: node.node_type, # 增强字段 char_count: len(node.content), has_code: node.node_type code, depth: node.section_path.count( ) 1, }depth字段表示 chunk 在文档中的深度检索时可以据此调整权重——浅层节点章节标题附近往往概括性强深层节点细节多可以根据查询类型动态调整。5.3 为后续 PDF/Word 解析预留扩展点这一篇虽然只讲 txt 和 Markdown但架构上要为后续格式预留扩展点。我的做法是定义统一的 Parser 接口from abc import ABC, abstractmethod class BaseParser(ABC): abstractmethod def parse(self, file_path: str) - List[DocNode]: pass class PdfParser(BaseParser): def parse(self, file_path: str) - List[DocNode]: # 后续实现 pass所有解析器都实现parse方法返回统一的DocNode列表。这样归一化层和切分层完全不用改加新格式只需要写一个新的 Parser 类。这个扩展点在后续处理 PDF、Word、Excel 时会非常省事。6. 实操中的几点个人体会6.1 解析质量比解析速度重要得多我早期做 RAG 时追求解析速度用了各种优化结果检索效果一直上不去。后来把解析速度降下来加了编码探测、标题识别、结构保留这些慢操作检索准确率直接涨了一截。解析是离线的一次性工作慢一点没关系质量才是关键。一个文档解析花 10 秒还是 1 秒对用户体验没影响但解析质量差 10%检索效果可能差一半。6.2 保留原始文本永远是对的解析过程中我坚持保留原始文本。即使做了归一化、做了结构提取原始字符串也存一份。因为后面调试时你永远不知道会需要回看原文的哪个细节。我踩过的坑是早期为了省空间解析后只存结构化结果结果发现某个 chunk 内容异常想对比原文却找不到了只能重新解析。从那以后原始文本必存。6.3 中文场景的规则要自己调网上开源的解析规则大多是英文场景调好的直接拿来用中文会翻车。比如英文标题识别靠首字母大写中文没这个概念英文句子靠句号分句中文还有分号、顿号、省略号。中文场景的解析规则必须基于中文语料自己调。我一般会准备一个 100 篇左右的中文测试集每次改规则都跑一遍看准确率和召回率的变化。6.4 给解析结果加一个人工校验入口再好的自动解析也有出错的时候。我会在解析流程最后加一个人工校验入口把解析结果渲染成可视化界面让用户能快速浏览、标记错误、手动修正。这个入口看起来增加了工作量但实际用下来它极大提升了数据质量也降低了后续排查问题的成本。用户花 5 分钟校验可能省下后面 5 小时的调试。最后分享一个小技巧解析日志要详细。每次解析记录用了什么编码、识别了多少标题、有多少节点、耗时多少。这些日志在排查问题时是金矿。我现在的解析器日志详细到每一行被判定为什么类型出问题时直接看日志就能定位。这个习惯是从无数次解析结果不对但不知道哪一步错了的痛苦中养成的强烈建议你也加上。