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

1.1 为什么数据导入是 RAG 系统的第一道生死关

做 RAG 的人都有一个共识：检索效果差，八成问题出在数据导入和解析环节，而不是模型本身。我见过太多团队花大价钱调 embedding 模型、换向量库、优化 prompt，结果回头一看，原始文档解析出来全是乱码、表格错位、段落粘连，检索出来的 chunk 根本没法用。

RAG 的全称是 Retrieval-Augmented Generation，检索增强生成。它的核心逻辑是：用户提问 → 检索相关文档片段 → 把片段作为上下文喂给大模型 → 生成回答。这个链条里，检索质量直接决定了生成质量，而检索质量的上限，在你把文档导入系统的那一刻就已经被锁死了。

打个比方：RAG 系统就像一座图书馆。向量数据库是书架，embedding 模型是图书管理员，大模型是帮你写研究报告的专家。但如果送进图书馆的书本身就是缺页的、装订错乱的、字迹模糊的，那管理员再厉害、专家再聪明，也写不出靠谱的报告。数据导入与解析，就是“把书整理好再上架”的过程。

这一篇主要聚焦在通用文本与结构化文档的解析，具体来说就是从最朴素的 txt 文件，到带层级结构的 Markdown 文件，怎么把它们干净、完整、有结构地导入到 RAG 流程里。选这个切入点是因为：txt 和 Markdown 是所有文档格式的“最大公约数”——PDF 解析完要转成文本，Word 解析完要保留结构，HTML 抓下来要清洗，最终都会落到纯文本或类 Markdown 的中间态。把这两种格式吃透，后面处理 PDF、Word、Excel 就是举一反三的事。

适合谁看？如果你正在用 LangChain 搭 RAG 知识库，或者准备从零做一个文档问答系统，又或者你已经跑通了 demo 但发现检索效果不理想，这篇内容都能给你直接的参考。不需要你精通 LangChain，但至少要跑过一两个 Chain，知道 Document 对象长什么样。

1.2 从 txt 到 Markdown：为什么解析策略要分层次

很多人会问：txt 和 Markdown 不都是纯文本吗？直接open().read()不就完了，为什么还要搞一套 Loader？

这个问题问到点子上了。纯文本不等于无结构文本。txt 文件确实没有显式结构标记，但它的内容里隐含了结构——段落之间的空行、章节标题的独立成行、列表项的缩进、代码块的边界。这些隐含结构如果不在导入阶段提取出来，后面 chunk 切分时就会把不相关的段落切到一起，检索时自然召回不准。

Markdown 则更进一步，它用#、##、-、**这些符号显式标注了层级结构。这些标记本身就是天然的语义边界。一个##标题下面的内容，大概率属于同一个主题；一个代码块里的内容，不应该和外面的说明文字混在一个 chunk 里。

所以解析策略要分层次：

• 第一层：字符编码与格式清洗。处理 BOM 头、换行符统一（\r\n→\n）、特殊字符转义。这一步不做，后面全是坑。
• 第二层：结构识别。从文本中识别出标题、段落、列表、代码块、表格等元素，并保留它们的层级关系。
• 第三层：元数据注入。给每个解析出来的 Document 对象打上来源、标题路径、时间等标签，方便后续过滤和溯源。
• 第四层：分块策略适配。根据文档结构选择合适的分块方式，而不是无脑按固定字符数切。

LangChain 的 Document Loader 体系就是为这四层服务的。它把不同格式的文档统一转换成Document对象（包含page_content和metadata两个核心字段），让后续的 splitter、embedding、vectorstore 都能用同一套接口处理。统一接口是 LangChain 最大的价值之一，你不需要为每种文件格式写一套下游逻辑。

1.3 方案选型：自研解析 vs LangChain Loader

在实际项目里，我一般会分两种情况选型：

情况一：格式单一、结构简单。比如你只有几百个 txt 文件，内容就是纯段落，没有表格和代码。这种情况直接自己写 Python 脚本读取、清洗、切分，反而比引入 LangChain 更轻量。代码量可能就几十行，可控性还更强。

