RAG数据管道第一关：从txt到Markdown的通用文本解析与LangChain Loader实战

1. 为什么数据导入与解析是 RAG 系统的第一道生死关

做 RAG 项目的人都有一个共识：检索效果差，八成不是模型不行，而是数据没处理好。我见过太多团队花大价钱调 embedding 模型、换向量库、加 rerank，结果回头一看，原始文档里全是乱码、断行、页眉页脚混在正文里，切出来的 chunk 语义支离破碎，再强的模型也救不回来。

这一篇聚焦的是 RAG 数据管道最前端、也最容易被低估的环节：从 txt 到 Markdown 的通用文本与结构化解析。说白了，就是把各种格式的原始文件，变成干净、有结构、可被 LangChain 的 Document Loader 正确读取的文本。这件事听起来简单，但真正做过的人都知道，坑多到能写一本书。

这篇文章适合三类人：一是刚接触 RAG、准备搭第一个知识库的开发者；二是已经跑通了 demo、但检索质量始终上不去的工程师；三是需要处理大量异构文档（合同、手册、论文、网页存档）的数据侧同学。我会围绕 LangChain 的 Document 抽象、Loader 选型、Markdown 作为中间格式的价值，以及实际解析中的参数计算和避坑经验，把这条链路讲透。

先明确一个核心观点：Markdown 不是终点，而是 RAG 数据管道里最理想的"中间表示层"。它比纯 txt 多了结构信息（标题层级、列表、表格、代码块），又比 HTML、PDF 的原始结构更干净、更接近自然语言。把各种格式先归一化到 Markdown，再做 chunk 切分，是目前工程上最稳的一条路。

2. RAG 数据管道的整体设计与格式选型逻辑

2.1 从原始文件到 Document 对象的完整链路

LangChain 里所有数据的起点都是Document对象，它只有两个核心字段：page_content（文本内容）和metadata（元数据字典）。别小看这两个字段，整个 RAG 的检索质量、引用溯源、权限过滤，全都挂在这上面。

一条完整的导入链路通常是这样：

原始文件（pdf/docx/txt/html/md） ↓ Loader 读取 Document 对象（含 page_content + metadata） ↓ 清洗与归一化 标准化 Markdown 文本 ↓ TextSplitter 切分 Chunk 列表（每个 chunk 仍是 Document） ↓ Embedding 向量库

这里有个关键决策点：清洗和归一化到底放在 Loader 之前还是之后。我的经验是，能放在 Loader 之前就放之前。原因很简单，Loader 本身会引入噪声，比如 PDF 解析器会把页眉页脚、页码、水印都塞进page_content，你拿到 Document 之后再清洗，等于在垃圾堆里挑东西。更好的做法是先用专门的解析工具把文件转成干净的 Markdown，再用UnstructuredMarkdownLoader或TextLoader读进来。

2.2 为什么选 Markdown 作为中间格式

很多人会问，为什么不直接用 txt？txt 确实最干净，但它丢掉了所有结构信息。RAG 的 chunk 切分策略高度依赖结构：一个二级标题下的内容应该尽量切在一起，表格不应该被拦腰截断，代码块必须保持完整。这些信息在 txt 里全没了，你只能靠固定长度硬切，语义完整性必然受损。

Markdown 的优势在于它是纯文本 + 轻量结构的组合。标题用#表示，列表用-或1.，表格用|，代码块用 ``` 包裹。这些标记既不影响人阅读，又能被程序精确解析。LangChain 的MarkdownHeaderTextSplitter就是专门吃这个结构的，它能按标题层级切分，并把标题路径写进每个 chunk 的 metadata，检索时可以直接按章节过滤。

对比一下几种常见中间格式：

结论很清晰：只要你的源数据是文档类内容，Markdown 就是性价比最高的中间格式。

2.3 Loader 选型的三个判断维度

LangChain 提供了几十种 Loader，选哪个不是看名字顺眼，而是看三个维度：

• 数据源类型：本地文件、网页、数据库、云盘，决定了 Loader 的大类。
• 是否需要保留结构：如果后续要用标题切分，就必须选能输出 Markdown 的 Loader。
• metadata 丰富度：有些 Loader 会带上文件路径、修改时间、页码，这些对溯源和过滤极其有用。

