1. RAG 数据导入与解析的整体设计思路

做 RAG 应用最容易被低估的环节，不是向量检索，也不是大模型选型，而是数据导入与解析。我见过太多项目在 demo 阶段跑得挺欢，一上真实文档就翻车——PDF 里的表格变成乱码、扫描件一个字都读不出来、Markdown 的层级结构全丢了。问题的根源往往不在模型，而在数据进入向量库之前的那几步处理。

这个系列的第一篇，我聚焦在最基础但也最通用的场景：纯文本 txt 和结构化 Markdown 的导入与解析。为什么从这两类开始？因为它们是所有文档格式的“最大公约数”。你从网页抓下来的内容、从 PDF 提取出来的文字、从数据库导出的字段，最终几乎都会落到 txt 或 Markdown 这两种形态上。把这两类吃透，后面处理 PDF、Word、HTML 就是在此基础上加解析器的事。

核心思路其实就一句话：把非结构化的文本，切成有语义边界、带元数据、可追溯的 Document 对象。LangChain 的 Document Loader 体系就是干这个的。但很多人用 Loader 只是loader.load()一把梭，结果切出来的 chunk 要么把一句话拦腰截断，要么把标题和正文混在一起，检索时召回的全是噪音。这篇我会把从文件读取、编码处理、结构解析、分块策略到元数据注入的完整链路拆开讲，每个环节都给出可复现的代码和参数选择的理由。

适合谁看？如果你正在搭 RAG 知识库，手头有一堆 txt 笔记或 Markdown 文档要入库，或者你用过 LangChain 但对其中的分块逻辑一知半解，这篇能帮你少走至少两周弯路。我默认你有 Python 基础，知道什么是向量库，但不需要你精通 LangChain——所有代码我都会解释清楚每一步在干什么。

2. 核心概念与工具选型解析

2.1 为什么是 Document 对象而不是纯字符串

LangChain 里所有 Loader 的产出都是Document对象，不是裸字符串。这个设计很多人一开始不理解，觉得多此一举。但等你做检索溯源的时候就明白了：Document有两个核心字段，page_content存文本内容，metadata存元数据。元数据里可以放来源文件路径、页码、标题层级、创建时间等等。

检索的时候，向量库返回的是相似的 chunk，但用户想知道“这段话出自哪个文件的哪一部分”，靠的就是 metadata。如果一开始图省事直接存字符串，后面想加溯源信息就得重新处理一遍全量数据，代价极大。所以我的习惯是：从导入的第一行代码开始，就把 metadata 设计好，哪怕暂时用不上。

2.2 Loader 选型的三个判断维度

LangChain 社区提供了大量 Loader，光文本类就有TextLoader、UnstructuredFileLoader、MarkdownLoader、DirectoryLoader等。选哪个不是看哪个高级，而是看三个维度：

我实测下来的经验是：能用专用 Loader 就别用通用的。UnstructuredFileLoader虽然什么都能读，但它内部要判断文件类型、调用不同的解析后端，速度和稳定性都不如专用 Loader。而且通用 Loader 对 Markdown 的结构保留往往不如专门的 Markdown 解析器。

2.3 分块策略：RAG 成败的关键一环

数据导入里最容易被忽视、但对检索质量影响最大的就是分块。分块太大，检索时召回的内容包含太多无关信息，浪费上下文窗口；分块太小，语义被切碎，检索出来的片段答非所问。

LangChain 提供了多种 TextSplitter，常用的有CharacterTextSplitter、RecursiveCharacterTextSplitter、MarkdownHeaderTextSplitter。我的选型逻辑是这样的：

• 纯文本 txt：用RecursiveCharacterTextSplitter，按段落、换行、句号、逗号逐级降级切分，尽量保持语义完整。
• 结构化 Markdown：先用MarkdownHeaderTextSplitter按标题层级切，再用RecursiveCharacterTextSplitter对过长的段落做二次切分。