情况二：格式多样、结构复杂。比如你要同时处理 txt、Markdown、PDF、Word、HTML，还要保留层级结构、表格、代码块。这种情况强烈建议用 LangChain 的 Loader 体系，因为：

• 它已经内置了几十种 Loader，覆盖主流格式
• 所有 Loader 输出统一的Document对象，下游不用改
• 社区活跃，遇到问题容易找到解决方案
• 可以自定义 Loader，继承BaseLoader实现自己的解析逻辑

我自己的项目里，通常是混合策略：用 LangChain 的TextLoader和UnstructuredMarkdownLoader做基础解析，然后在它们之上包一层自定义的清洗和元数据注入逻辑。这样既享受了框架的便利，又保留了业务定制的空间。

提示：不要迷信“一个 Loader 走天下”。LangChain 的 Loader 质量参差不齐，有些 Loader 对中文支持很差，有些会把表格解析成乱码。选 Loader 之前，一定拿真实数据跑一遍，看解析结果再决定。

2. 核心细节解析与实操要点

2.1 txt 文件解析：看似简单，坑最多

txt 文件是 RAG 数据导入里最容易被低估的格式。很多人觉得TextLoader一行代码就搞定了，但实际跑起来问题一堆。我整理了几个最常见的坑和对应的处理方式。

坑一：编码问题。中文 txt 文件可能是 UTF-8、GBK、GB2312、GB18030 等编码。如果编码识别错了，读出来就是乱码。LangChain 的TextLoader默认用 UTF-8，遇到 GBK 文件直接报错或输出乱码。

处理方式：用chardet库自动检测编码，或者手动指定encoding参数。我一般会写一个编码探测函数，先读前 10000 字节检测编码，再用检测到的编码读取全文。

import chardet def detect_encoding(file_path): with open(file_path, 'rb') as f: raw = f.read(10000) result = chardet.detect(raw) return result['encoding'] encoding = detect_encoding('data.txt') with open('data.txt', 'r', encoding=encoding) as f: content = f.read()

坑二：换行符不统一。Windows 的\r\n、Linux 的\n、老 Mac 的\r，混在一起会让后续的段落切分逻辑失效。统一替换成\n是最稳妥的做法。

坑三：BOM 头。有些 Windows 记事本保存的 UTF-8 文件会带 BOM（\ufeff），这个字符会出现在文本开头，影响后续处理。读取时用encoding='utf-8-sig'可以自动去掉 BOM。

坑四：段落边界模糊。txt 里段落之间可能用空行分隔，也可能用缩进，还可能连续多行没有空行。如果直接按\n\n切分，遇到没有空行的文档就切不开。我的经验是：先按空行切，如果切出来的段落平均长度超过 1000 字，再按单换行切，同时用标点符号做二次判断。

坑五：超长单行。有些 txt 是从网页复制下来的，整篇内容挤在一行里，没有任何换行。这种情况需要先做“软换行”处理，在句号、问号、感叹号后面插入换行符。

import re def normalize_text(text): text = text.replace('\r\n', '\n').replace('\r', '\n') text = text.replace('\ufeff', '') text = re.sub(r'([。！？；])', r'\1\n', text) text = re.sub(r'\n{3,}', '\n\n', text) return text.strip()

2.2 Markdown 解析：结构就是信息

Markdown 的解析核心思路和 txt 完全不同。txt 是“从无结构中推断结构”，Markdown 是“从显式标记中提取结构”。这个区别决定了 Markdown 的解析质量上限更高，但也更依赖解析器对 Markdown 语法的支持程度。

LangChain 里处理 Markdown 主要有两个 Loader：

• UnstructuredMarkdownLoader：基于unstructured库，能识别标题、列表、代码块、表格等元素，输出带category元数据的 Document 列表。
• MarkdownHeaderTextSplitter：专门按 Markdown 标题层级切分，把标题路径写入 metadata，非常适合做结构化 chunk。

我一般会组合使用：先用UnstructuredMarkdownLoader把 Markdown 解析成元素列表，再用MarkdownHeaderTextSplitter按标题切分，最后合并两者的元数据。

MarkdownHeaderTextSplitter的用法很直观：