举个实际例子，同样是读 Markdown 文件，TextLoader和UnstructuredMarkdownLoader差别很大。前者只是把整个文件当纯文本读进来，page_content里还带着#符号；后者会解析结构，把标题、段落、列表拆成独立的 element，metadata 里带上category字段。如果你后续要用MarkdownHeaderTextSplitter，其实用TextLoader就够了，因为切分器自己会解析#。但如果你要做更细的元素级处理，UnstructuredMarkdownLoader更合适。

提示：不要迷信"功能最全"的 Loader。功能越多，引入的依赖越重，解析结果越不可控。能用简单 Loader 解决的，就别上重型工具。

3. 通用文本解析的核心细节与实操要点

3.1 txt 文件的编码陷阱与处理

txt 看起来最简单，但编码问题能坑死人。国内很多老系统导出的 txt 是 GBK 或 GB2312 编码，你直接用 UTF-8 读会报UnicodeDecodeError，或者读出来一堆乱码。更麻烦的是，有些文件是混合编码，前半段 GBK 后半段 UTF-8，这种情况没有银弹。

我的处理策略是分三步走：

1. 先探测编码：用chardet库检测文件编码，拿到置信度最高的结果。
2. 带容错读取：用errors='replace'或errors='ignore'兜底，避免程序直接崩。
3. 人工抽检：对置信度低于 0.8 的文件，抽样看几行，确认没问题再批量处理。

import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: raw = f.read(100000) # 只读前100KB，够判断了 result = chardet.detect(raw) return result['encoding'], result['confidence'] def safe_read(file_path): encoding, confidence = detect_encoding(file_path) if confidence < 0.8: print(f"警告：{file_path} 编码置信度仅 {confidence:.2f}，建议人工检查") with open(file_path, 'r', encoding=encoding, errors='replace') as f: return f.read()

这里有个经验值：只读前 100KB 做编码探测就够了。整文件读取在大文件上很慢，而且编码通常是一致的，没必要全读。置信度阈值我一般设 0.8，低于这个值就标记出来人工看，别偷懒。

3.2 换行符与空白字符的归一化

跨平台文件的换行符是个隐形杀手。Windows 用\r\n，Linux 用\n，老 Mac 用\r。如果不统一，你在做正则匹配和切分时会遇到各种诡异问题。比如按\n\n切段落，Windows 文件里实际是\r\n\r\n，匹配不上。

归一化的标准动作：

import re def normalize_text(text): # 统一换行符 text = text.replace('\r\n', '\n').replace('\r', '\n') # 合并连续空行（超过2个的压成2个） text = re.sub(r'\n{3,}', '\n\n', text) # 去掉行尾空白 text = '\n'.join(line.rstrip() for line in text.split('\n')) # 全角空格转半角（中文文档常见） text = text.replace('\u3000', ' ') return text

注意：全角空格转半角这一步要谨慎。如果你的文档里有对齐用的全角空格（比如中文排版里的缩进），转掉之后排版会乱。我的建议是只在明确知道文档没有排版依赖时才做这一步。

3.3 从纯文本到 Markdown 的结构化转换

纯 txt 转 Markdown，核心是识别出隐含的结构。很多 txt 文档其实是有结构的，只是没用 Markdown 标记表达出来。常见的模式有：

• 用数字编号的章节：第一章、1. 概述、1.1 背景
• 用特殊符号分隔的标题：====、----、****
• 用缩进表示的层级

我写过一个简单的启发式转换器，核心逻辑是识别这些模式并替换成对应的 Markdown 标记：

import re def txt_to_markdown(text): lines = text.split('\n') result = [] for line in lines: stripped = line.strip() # 一级标题：第X章 / 第X部分 if re.match(r'^第[一二三四五六七八九十百]+[章部分]', stripped): result.append(f'# {stripped}') # 二级标题：1. xxx / 1、xxx elif re.match(r'^\d+[\.、]\s*\S', stripped) and len(stripped) < 40: result.append(f'## {stripped}') # 三级标题：1.1 xxx elif re.match(r'^\d+\.\d+\s+\S', stripped) and len(stripped) < 40: result.append(f'### {stripped}') # 列表项 elif re.match(r'^[•·▪]\s*', stripped): result.append(f'- {re.sub(r"^[•·▪]\s*", "", stripped)}') else: result.append(line) return '\n'.join(result)

这里的关键判断是标题长度阈值。我设的是 40 个字符，因为真正的标题通常很短，而正文里出现的"1. 首先我们要..."这种句子往往很长。这个阈值需要根据你的文档特点调整，没有万能值。