这里有个关键参数chunk_size和chunk_overlap。chunk_size不是越大越好，也不是越小越好。我的经验值是：中文文本 chunk_size 设在 500-800 字符，overlap 设在 50-100 字符。为什么是这个范围？因为中文一个字符承载的信息量比英文单词大，500 字符大约对应 300-400 个汉字，正好是一个完整段落的长度。overlap 的作用是防止关键信息正好落在切分边界上被切断，50-100 字符能保证上下文的连续性。

注意：chunk_size 的单位是字符数不是 token 数。如果你用的是按 token 计费的 embedding 模型，需要自己换算。中文大致 1 字符约等于 0.6-1 个 token，具体取决于分词器。

3. 纯文本 txt 的导入与解析实操

3.1 编码问题是第一个坑

处理 txt 文件，十有八九会碰到编码问题。Windows 上创建的 txt 默认可能是 GBK 或 GB2312，Linux 和 Mac 上一般是 UTF-8。如果你直接用TextLoader不加encoding参数，遇到非 UTF-8 文件就会抛UnicodeDecodeError。

我的处理方案是写一个编码探测函数，用chardet库自动检测，检测不出来再按优先级尝试：

import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: raw = f.read(10000) # 读前 10000 字节做检测 result = chardet.detect(raw) encoding = result['encoding'] confidence = result['confidence'] # 置信度太低时按优先级回退 if confidence < 0.7: for enc in ['utf-8', 'gbk', 'gb2312', 'utf-16']: try: with open(file_path, 'r', encoding=enc) as f: f.read(1000) return enc except UnicodeDecodeError: continue return encoding or 'utf-8'

这个函数先读前 10000 字节做统计检测，因为chardet对短文本的检测准确率不高。如果置信度低于 0.7，就按 utf-8、gbk、gb2312、utf-16 的顺序逐个尝试，哪个能正常读就用哪个。实测下来，这套逻辑能覆盖 95% 以上的中文 txt 文件。

3.2 TextLoader 的正确用法

拿到编码后，用TextLoader加载就很简单了：

from langchain_community.document_loaders import TextLoader loader = TextLoader( file_path="data/notes.txt", encoding=detect_encoding("data/notes.txt"), autodetect_encoding=False # 我们已经手动检测了，关掉自动检测 ) documents = loader.load()

这里有个细节：autodetect_encoding参数。LangChain 的TextLoader支持自动检测编码，但它内部用的也是chardet，而且检测逻辑比较简单。我建议手动检测后传入encoding，把autodetect_encoding设为False，避免重复检测和潜在的误判。

加载出来的documents是一个列表，每个元素是一个Document对象。对于单个 txt 文件，列表通常只有一个元素，page_content是全文，metadata里默认只有source字段，值是文件路径。

3.3 元数据注入：让每个 chunk 都可追溯

默认的 metadata 只有 source，信息太少。我通常会在加载后手动补充元数据：

import os from datetime import datetime for doc in documents: doc.metadata.update({ "file_name": os.path.basename(doc.metadata["source"]), "file_type": "txt", "load_time": datetime.now().isoformat(), "char_count": len(doc.page_content), })

这些字段看起来不起眼，但在后续排查问题时非常有用。比如检索结果不理想，你可以先看char_count，如果某个 chunk 只有几十个字符，那大概率是分块出了问题。load_time则能帮你定位是哪一批数据导入的。

3.4 分块参数的计算与选择

txt 文件加载后是一整块文本，必须分块。我用RecursiveCharacterTextSplitter：

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, length_function=len, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""], ) chunks = text_splitter.split_documents(documents)

重点说separators这个参数。它的逻辑是：先尝试用\n\n（段落分隔）切，如果切出来的块还是超过chunk_size，就用\n（换行）切，再不行就用中文句号、感叹号、问号、分号、逗号，最后才用空格和空字符串逐字符切。这个降级顺序保证了优先在语义边界处切分。