from langchain_text_splitters import MarkdownHeaderTextSplitter headers_to_split_on = [ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on) chunks = splitter.split_text(markdown_content) for chunk in chunks: print(chunk.metadata) print(chunk.page_content[:100])

这样切出来的每个 chunk，metadata 里都带着它所属的标题路径，比如{"Header 1": "第一章", "Header 2": "1.1 节"}。检索时如果用户问的是“1.1 节讲了什么”，你可以直接用 metadata 过滤，精准度比纯向量检索高得多。

Markdown 解析的几个关键细节：

• 代码块要单独处理。代码块里的内容不应该和说明文字混在一个 chunk 里，因为它们的语义完全不同。我一般会把代码块单独抽出来，作为一个独立的 Document，metadata 里标记type: code。
• 表格要保留结构。Markdown 表格转成纯文本后，行列关系会丢失。如果表格内容重要，建议转成 HTML 或 JSON 格式保留结构，或者用专门的表格解析逻辑。
• 图片和链接要提取。![alt](url)和[text](url)里的 URL 和 alt 文本是有价值的信息，可以提取到 metadata 里，方便后续做多模态检索或引用溯源。
• 数学公式要保护。$...$和$$...$$里的内容如果被当成普通文本切分，公式会被切碎。建议在切分前用占位符替换公式，切分后再还原。

2.3 Document 对象的元数据设计

不管用哪种 Loader，最终输出的都是Document对象。page_content是文本内容，metadata是元数据字典。元数据设计得好不好，直接决定了后续检索的灵活度。

我一般会注入这几类元数据：

这些元数据在检索阶段非常有用。比如你可以做混合检索：先用向量检索召回一批 chunk，再用title_path做二次过滤，把不属于目标章节的 chunk 剔除。或者做时间过滤：只检索最近一年的文档。

注意：元数据不要塞太多。每个 chunk 的 metadata 都会存进向量库，字段太多会显著增加存储和检索开销。我一般控制在 5-8 个字段，只保留真正会用到的。

2.4 分块策略：结构感知切分优于固定长度切分

分块（chunking）是 RAG 里最容易被忽视但影响最大的环节。很多人直接用RecursiveCharacterTextSplitter，设个chunk_size=1000、chunk_overlap=200就完事了。这种做法在通用场景下能用，但在结构化文档上就是浪费信息。

我的建议是结构感知切分：

1. 优先按文档结构切。Markdown 按标题层级切，txt 按段落切。这样每个 chunk 的语义是完整的。
2. 结构切完后如果 chunk 太大，再按长度二次切。比如一个##标题下的内容有 3000 字，那就按 800-1000 字再切，但保留标题路径作为 metadata。
3. 结构切完后如果 chunk 太小，做合并。比如连续几个###小节都只有一两句话，可以合并成一个 chunk，避免碎片化。

RecursiveCharacterTextSplitter支持自定义分隔符优先级，可以模拟结构感知切分：

from langchain_text_splitters import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=100, separators=["\n## ", "\n### ", "\n\n", "\n", "。", " ", ""], length_function=len, )

这个分隔符列表的意思是：优先按##标题切，其次按###标题切，再按空行切，再按单换行切，再按句号切，最后才按字符切。这样切出来的 chunk 会尽量保持结构完整。

chunk_size 怎么定？没有标准答案，但有几个参考原则：

• 中文场景下，800-1200 字符是比较稳妥的范围。太小了语义不完整，太大了检索精度下降。
• 如果 embedding 模型支持长文本（比如 8192 token），可以适当放大到 1500-2000 字符。
• 代码类内容建议单独设置更小的 chunk_size（300-500），因为代码的语义密度高，切太大反而不好检索。

chunk_overlap 怎么定？一般是 chunk_size 的 10%-20%。overlap 的作用是防止关键信息刚好被切在边界上。但 overlap 太大会导致重复内容多，检索时召回一堆相似 chunk。我一般用 100-150 字符。

3. 实操过程与核心环节实现

3.1 环境准备与依赖安装

先把环境搭起来。我用的 Python 版本是 3.10，LangChain 相关包建议用最新稳定版。