3.4 表格与代码块的识别处理

txt 里的表格通常是用空格或制表符对齐的，转 Markdown 表格需要先识别列边界。简单的方法是找连续多行都有相似的空格分布，但这在中文文档里很难做，因为中英文混排时字符宽度不一致。

我的建议是：如果 txt 里表格不多，别硬转，保留原样并在 metadata 里标记。强行转换出来的 Markdown 表格往往是错的，反而误导后续处理。如果表格确实重要，考虑用专门的工具（比如把源文件重新导出为 CSV 或 Excel）单独处理。

代码块的识别相对简单，找连续的、有统一缩进（通常是 4 个空格或 1 个 Tab）的行，或者找被包裹的内容。但 txt 里很少有标记，所以主要靠缩进判断。这里要小心，中文文档里的引用段落也常用缩进，容易误判。我的做法是结合内容特征，比如代码行通常包含{}、()、=、;等符号，用这个辅助判断。

4. LangChain Document Loader 实战配置

4.1 TextLoader 与 UnstructuredMarkdownLoader 的取舍

先看两个 Loader 的实际输出差异。假设有一个test.md：

# 产品手册 ## 安装步骤 1. 下载安装包 2. 双击运行 ## 配置说明 修改 config.yaml 文件。

用TextLoader读取：

from langchain_community.document_loaders import TextLoader loader = TextLoader("test.md", encoding="utf-8") docs = loader.load() print(docs[0].page_content) # 输出：整个文件内容，包含 # 和 ## 符号 print(docs[0].metadata) # 输出：{'source': 'test.md'}

用UnstructuredMarkdownLoader读取：

from langchain_community.document_loaders import UnstructuredMarkdownLoader loader = UnstructuredMarkdownLoader("test.md") docs = loader.load() # 输出：多个 Document，每个对应一个 element # metadata 里带 category 字段，如 'Title'、'ListItem'、'NarrativeText'

选择逻辑很明确：如果你后续要用MarkdownHeaderTextSplitter按标题切分，用TextLoader就够了，因为切分器自己会解析#符号。如果你要做元素级的精细处理（比如单独提取所有列表项），才需要UnstructuredMarkdownLoader。

UnstructuredMarkdownLoader有个坑：它依赖unstructured库，这个库在不同版本间行为差异很大，而且对中文的支持时好时坏。我遇到过中文标题被识别成NarrativeText而不是Title的情况。所以生产环境里，我倾向于用TextLoader+ 自己写的解析逻辑，可控性更强。

4.2 DirectoryLoader 批量导入的正确姿势

实际项目里不可能一个个文件手动加载，DirectoryLoader是批量处理的主力。但它的默认配置有几个坑：