为什么把中文标点放在空格前面？因为中文文本里空格很少，如果按默认的英文分隔符（空格优先），中文长句会被硬切。把中文标点提前，能保证句子完整性。实测下来，这套分隔符对中文 txt 的切分效果比默认配置好很多，基本不会出现把一句话切成两半的情况。

chunk_size=600和chunk_overlap=80是我在多个项目里调出来的经验值。600 字符大约对应 400 个汉字，是一个中等长度段落的体量。80 字符的 overlap 约等于一句话的长度，能保证跨块的语义连续性。当然这不是金标准，你可以根据自己文档的特点微调。如果文档段落普遍很短，chunk_size 可以降到 400；如果段落很长，可以升到 800。

4. Markdown 结构化解析的完整流程

4.1 为什么 Markdown 不能当纯文本处理

Markdown 看起来是纯文本，但它有隐含的结构：#是一级标题，##是二级标题，-是无序列表，```是代码块。如果你用处理 txt 的方式处理 Markdown，这些结构信息就全丢了。

结构信息对 RAG 有多重要？举个例子，用户问“XX 功能的参数怎么配置”，如果 chunk 里保留了## 配置参数这个标题，检索时模型能明确知道这段内容属于配置章节，回答的准确率会明显提升。反过来，如果标题和正文被切散，模型看到的就是一堆没有上下文的参数列表，很容易答错。

所以 Markdown 的处理必须分两步：先按标题层级切分并保留层级信息，再对过长的段落做二次切分。

4.2 MarkdownHeaderTextSplitter 的层级保留机制

LangChain 的MarkdownHeaderTextSplitter专门干这件事：

from langchain.text_splitter import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "h1"), ("##", "h2"), ("###", "h3"), ("####", "h4"), ] markdown_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False, # 保留标题在内容中 ) md_chunks = markdown_splitter.split_text(markdown_text)

headers_to_split_on定义了要识别的标题层级和对应的元数据键名。strip_headers=False表示切分后标题文本保留在page_content里，而不是只放在 metadata 中。我建议设为False，因为标题本身携带重要语义信息，保留在内容里能提升 embedding 的质量。

切分后，每个 chunk 的 metadata 里会带上它所属的标题层级。比如一个 chunk 在## 配置参数下面，它的 metadata 就是{"h1": "使用指南", "h2": "配置参数"}。这个层级链就是天然的上下文，检索时能直接告诉模型这段内容的归属。

4.3 二次切分：处理超长段落

MarkdownHeaderTextSplitter只按标题切，如果一个标题下面的内容特别长（比如一个章节有几千字），切出来的 chunk 还是太大。这时候需要二次切分：

from langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""], ) final_chunks = text_splitter.split_documents(md_chunks)

注意这里传入的是split_documents而不是split_text，因为md_chunks是Document对象列表，split_documents会保留原有的 metadata，把标题层级信息带到二次切分后的每个 chunk 上。这一点非常关键——如果用了split_text，metadata 就丢了。

4.4 代码块和表格的特殊处理

Markdown 里的代码块和表格是两类特殊内容，处理不当会严重影响检索质量。

代码块的问题在于：RecursiveCharacterTextSplitter的分隔符里没有针对代码块的保护机制，一个长代码块可能被从中间切断，导致语法不完整。我的处理方案是在分块前先把代码块提取出来单独处理：

import re def extract_code_blocks(text): """提取 Markdown 中的代码块，返回代码块列表和替换后的文本""" pattern = r'```(\w*)\n(.*?)```' code_blocks = [] def replacer(match): lang = match.group(1) code = match.group(2) placeholder = f"__CODE_BLOCK_{len(code_blocks)}__" code_blocks.append({"lang": lang, "code": code, "placeholder": placeholder}) return placeholder text_without_code = re.sub(pattern, replacer, text, flags=re.DOTALL) return code_blocks, text_without_code