pip install langchain langchain-community langchain-text-splitters pip install unstructured markdown chardet

如果你要处理 Markdown 里的表格和复杂元素，还需要装unstructured[md]：

pip install "unstructured[md]"

unstructured这个库依赖比较多，安装时可能会遇到系统级依赖问题。在 Ubuntu 上可能需要先装libmagic-dev、poppler-utils等。如果只是处理 txt 和 Markdown，其实不装unstructured也能跑，用TextLoader加自定义解析就够了。

3.2 通用文本加载器实现

先写一个通用的文本加载函数，处理编码检测、换行统一、BOM 清理：

import chardet from langchain_core.documents import Document def load_text_file(file_path, encoding=None): if encoding is None: with open(file_path, 'rb') as f: raw = f.read(10000) encoding = chardet.detect(raw)['encoding'] or 'utf-8' with open(file_path, 'r', encoding=encoding, errors='replace') as f: content = f.read() content = content.replace('\r\n', '\n').replace('\r', '\n') content = content.replace('\ufeff', '') metadata = { 'source': file_path, 'file_type': 'text', 'encoding': encoding, } return Document(page_content=content, metadata=metadata)

这个函数比 LangChain 自带的TextLoader多了编码自动检测和换行统一，实测下来对中文文档的兼容性好很多。

3.3 Markdown 结构化解析实现

Markdown 的解析分两步：先按标题切分，再对每个 chunk 做内容清洗和元数据增强。

from langchain_text_splitters import MarkdownHeaderTextSplitter def load_markdown_file(file_path): with open(file_path, 'r', encoding='utf-8') as f: content = f.read() headers_to_split_on = [ ("#", "h1"), ("##", "h2"), ("###", "h3"), ("####", "h4"), ] splitter = MarkdownHeaderTextSplitter( headers_to_split_on=headers_to_split_on, strip_headers=False, ) chunks = splitter.split_text(content) documents = [] for i, chunk in enumerate(chunks): title_path = ' > '.join([ chunk.metadata.get(h, '') for h in ['h1', 'h2', 'h3', 'h4'] if chunk.metadata.get(h) ]) metadata = { 'source': file_path, 'file_type': 'markdown', 'title_path': title_path, 'chunk_index': i, 'char_count': len(chunk.page_content), } documents.append(Document( page_content=chunk.page_content, metadata=metadata, )) return documents

strip_headers=False这个参数很关键。如果设为True，标题行会从page_content里去掉，只保留在 metadata 里。但实际检索时，标题本身也是重要的语义信息，去掉后反而影响召回。所以我一般保留标题在正文里。

3.4 代码块与表格的单独处理

Markdown 里的代码块和表格，如果混在普通文本里切分，效果很差。我的做法是在切分前先把它们抽出来，单独处理。

import re def extract_code_blocks(markdown_text): pattern = r'```(\w*)\n(.*?)```' blocks = [] for match in re.finditer(pattern, markdown_text, re.DOTALL): lang = match.group(1) or 'unknown' code = match.group(2) blocks.append({ 'language': lang, 'code': code, 'start': match.start(), 'end': match.end(), }) return blocks

抽出来的代码块，每个作为一个独立的 Document，metadata 里标记type: code、language: python。这样检索时如果用户问的是代码相关问题，可以优先召回代码块。

表格的处理类似，用正则匹配|...|格式的表格，转成 JSON 或 HTML 保留结构。如果表格内容不多，也可以直接转成自然语言描述，比如“下表列出了 A、B、C 三列，第一行是表头……”。

3.5 完整流程串联

把上面的步骤串起来，形成一个完整的导入流程：

def load_documents(file_paths): all_docs = [] for path in file_paths: if path.endswith('.md'): docs = load_markdown_file(path) elif path.endswith('.txt'): doc = load_text_file(path) docs = [doc] else: continue all_docs.extend(docs) return all_docs def split_documents(documents, chunk_size=800, chunk_overlap=100): splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n## ", "\n### ", "\n\n", "\n", "。", " ", ""], ) return splitter.split_documents(documents)

这个流程跑下来，txt 和 Markdown 都能被解析成带元数据的 Document 列表，再经过结构感知切分，就可以送进 embedding 和向量库了。

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