from langchain_community.document_loaders import DirectoryLoader, TextLoader loader = DirectoryLoader( "./docs", glob="**/*.md", # 递归匹配所有 md 文件 loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, show_progress=True, # 大目录一定要开 use_multithreading=True, # 多线程加速 max_concurrency=4, # 并发数，别设太高 silent_errors=True, # 单个文件失败不中断整体 ) docs = loader.load()

几个关键参数的解释：

• glob：**/*.md表示递归所有子目录，只写*.md只匹配当前目录。
• silent_errors：设为True时，某个文件读取失败会跳过并继续，不会让整个任务崩掉。生产环境必开。
• max_concurrency：并发数不是越高越好。IO 密集型任务设 4-8 比较稳，设太高反而因为文件句柄竞争变慢。
• loader_kwargs：把编码等参数透传给底层 Loader，这个很容易漏。

提示：silent_errors=True虽然方便，但会吞掉错误信息。建议在加载完成后，单独跑一遍校验，统计成功和失败的文件数，对失败的文件单独排查。

4.3 metadata 的定制与增强

默认的 metadata 只有source一个字段，这在生产环境远远不够。我通常会在加载后手动增强 metadata，加上这些字段：

import os from datetime import datetime def enrich_metadata(doc, file_path): stat = os.stat(file_path) doc.metadata.update({ "file_name": os.path.basename(file_path), "file_dir": os.path.dirname(file_path), "file_size": stat.st_size, "modified_time": datetime.fromtimestamp(stat.st_mtime).isoformat(), "file_type": os.path.splitext(file_path)[1], }) return doc

这些字段的用途：file_name用于引用溯源，modified_time用于增量更新（只处理修改过的文件），file_dir用于按目录做权限过滤。别小看这些，等到你要做"只检索某个部门文档"或者"只返回最近更新的内容"时，没有这些 metadata 就得推倒重来。

5. Markdown 结构化切分的参数计算与实操

5.1 MarkdownHeaderTextSplitter 的工作原理

这个切分器的逻辑是：扫描 Markdown 文本，遇到指定级别的标题就切一刀，并把标题路径记录到 metadata 里。配置长这样：

from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "h1"), ("##", "h2"), ("###", "h3"), ] splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False, # 是否在正文里去掉标题行 ) chunks = splitter.split_text(markdown_text)

strip_headers这个参数很关键。设为False时，标题行会保留在 chunk 内容里；设为True时，标题只进 metadata，正文里没有。我的建议是设为 False，因为标题本身携带语义信息，保留在正文里能让 embedding 更准确。比如"安装步骤"这个标题，如果只放在 metadata 里，正文 chunk 可能只有"1. 下载 2. 运行"，语义很弱；保留标题后，chunk 变成"安装步骤 1. 下载 2. 运行"，语义完整多了。

5.2 标题层级与 chunk 大小的平衡

纯按标题切分有个问题：有些章节特别长，切出来的 chunk 超过 embedding 模型的 token 上限。比如一个"常见问题"章节下面有 50 个问答，全切在一起可能有上万 token。

解决方案是两级切分：先按标题切，再对超长的 chunk 按字符数二次切分。

from langchain_text_splitters import RecursiveCharacterTextSplitter # 第一级：按标题切 header_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")], strip_headers=False, ) header_chunks = header_splitter.split_text(markdown_text) # 第二级：对超长 chunk 按字符切 char_splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=100, separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""], ) final_chunks = [] for chunk in header_chunks: if len(chunk.page_content) > 1000: sub_chunks = char_splitter.split_documents([chunk]) final_chunks.extend(sub_chunks) else: final_chunks.append(chunk)

参数怎么定？chunk_size=800是基于中文的经验值。中文一个字符大约对应 1-2 个 token（取决于 tokenizer），800 字符大概 800-1600 token，留足余量给 embedding 模型的 512 或 8192 上限。chunk_overlap=100是防止语义在边界处断裂，一般取 chunk_size 的 10%-15%。

separators的顺序很重要，从粗到细排列。先按段落切（\n\n），不行再按行（\n），再不行按句子（。！？），最后才按字符硬切。这样能最大程度保证语义完整。

5.3 表格和代码块的保护策略

MarkdownHeaderTextSplitter和RecursiveCharacterTextSplitter都不认识表格和代码块，会把它们拦腰截断。一个表格被切成两半，检索时两边都语义不全，这是很常见的问题。

我的处理方式是在切分前先把表格和代码块"保护"起来，用占位符替换，切分后再还原：

import re def protect_blocks(text): blocks = [] def replacer(match): blocks.append(match.group(0)) return f"__BLOCK_{len(blocks)-1}__" # 保护代码块 text = re.sub(r'```[\s\S]*?```', replacer, text) # 保护表格（连续多行以 | 开头） text = re.sub(r'(\|.*\|\n)+', replacer, text) return text, blocks def restore_blocks(text, blocks): for i, block in enumerate(blocks): text = text.replace(f"__BLOCK_{i}__", block) return text

这个技巧实测很有效，尤其是技术文档里代码块多的情况。缺点是占位符本身可能被切分器切开，所以占位符要尽量短且不含特殊字符。

6. 常见问题与排查技巧实录

6.1 解析结果乱码或丢字

这是最高频的问题，排查顺序如下：

我的经验是，90% 的乱码问题都是编码问题。先用 chardet 跑一遍，基本能定位。剩下 10% 是文件本身损坏，这种只能跳过或找源文件重新导出。

6.2 chunk 切分后语义断裂

典型表现是检索时明明文档里有答案，但就是召不回来。原因通常是 chunk 切得太碎，或者切在了语义边界上。

排查方法：随机抽 10 个 chunk，人工读一遍，看是否每段都能独立表达一个完整意思。如果发现大量 chunk 以"因此"、"所以"、"但是"开头，说明上一句被切走了，需要增大chunk_overlap。

另一个技巧是在 chunk 前面拼接标题路径。比如一个 chunk 来自"产品手册 > 安装步骤 > Windows 环境"，就在 chunk 内容前加上这个路径。这样即使 chunk 本身语义不完整，加上路径后也能被正确检索到。

def add_header_path(chunk): headers = [] for key in ["h1", "h2", "h3"]: if key in chunk.metadata: headers.append(chunk.metadata[key]) if headers: prefix = " > ".join(headers) chunk.page_content = f"[{prefix}]\n{chunk.page_content}" return chunk

6.3 大文件加载内存溢出

处理几百 MB 的 txt 或几千个文件时，loader.load()一次性把所有 Document 读进内存，很容易 OOM。解决方案是用lazy_load()：

for doc in loader.lazy_load(): process(doc) # 逐个处理，处理完就释放

lazy_load返回生成器，每次只产出一个 Document，内存占用恒定。配合流式写入向量库，可以处理任意大小的数据集。

注意：lazy_load和use_multithreading不能同时用，多线程模式下必须一次性加载。所以大文件场景要二选一：要么单线程流式，要么多线程分批。

6.4 增量更新时如何避免重复导入

生产环境里文档会不断更新，全量重跑既慢又浪费。我的做法是用modified_time做增量判断：

import json import os STATE_FILE = "import_state.json" def load_state(): if os.path.exists(STATE_FILE): with open(STATE_FILE) as f: return json.load(f) return {} def should_process(file_path, state): mtime = os.stat(file_path).st_mtime return state.get(file_path, 0) < mtime def update_state(file_path, state): state[file_path] = os.stat(file_path).st_mtime with open(STATE_FILE, "w") as f: json.dump(state, f)

这个状态文件记录了每个文件的最后处理时间，只有修改过的文件才会重新导入。配合向量库的delete接口，先删旧 chunk 再插新 chunk，就能实现干净的增量更新。

6.5 常见问题速查表

7. 我在实际项目里踩过的坑和总结的经验

说几个文档里不会写、但实际做项目一定会遇到的坑。

第一个是中文标点的处理。中文文档里的句号是"。"，问号是"？"，这些和英文标点不一样。如果你在separators里只写了英文标点，中文文档就切不开，只能按字符硬切。我现在的标准配置里，中英文标点都列上，顺序是["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " ", ""]。

第二个是空 chunk 的过滤。切分后经常出现只包含空白字符或单个标点的 chunk，这些 chunk 进向量库纯属浪费空间，还会干扰检索。加载后一定要过滤：

final_chunks = [c for c in final_chunks if len(c.page_content.strip()) > 20]

阈值我一般设 20 个字符，低于这个长度的 chunk 基本没有检索价值。

第三个是metadata 的序列化问题。向量库对 metadata 的类型有要求，通常只支持字符串、数字、布尔值。如果你往 metadata 里塞了datetime对象或Path对象，写入时会报错。统一转成字符串，这是最稳的做法。

第四个是文件名的特殊字符。有些文件名带空格、中文、括号，在某些 Loader 里会出问题。我的做法是在导入前统一重命名，用 UUID 或哈希值做文件名，原始文件名存进 metadata。这样既避免了路径问题，又保留了溯源信息。

最后分享一个实用的小工具函数，把整个流程串起来：

def build_rag_documents(directory): loader = DirectoryLoader( directory, glob="**/*.md", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, silent_errors=True, show_progress=True, ) raw_docs = loader.load() header_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")], strip_headers=False, ) char_splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=100, separators=["\n\n", "\n", "。", "！", "？", "；", ".", "!", "?", ";", " ", ""], ) final = [] for doc in raw_docs: header_chunks = header_splitter.split_text(doc.page_content) for hc in header_chunks: hc.metadata.update(doc.metadata) if len(hc.page_content) > 1000: final.extend(char_splitter.split_documents([hc])) else: final.append(hc) final = [c for c in final if len(c.page_content.strip()) > 20] return final

这套流程我在多个项目里跑过，处理几万个 Markdown 文件没问题。核心思路就是：先归一化到 Markdown，再按结构切分，最后做长度兜底和过滤。每一步都有明确的理由，不是拍脑袋定的。

后续如果要做图片、表格的深度解析，或者接入 PDF、Word 的转换，可以在这个基础上扩展。但无论怎么扩展，Markdown 作为中间层的地位不会变，因为它是在"结构丰富"和"处理简单"之间找到的最佳平衡点。