1. 为什么 RAG 的第一步不是模型而是数据导入很多人一上来就研究向量库选型、Embedding 模型对比、重排序策略结果折腾了两周检索效果依然稀烂。我踩过这个坑之后才明白RAG 系统的上限在数据导入与解析阶段就已经被锁死了。你喂进去的是碎片化的乱码检索出来的必然是驴唇不对马嘴的片段再强的模型也救不回来。这一篇聚焦的是 RAG 数据管道最前端、也最容易被轻视的一环通用文本与结构化文本的导入和解析具体覆盖从纯 txt 到 Markdown 这两类最常见格式的处理。为什么先讲这两类因为在实际项目里txt 和 Markdown 是知识库语料的两大主力——前者来自各种导出、爬取、日志、词典、小说、配置说明后者来自技术文档、笔记系统、代码仓库的 README。把这两类吃透后面处理 PDF、Word、HTML、Excel 才有稳定的参照系。这篇文章适合谁看如果你正在搭自己的 RAG 知识库或者负责企业文档的入库治理又或者只是想把一堆散落的 txt 和 md 文件变成可检索的知识那接下来的内容可以直接抄作业。我会把每一步的为什么这么做讲清楚而不是甩一段代码让你自己猜。核心关键词 RAG、数据导入、解析、txt、Markdown 会贯穿全文但重点永远落在能落地的操作上。先说一个反直觉的结论解析的目标不是读出来而是读出来之后还能还原结构。txt 没有结构Markdown 有轻量结构这两者的处理策略完全不同。搞混了后面切分和检索全乱套。2. 数据导入的整体设计与格式选型思路2.1 先想清楚你的语料到底长什么样动手写代码之前我习惯先做一件事把待入库的文件全部列出来按格式、来源、体量分个类。这一步花不了十分钟但能省掉后面几天的返工。因为不同来源的 txt解析策略天差地别。举个真实的例子。我曾经接手一个知识库项目语料里混着三类 txt一类是从业务系统导出的结构化日志每行都是时间|模块|级别|内容这种固定分隔一类是从网页复制粘贴保存的说明文字段落之间靠空行分隔还有一类是词典文件格式是词条\t释义。如果我用同一套按行切分的逻辑处理第一类会切得刚好第二类会被切得稀碎第三类则会把词条和释义拆散。所以我的分类维度通常是这三个是否有固定分隔符有的话优先按分隔符解析成结构化字段没有的走纯文本流程。是否有层级语义Markdown 的标题层级、列表嵌套属于显式层级txt 里的缩进、编号属于隐式层级。单文件体量几 KB 的小文件和几百 MB 的大文件读取方式必须分开设计否则内存直接爆掉。提示分类这一步建议产出一张清单表把文件名、格式、来源、预估 token 数、解析策略都记下来。后面排查问题时这张表就是你的地图。2.2 为什么 txt 和 Markdown 要分开处理有人会问不都是文本吗统一按字符切分不就行了问题就出在统一两个字上。txt 的本质是无结构纯文本它不告诉你哪里是标题、哪里是正文、哪里是列表。你唯一能依赖的线索是换行符、空行、缩进、以及可能存在的分隔符。这意味着 txt 的解析核心是推断结构——从排版习惯里猜出作者的意图。Markdown 的本质是带标记的轻量结构化文本。#是标题-是列表是引用代码块有围栏。这些标记是作者显式写下的结构信息解析时应该忠实还原而不是当成普通字符切掉。把这两者混在一起处理最典型的后果就是Markdown 的标题被当成正文切进了 chunk导致每个片段的语义边界模糊或者 txt 里的空行被当成段落分隔结果把本该连在一起的句子拆开。所以我的原则很明确先按格式分流再各自走专用解析器最后统一输出成中间格式。2.3 中间格式为什么选 Markdown这里有个关键设计决策解析完的产物用什么格式承载我的答案是Markdown 作为统一中间表示。理由有三条。第一Markdown 天然表达层级标题、列表、引用、代码块一应俱全足以承载绝大多数文档结构。第二它是纯文本人眼可读调试的时候直接打开就能看不用借助任何工具。第三几乎所有主流的文本切分库和向量化流程都对 Markdown 有良好支持标题感知切分heading-aware splitting是现成能力。所以整条链路是这样的原始文件 → 格式识别 → 专用解析器 → 统一 Markdown → 结构感知切分 → 向量化入库。这一篇重点讲前四步也就是从 txt 和 Markdown 到统一 Markdown 的过程。阶段输入输出核心动作格式识别混合文件分类清单按扩展名与内容嗅探专用解析txt / md结构化对象推断或还原结构统一表示结构化对象标准 Markdown规范化标题与段落切分准备标准 Markdown待切分文本保留层级元数据3. txt 文件解析的核心细节与实操要点3.1 编码识别第一步就翻车的地方txt 解析最常见的翻车点不是逻辑而是编码。中文语料里 GBK、GB18030、UTF-8、UTF-8 with BOM 混着来直接open(file)读取轻则乱码重则直接抛异常中断整个导入流程。我的处理策略是先探测再读取。Python 里可以用chardet或charset-normalizer做编码探测但要注意探测不是百分百准确尤其是短文本。所以我会加一层兜底逻辑——探测出来的编码先试读如果解码失败就按优先级列表逐个尝试。from charset_normalizer import from_path def detect_encoding(file_path): result from_path(file_path).best() if result is None: return utf-8 return result.encoding def safe_read(file_path): encodings [detect_encoding(file_path), utf-8, gb18030, gbk, latin-1] for enc in encodings: try: with open(file_path, r, encodingenc) as f: return f.read(), enc except (UnicodeDecodeError, LookupError): continue raise ValueError(f无法解码文件: {file_path})注意latin-1放在最后因为它几乎不会解码失败但会把所有字节原样映射成字符属于保底不报错的方案。实际项目里如果走到这一步说明文件本身可能有问题需要人工介入。注意UTF-8 with BOM 的文件开头会多出一个不可见字符\ufeff它会在后续切分和检索时污染文本。读取后记得用text.lstrip(\ufeff)清掉。3.2 换行符归一化跨平台语料的隐形坑Windows 的\r\n、Linux 的\n、老 Mac 的\r这三种换行符混在一起会让你的段落切分逻辑彻底失效。我见过一个案例语料从 Windows 机器导出解析脚本跑在 Linux 上结果所有段落被当成一整块切分后每个 chunk 都是几千字检索精度惨不忍睹。处理方式很简单读取后统一替换text text.replace(\r\n, \n).replace(\r, \n)这一步必须在所有后续处理之前做顺序不能乱。归一化之后段落分隔就统一用\n\n来判断逻辑清晰很多。3.3 结构推断从排版习惯里还原作者意图txt 没有显式结构但作者写的时候是有意图的。我们要做的就是从排版线索里把这些意图猜出来。常用的线索有这几类空行通常表示段落分隔是最可靠的信号。缩进行首的空格或制表符可能表示列表、引用或代码。编号模式1.一、1这类开头大概率是列表或章节标题。分隔符---、、***这类整行符号通常是章节分隔线。固定分隔符|、\t、,这类说明是结构化数据而非自然语言。我通常会写一个规则引擎按优先级依次匹配。比如先判断是不是固定分隔符的结构化行如果是就按字段解析不是的话再看是不是编号开头是的话标记为列表项再不然看空行分段。import re def infer_structure(text): lines text.split(\n) blocks [] buffer [] for line in lines: stripped line.strip() if not stripped: if buffer: blocks.append({type: paragraph, content: \n.join(buffer)}) buffer [] continue if re.match(r^#{1,6}\s, stripped): if buffer: blocks.append({type: paragraph, content: \n.join(buffer)}) buffer [] level len(stripped) - len(stripped.lstrip(#)) blocks.append({type: heading, level: level, content: stripped.lstrip(# )}) elif re.match(r^(\d[\.、]|[一二三四五六七八九十]、|\d), stripped): buffer.append(stripped) else: buffer.append(stripped) if buffer: blocks.append({type: paragraph, content: \n.join(buffer)}) return blocks这段逻辑不复杂但覆盖了大部分常见 txt 排版。实际用的时候我会根据语料特点调整正则比如有些语料用【标题】这种方括号标记章节那就加一条规则。3.4 结构化 txt 的字段解析前面提到的那类时间|模块|级别|内容的日志属于结构化 txt。这类文件的解析思路完全不同——不要试图推断语义直接按分隔符拆字段然后决定哪些字段进正文、哪些进元数据。def parse_structured_txt(text, delimiter|): records [] for line in text.split(\n): if not line.strip(): continue parts line.split(delimiter) if len(parts) 4: continue records.append({ timestamp: parts[0].strip(), module: parts[1].strip(), level: parts[2].strip(), content: delimiter.join(parts[3:]).strip() }) return records这里有个经验内容字段可能本身包含分隔符所以用delimiter.join(parts[3:])把剩余部分重新拼起来而不是直接取parts[3]。这个细节不注意内容里带|的记录就会被截断。解析完之后content进正文用于向量化timestamp、module、level进元数据用于过滤。这样检索时可以支持只看某个模块的错误日志这类精确查询比纯语义检索靠谱得多。3.5 大文件的分块读取几百 MB 的 txt 直接read()进内存机器稍微差一点就卡死。我的做法是流式读取 分块处理def stream_read(file_path, chunk_size1024 * 1024): with open(file_path, r, encodingutf-8) as f: while True: chunk f.read(chunk_size) if not chunk: break yield chunk但流式读取有个麻烦块边界可能切断一个段落甚至一个句子。所以我会维护一个缓冲区把上一块末尾不完整的部分留到下一块拼接。判断完整的依据就是看块末尾有没有换行符。提示如果语料是超大日志文件其实更适合先按行切分落盘成小文件再逐个解析。这样既能并行处理又方便失败重试。4. Markdown 解析的层级还原与规范化4.1 Markdown 解析的核心目标保住层级Markdown 解析和 txt 最大的区别在于结构是现成的你的任务是别把它弄丢。很多人的做法是用正则把#、*、这些符号删掉只留纯文本结果层级信息全没了切分时只能按固定长度硬切语义边界荡然无存。正确的做法是解析成抽象语法树AST保留每个节点的类型和层级然后再决定怎么输出。Python 里我常用markdown-it-py或mistune它们能把 Markdown 解析成结构化的 token 流。from markdown_it import MarkdownIt def parse_markdown(text): md MarkdownIt() tokens md.parse(text) structure [] for token in tokens: if token.type heading_open: structure.append({type: heading, level: int(token.tag[1])}) elif token.type inline: if structure and structure[-1][type] heading: structure[-1][content] token.content else: structure.append({type: paragraph, content: token.content}) return structure拿到这个结构列表之后每个标题、每个段落的位置和层级都清清楚楚。切分的时候就可以按标题边界来切保证每个 chunk 不会跨越章节。4.2 标题层级规范化从 H1 到 H6 的坑Markdown 允许 H1 到 H6 六级标题但实际文档里经常出现层级跳跃——比如 H1 下面直接跟 H3中间没有 H2。这在渲染时问题不大但在切分和检索时会出问题你无法确定 H3 到底属于哪个父章节。我的处理方式是构建标题树补齐缺失层级。具体做法是维护一个栈遇到新标题时如果它的层级比栈顶深超过一级就插入虚拟的中间层级如果比栈顶浅就弹出直到找到合适的父节点。def build_heading_tree(blocks): root {level: 0, title: root, children: [], content: []} stack [root] for block in blocks: if block[type] heading: level block[level] while stack and stack[-1][level] level: stack.pop() node {level: level, title: block[content], children: [], content: []} stack[-1][children].append(node) stack.append(node) else: stack[-1][content].append(block.get(content, )) return root这样每个段落都能追溯到完整的标题路径比如RAG 教程 数据导入 txt 解析。这个路径在切分时可以作为元数据附在 chunk 上检索时能大幅提升上下文准确性。4.3 代码块与行内代码的保护技术文档里代码块是重灾区。如果按普通文本切分代码块很容易被从中间切断导致检索出来的片段是半截代码毫无用处。所以解析时必须把代码块识别为原子单元不可分割。markdown-it-py会把代码块解析成fence类型的 token行内代码是code_inline。处理时把它们单独标记出来def extract_code_blocks(tokens): code_blocks [] for i, token in enumerate(tokens): if token.type fence: code_blocks.append({ language: token.info.strip() or text, content: token.content }) return code_blocks切分时代码块要么整体进一个 chunk要么单独成块绝不能跨块。我一般会给代码块单独设一个切分策略如果代码块本身超过 chunk 上限就按函数或逻辑块边界切而不是按字符数硬切。注意行内代码like this虽然短但也不能在切分时被拆开。处理方式是先把行内代码替换成占位符切分完再还原避免切分点落在反引号中间。4.4 列表与嵌套结构的处理Markdown 的列表支持多层嵌套-、*、和数字编号混用。嵌套列表在语义上是一个整体切分时如果从中间断开会丢失层级关系。我的做法是把整个列表块作为一个语义单元如果它超过 chunk 上限再按顶层列表项切分每个顶层项连同它的子项一起走。这样至少保证每个 chunk 里的列表是完整的子树。def flatten_list(items, prefix): lines [] for item in items: lines.append(f{prefix}- {item[content]}) if item.get(children): lines.extend(flatten_list(item[children], prefix )) return lines输出的时候保留缩进这样还原成 Markdown 时层级还在。别小看这个缩进检索时用户看到的是一个完整的列表而不是一堆散落的短句。4.5 表格、引用与数学公式的特殊处理Markdown 表格在解析时会被拆成thead、tbody、tr、td等 token。表格的语义完整性很重要切分时应该整表保留。如果表格特别大按行切分时要重复表头否则检索出来的片段没有列名根本看不懂。引用块的处理相对简单保留引用标记即可但要注意嵌套引用的层级。数学公式分两种行内$...$和块级$$...$$。块级公式必须整体保留行内公式不能被切分点打断。如果你的知识库涉及技术文档这两类内容的保护一定要做。元素类型是否可切分切分策略注意事项标题否作为边界保留层级路径段落是按句切分避免切断句子代码块否整体保留超长按逻辑切列表谨慎按顶层项切保留子项完整表格否整表保留大表重复表头块级公式否整体保留不可打断5. 统一 Markdown 输出的实操流程5.1 从解析结果到标准 Markdown 的转换前面 txt 和 Markdown 各自解析完产物都是结构化的块列表。接下来要做的是把它们统一转换成标准 Markdown。这一步的价值在于下游只需要处理一种格式切分、向量化、检索的逻辑可以完全复用。转换规则很直接标题输出#加内容段落直接输出列表加-前缀代码块用围栏包裹表格还原成管道语法。关键是保持层级和顺序不变。def blocks_to_markdown(blocks): lines [] for block in blocks: btype block[type] if btype heading: lines.append(# * block[level] block[content]) elif btype paragraph: lines.append(block[content]) elif btype list: lines.extend(flatten_list(block[items])) elif btype code: lines.append(f{block.get(language, )}) lines.append(block[content]) lines.append() lines.append() return \n.join(lines)输出时每个块之间留一个空行这是 Markdown 的标准段落分隔方式也方便后续按\n\n切分。5.2 元数据的注入与保留光有正文不够RAG 检索时元数据往往比正文还重要。我习惯在转换阶段就把元数据注入进去用 YAML front matter 的形式放在文件开头--- source: logs/2024-01-app.txt format: structured_txt encoding: gb18030 heading_path: 系统日志 应用日志 record_count: 1523 ---这样每个文件自带来源信息切分后每个 chunk 都能继承这些元数据。检索时可以用source过滤特定文件用heading_path限定章节范围精度提升非常明显。提示heading_path这种字段建议在切分阶段动态更新让每个 chunk 记录自己所属的完整标题路径而不是整个文件的路径。5.3 完整流程串起来把前面的步骤串成一条流水线大概是这样的扫描目录按扩展名和内容嗅探分类文件。txt 文件走编码探测、换行归一化、结构推断、字段解析。Markdown 文件走 AST 解析、标题树构建、代码块与列表提取。两者统一转换成标准 Markdown注入元数据。输出到统一的中间目录等待切分。def process_file(file_path): ext file_path.suffix.lower() if ext .txt: text, enc safe_read(file_path) text text.replace(\r\n, \n).replace(\r, \n).lstrip(\ufeff) blocks infer_structure(text) meta {source: str(file_path), format: txt, encoding: enc} elif ext in (.md, .markdown): text file_path.read_text(encodingutf-8) blocks parse_markdown(text) meta {source: str(file_path), format: markdown} else: return None md_content blocks_to_markdown(blocks) front_matter ---\n \n.join(f{k}: {v} for k, v in meta.items()) \n---\n\n return front_matter md_content这段代码不复杂但每一步都有讲究。实际项目里我会加上日志、异常捕获、进度显示方便排查问题。5.4 批量处理的性能与稳定性单文件处理没问题但几千个文件批量跑的时候性能和稳定性就成了问题。我的经验是并行处理用concurrent.futures的线程池IO 密集型任务开 8 到 16 个线程比较合适太多反而因为磁盘争抢变慢。失败隔离单个文件解析失败不能中断整个流程用 try-except 包住失败的记录到单独的错误日志最后统一处理。断点续传处理前先检查输出目录已处理的文件跳过避免重复劳动。进度可见每处理 100 个文件打印一次进度长任务没有反馈会让人焦虑。from concurrent.futures import ThreadPoolExecutor, as_completed def batch_process(file_list, output_dir, max_workers8): failed [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(process_file, f): f for f in file_list} for i, future in enumerate(as_completed(futures)): file_path futures[future] try: result future.result() if result: out_path output_dir / (file_path.stem .md) out_path.write_text(result, encodingutf-8) except Exception as e: failed.append((str(file_path), str(e))) if (i 1) % 100 0: print(f已处理 {i 1}/{len(file_list)}) return failed跑完之后检查failed列表逐个分析失败原因。常见的失败原因就那么几类编码问题、文件损坏、权限不足、路径过长。处理完这些你的数据导入管道基本就稳了。6. 常见问题与排查技巧实录6.1 编码乱码问题速查编码问题占了 txt 解析故障的一大半。下面这张表是我这些年攒下来的排查清单现象可能原因排查方法解决方案中文全是问号用 latin-1 解码检查读取编码改用 gb18030 重试开头有奇怪字符UTF-8 BOM查看首字节lstrip(\ufeff)部分字符乱码编码探测错误抽样人工核对手动指定编码整个文件报错二进制文件误判检查文件头排除或转格式换行全丢失换行符不统一查看原始字节归一化处理我的习惯是任何编码相关的异常先把文件用十六进制编辑器打开看前几个字节。UTF-8 BOM 是EF BB BFUTF-16 是FF FE或FE FFGBK 的中文通常以B0到F7开头。看一眼心里就有数了。6.2 结构推断失败的典型场景结构推断靠的是规则规则就有覆盖不到的地方。我遇到过几种典型失败第一种是全篇没有空行。有些从数据库导出的 txt每行都是独立记录中间没有空行。这时候按空行分段就完全失效得改成按行处理每行一个块。第二种是空行过多。有些文档每个句子后面都空一行按空行分段会把句子拆散。这时候要判断如果连续多个块都很短考虑合并。第三种是缩进不一致。同一份文档里有的用空格缩进有的用制表符有的混用。处理时统一转成空格再按固定宽度判断层级。提示结构推断没有万能规则最好的办法是先抽样看 20 个文件总结出这个语料的排版规律再针对性写规则。通用规则只能兜底专用规则才有效。6.3 Markdown 解析的边界情况Markdown 解析看起来简单边界情况却不少标题里带链接# [标题](url)解析时要提取纯文本别把链接语法带进去。代码块里的反引号代码内容本身包含 时围栏要用更多反引号包裹。列表项跨多段列表项下面跟一个空行再接段落这个段落仍属于该列表项不能拆开。HTML 混入Markdown 里嵌div这类 HTML解析器可能直接跳过需要单独处理。表格对齐行|---|---|这行是格式标记不是内容解析时要识别并跳过。这些情况我都是踩过坑才记住的。建议你拿一份复杂的 Markdown 文档跑一遍解析把输出和原文对照看看哪里丢了、哪里多了很快就能发现问题。6.4 大文件与特殊字符的处理经验大文件处理的核心是别一次性读进内存。除了前面说的流式读取还有个技巧是先按行数切分成小文件再并行处理。比如一个 500MB 的日志按 10 万行切一个文件切成几十个小文件处理起来又快又稳。特殊字符方面有几个必须处理零宽字符\u200b、\u200c、\u200d这些不可见字符会污染检索结果读取后统一清除。全角空格\u3000中文文档里常见切分时容易被当成普通字符建议归一化成半角。控制字符\x00到\x1f除了\n、\t之外基本都是噪音直接过滤。import unicodedata def clean_text(text): text text.replace(\u3000, ) text .join(c for c in text if unicodedata.category(c) ! Cf) text .join(c for c in text if c \n or c \t or unicodedata.category(c)[0] ! C) return text这段清洗逻辑放在解析之后、转换之前能去掉大部分隐形噪音。别小看这些字符它们会让你的检索结果莫名其妙地匹配不上。6.5 我踩过的三个真实坑第一个坑用split(\n\n)切段落结果 Windows 换行没归一化切了个寂寞。整个文件变成一个块切分后每个 chunk 几千字。后来加了换行归一化才解决。第二个坑Markdown 标题层级跳跃导致标题树错乱。H1 下面直接 H3我的栈逻辑没处理这种情况结果 H3 被挂到了错误的父节点下。后来加了虚拟层级补齐才修好。第三个坑结构化 txt 的内容字段包含分隔符被截断了。日志内容里带了|按split(|)之后内容只剩前半截。改成join剩余部分才修复。这个坑很隐蔽因为大部分记录没问题只有少数带分隔符的记录出错不仔细看根本发现不了。这三个坑的共同点是问题不在逻辑复杂度而在细节疏忽。数据导入这活儿拼的就是细节。7. 从导入到切分的衔接要点7.1 切分前必须确认的三件事统一 Markdown 输出之后下一步就是切分。但在切分之前有三件事必须确认第一标题层级是否完整。随便打开几个输出文件看看标题树是不是连贯的有没有孤立的深层标题。第二代码块和表格是否完整。搜索一下围栏符号确认成对出现表格的管道符数量是否一致。第三元数据是否正确注入。检查 front matter 的字段是否齐全heading_path是否准确。这三件事花不了几分钟但能避免后面大量的返工。我见过太多人跳过这步切分完发现结构全乱只能从头再来。7.2 为切分预留的元数据设计切分阶段需要什么元数据导入阶段就要准备好。我的经验是至少准备这几类来源信息文件路径、格式、编码用于溯源和过滤。层级信息标题路径用于限定检索范围。类型信息段落、代码、表格、列表用于差异化处理。统计信息字符数、token 数用于控制 chunk 大小。这些信息在导入阶段都是现成的顺手记下来切分时直接用不用再回头解析一遍。7.3 一个可复用的目录结构最后分享一个我常用的目录结构清晰且好维护project/ ├── raw/ # 原始文件 │ ├── txt/ │ └── markdown/ ├── parsed/ # 解析后的统一 Markdown │ ├── txt/ │ └── markdown/ ├── chunks/ # 切分后的片段 ├── logs/ # 处理日志 │ ├── success.log │ └── failed.log └── scripts/ # 处理脚本 ├── parse.py └── chunk.pyraw只读不动parsed是解析产物chunks是切分产物每一层职责分明。出问题时可以逐层排查定位到具体环节。这个结构我用了好几年从没想过换。我个人在实际操作中的体会是数据导入这活儿慢就是快。前期在编码、结构、元数据上多花的时间后面都会以检索精度的形式还给你。急着往向量库里灌数据最后只会花更多时间调检索效果。下一篇我会接着讲切分策略和向量化入库把这条链路走完。