4.1 解析结果乱码或内容缺失

这是最常见的问题，排查思路按优先级来：

我踩过最坑的一次是：一批从旧系统导出的 txt 文件，编码是 GB18030，但文件头没有 BOM，chardet检测出来是 GBK。GBK 是 GB18030 的子集，大部分字符能正常解码，但少数生僻字会变成乱码。后来我强制用 GB18030 读取才解决。所以编码检测不能全信，关键数据要人工抽查。

4.2 Markdown 标题层级识别错误

MarkdownHeaderTextSplitter对标题的识别依赖正则匹配，如果 Markdown 文件里标题格式不规范，就会识别失败。常见的不规范情况：

• #标题（井号后没有空格）——标准 Markdown 要求井号后必须有空格
• # 标题 #（标题末尾有闭合井号）——有些解析器不认
• 标题行前面有空格——会被当成普通文本
• 用了 Setext 风格标题（===和---下划线）——MarkdownHeaderTextSplitter不支持

处理方式：在解析前先做一轮 Markdown 规范化，把不规范的标题修正成标准格式。

def normalize_markdown_headings(text): text = re.sub(r'^(#{1,6})([^#\s])', r'\1 \2', text, flags=re.MULTILINE) text = re.sub(r'^(#{1,6}\s+.+?)\s*#+\s*$', r'\1', text, flags=re.MULTILINE) return text

4.3 chunk 切分后语义不完整

这个问题通常有两个原因：一是chunk_size设得太小，二是分隔符优先级没配好。

排查方法：随机抽 10 个 chunk，人工读一遍，看是否有“半句话被切断”的情况。如果有，先调大chunk_size，再检查分隔符列表里有没有把句号、问号加进去。

我的一般配置是：chunk_size=800、chunk_overlap=100、分隔符包含\n\n、\n、。、！、？、；。这样切出来的 chunk 基本都能保持句子完整。

提示：如果你的文档里有大量列表项，建议把\n-和\n1.也加入分隔符，避免列表项被从中间切断。

4.4 元数据丢失或覆盖

用split_documents切分时，原始 Document 的 metadata 会自动继承到子 chunk 上。但如果你在切分后又手动创建了新的 Document，metadata 就可能丢失。每次创建 Document 时，都要显式传入 metadata，不要依赖默认值。

另一个坑是 metadata 键名冲突。比如两个 Loader 都往 metadata 里写source字段，但含义不同，合并时就会覆盖。建议在 metadata 键名上加前缀，比如md_source、txt_source，避免冲突。

4.5 性能问题：大文件解析慢

如果单个文件超过 10MB，或者文件数量超过几千个，解析速度会成为瓶颈。优化方向：

• 并行解析：用concurrent.futures.ThreadPoolExecutor并行处理多个文件。IO 密集型任务用多线程就够了。
• 增量导入：记录已导入文件的 hash 值，只处理新增或修改过的文件。
• 懒加载：不要一次性把所有 Document 加载到内存，用生成器逐批产出。

from concurrent.futures import ThreadPoolExecutor def load_documents_parallel(file_paths, max_workers=4): with ThreadPoolExecutor(max_workers=max_workers) as executor: results = executor.map(load_single_file, file_paths) return [doc for docs in results for doc in docs]

实测下来，4 线程并行比单线程快 3 倍左右。但注意不要开太多线程，否则磁盘 IO 会成为新瓶颈。

4.6 中文标点与英文标点混用

中文文档里经常混用中英文标点，比如逗号用,而不是，，句号用.而不是。。这会影响基于标点的切分逻辑。我的处理方式是在切分前做一轮标点规范化，把英文标点转成中文标点（在中文语境下），或者反过来。具体转哪边，取决于你的文档以哪种为主。

def normalize_punctuation(text): text = text.replace(',', '，').replace(';', '；') text = text.replace('!', '！').replace('?', '？') return text

但要注意：如果文档里有代码或英文内容，这个替换会破坏它们。所以标点规范化要分区域进行，只对中文段落做替换，代码块和英文段落跳过。

5. 从解析到入库的衔接要点