提取出来后，代码块作为独立的 chunk 存入，metadata 里标记content_type: "code"和language: "python"。这样检索时如果用户问的是代码相关的问题，可以优先召回代码块类型的 chunk。

表格的处理类似。Markdown 表格用|分隔，如果被切断，表头和表体会分离，检索出来就是一堆没有列名的数字。我的做法是把整个表格作为一个不可分割的单元，如果表格超过 chunk_size，就单独存为一个 chunk，不做二次切分。

提示：代码块和表格的独立存储会增加 chunk 数量，但能显著提升特定类型问题的检索准确率。如果你的知识库以技术文档为主，这个处理非常值得做。

5. 批量导入与常见问题排查

5.1 DirectoryLoader 批量处理目录

实际项目里很少只处理一个文件，通常是一整个目录。DirectoryLoader可以配合 glob 模式批量加载：

from langchain_community.document_loaders import DirectoryLoader, TextLoader loader = DirectoryLoader( path="data/docs", glob="**/*.txt", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"}, show_progress=True, use_multithreading=True, max_concurrency=4, ) documents = loader.load()

glob="**/*.txt"表示递归匹配所有子目录下的 txt 文件。use_multithreading=True开启多线程加载，max_concurrency=4控制并发数。这里并发数不建议设太高，因为文件 IO 本身有瓶颈，设到 4-8 就够了，再高反而会因为线程切换降低效率。

对于混合格式的目录（既有 txt 又有 md），需要分别用不同的 Loader 加载，然后合并结果：

txt_loader = DirectoryLoader("data/docs", glob="**/*.txt", loader_cls=TextLoader) md_loader = DirectoryLoader("data/docs", glob="**/*.md", loader_cls=UnstructuredMarkdownLoader) all_docs = txt_loader.load() + md_loader.load()

5.2 常见问题速查表

我在实际项目中踩过的坑，整理成一张表，方便你对照排查：

5.3 几个容易忽视的实操细节

第一个细节是空文件和空白内容的过滤。目录里经常有一些空文件或者只有几个空格的文件，加载后会生成空的 Document，这些空 Document 进入向量库会污染检索结果。我的做法是在加载后统一过滤：

documents = [doc for doc in documents if len(doc.page_content.strip()) > 10]

阈值设 10 是因为太短的内容（比如只有一两个词）作为独立 chunk 没有检索价值，反而会增加噪音。

第二个细节是文件路径的规范化。不同操作系统下路径分隔符不同，Windows 是反斜杠，Linux 和 Mac 是正斜杠。metadata 里的 source 字段如果直接存原始路径，跨平台迁移时会出问题。我习惯统一转成正斜杠：

doc.metadata["source"] = doc.metadata["source"].replace("\\", "/")

第三个细节是大文件的分批处理。如果单个 txt 文件超过 10MB，一次性加载到内存再分块可能会占用大量内存。我的做法是先用TextLoader加载，然后立即分块，分块后的 chunk 列表比原始全文小得多，内存压力会缓解。如果文件实在太大（比如超过 100MB），就需要用流式读取，按行读取并累积到一定大小就切分，而不是一次性读入。

5.4 分块质量的验证方法

分块做完后怎么知道效果好不好？不能凭感觉。我通常用两个方法验证：

方法一：人工抽样检查。随机抽 10-20 个 chunk，看内容是否语义完整、是否有明显的截断、metadata 是否正确。这个方法虽然原始，但最直接。

方法二：检索测试。准备 10-20 个典型问题，用这些 chunk 建一个临时向量库，跑一遍检索，看召回的内容是否相关。如果某个问题的召回结果明显不相关，就去检查对应的 chunk 是不是切分出了问题。

我一般会写一个简单的检查脚本，统计 chunk 的长度分布：

import statistics lengths = [len(chunk.page_content) for chunk in chunks] print(f"chunk 总数: {len(chunks)}") print(f"平均长度: {statistics.mean(lengths):.0f}") print(f"中位数长度: {statistics.median(lengths):.0f}") print(f"最短: {min(lengths)}, 最长: {max(lengths)}") print(f"标准差: {statistics.stdev(lengths):.0f}")

如果标准差很大，说明 chunk 长度参差不齐，可能有异常短的或异常长的 chunk，需要进一步排查。理想情况下，大部分 chunk 的长度应该集中在 chunk_size 附近，标准差控制在 150 以内。

6. 从导入到入库的完整链路串联

6.1 一个可复用的处理管道

把前面所有环节串起来，形成一个完整的处理管道。这个管道我封装成了一个函数，输入是目录路径，输出是处理好的 chunk 列表：

from langchain_community.document_loaders import TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter import os def build_ingestion_pipeline(directory): all_chunks = [] # 第一步：分别加载 txt 和 md for root, dirs, files in os.walk(directory): for file in files: file_path = os.path.join(root, file) if file.endswith('.txt'): encoding = detect_encoding(file_path) loader = TextLoader(file_path, encoding=encoding) docs = loader.load() splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = splitter.split_documents(docs) elif file.endswith('.md'): loader = UnstructuredMarkdownLoader(file_path) docs = loader.load() md_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[("#", "h1"), ("##", "h2"), ("###", "h3")], strip_headers=False ) md_chunks = md_splitter.split_text(docs[0].page_content) splitter = RecursiveCharacterTextSplitter( chunk_size=600, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = splitter.split_documents(md_chunks) else: continue # 补充元数据 for chunk in chunks: chunk.metadata.update({ "file_name": file, "file_type": file.split('.')[-1], "source": file_path.replace("\\", "/"), }) all_chunks.extend(chunks) # 过滤过短的 chunk all_chunks = [c for c in all_chunks if len(c.page_content.strip()) > 10] return all_chunks

这个管道覆盖了从文件遍历、编码检测、格式区分、分块到元数据注入的全流程。你可以直接拿去用，也可以根据自己的需求调整参数。

6.2 入库前的最后检查

chunk 准备好之后，别急着往向量库里灌。先做几项检查：

第一，去重。同一份文档可能被重复导入，或者不同文档里有完全相同的段落。用内容哈希去重：

import hashlib seen = set() unique_chunks = [] for chunk in all_chunks: content_hash = hashlib.md5(chunk.page_content.encode()).hexdigest() if content_hash not in seen: seen.add(content_hash) unique_chunks.append(chunk)

第二，检查 metadata 完整性。确保每个 chunk 都有 source、file_name、file_type 这几个关键字段，缺失的补上默认值。

第三，预估 embedding 成本。统计总字符数，按你的 embedding 模型的计费方式估算成本。中文大致 1 字符对应 0.6-1 个 token，600 字符的 chunk 大约 400-600 token。如果总共有 10000 个 chunk，那就是 400-600 万 token，心里有个数。

6.3 后续扩展方向

这套管道目前只处理 txt 和 Markdown，但它的架构是可扩展的。要加 PDF 支持，只需要在文件类型判断里加一个分支，用PyPDFLoader加载，后面的分块和元数据逻辑可以复用。要加 HTML，用UnstructuredHTMLLoader或者BSHTMLLoader，同样复用后续流程。

真正需要单独处理的是扫描件 PDF 和图片，那涉及到 OCR，是另一个技术栈。还有表格密集的 Excel 和 CSV，需要专门的表格解析逻辑。这些我会在后续的文章里展开。

我在实际项目里最大的体会是：数据导入这一步，花多少时间都值得。很多人急着调模型、调检索参数，但源头的数据质量不行，后面怎么调都是白费。把 txt 和 Markdown 这两类基础格式处理干净，你的 RAG 知识库就已经赢在起跑线上了。