5.1 embedding 前的最后一道清洗

Document 切分完、准备送 embedding 之前，还有几件事要做：

• 去除空白 chunk：有些 chunk 只有几个空格或换行符，送进 embedding 纯属浪费。
• 去除重复 chunk：用 hash 去重，避免同一内容被多次索引。
• 长度过滤：太短的 chunk（少于 20 字符）直接丢弃，太长的 chunk（超过模型最大长度）要截断或二次切分。
• 特殊字符清理：连续多个空格、制表符、不可见字符，统一清理掉。

def clean_chunks(documents, min_length=20, max_length=2000): seen = set() cleaned = [] for doc in documents: content = doc.page_content.strip() content = re.sub(r'\s+', ' ', content) if len(content) < min_length or len(content) > max_length: continue content_hash = hash(content) if content_hash in seen: continue seen.add(content_hash) doc.page_content = content cleaned.append(doc) return cleaned

5.2 向量库写入的批量策略

送进向量库时，不要一条一条写，要批量写。LangChain 的VectorStore.add_documents支持批量传入，底层会做批处理。我一般每批 100-500 个 chunk，具体看 embedding 模型的吞吐和 API 限流。

如果是本地 embedding 模型（比如 BGE、M3E），批量大小可以设大一点，500-1000 都行。如果是调用远程 API，要注意限流，批次之间加个time.sleep(0.5)。

5.3 导入后的验证清单

数据导入完成后，一定要做验证。我一般会检查这几项：

这个验证清单看着简单，但能拦住 80% 的低级问题。我见过太多人导入完直接上线，结果用户一问就露馅。

6. 一些实操心得与后续扩展

6.1 我踩过的三个印象最深的坑

第一个坑：以为 txt 不需要解析。早期项目里我直接用open().read()读 txt，结果遇到 GBK 编码的文件全是乱码，排查了半天才发现是编码问题。从那以后，任何文本读取我都先做编码检测。

第二个坑：Markdown 标题层级太深。有个文档用了#####五级标题，MarkdownHeaderTextSplitter配置里只写到####，结果五级标题被当成普通文本，切分全乱了。后来我把标题配置扩展到六级，并且加了规范化预处理。

第三个坑：chunk_overlap 设太大。有次为了“保险”，把 overlap 设成 300，chunk_size 是 800。结果检索时同一个内容被召回三四次，去重后有效信息反而少了。后来老老实实按 10%-15% 设置。

6.2 后续可以扩展的方向

这篇主要覆盖了 txt 和 Markdown 的解析。实际项目里还会遇到 PDF、Word、Excel、HTML、PPT 等格式。后续可以沿着这几个方向扩展：

• PDF 解析：重点解决表格提取、扫描件 OCR、多栏排版的问题。推荐试试PyMuPDF和pdfplumber。
• Word 解析：保留标题样式、表格、批注。python-docx是基础，unstructured对 Word 的支持更全面。
• HTML 解析：重点是正文提取和噪声去除。trafilatura和readability-lxml是两个好用的库。
• 多模态文档：图片、图表、公式的提取和描述。这块目前还在快速发展，可以关注一些专门的多模态解析工具。

6.3 一个实用的小技巧

如果你不确定自己的分块策略好不好，可以用一个简单的方法验证：把切分后的 chunk 随机打乱，然后人工尝试把它们重新拼回原文。如果能拼回去，说明切分没有丢信息；如果拼不回去，说明有些内容在切分过程中丢失了。这个方法虽然笨，但非常有效，我每次调整分块参数后都会做一遍。

另外，建议在项目初期就建立一套解析质量评估集：选 20-30 个有代表性的文档，人工标注出关键信息点，每次调整解析逻辑后，检查这些信息点是否还能被正确提取。这套评估集能帮你快速判断改动是优化还是劣化，避免凭感觉调参。

数据导入与解析这件事，说到底就是“把原始文档变成检索友好的 chunk”的过程。它没有太多高深的技术，但细节极多，每一个细节都可能成为 RAG 效果的瓶颈。把 txt 和 Markdown 这两个基础格式吃透，后面的 PDF、Word 就是水到渠成的事。