文档处理与分块策略完全指南

引言

在RAG（检索增强生成）系统中，文档处理是第一步也是最关键的一步。分块（Chunking）策略直接决定了后续检索的质量和生成回答的准确性。一个优秀的分块策略能够：

• 保持语义完整性，避免信息断裂
• 提高检索精度，减少噪声干扰
• 优化上下文利用，提升生成质量

本文将深入探讨各种文档处理技术和分块策略，帮助你构建高质量的RAG系统。

为什么分块如此重要？

分块不当的常见问题

分块的核心目标

1. 语义完整性 - 每个块包含完整的语义单元
2. 上下文适度 - 既不过大也不过小
3. 检索友好 - 便于匹配用户查询
4. 生成友好 - 为LLM提供有效上下文

文档处理完整流程

┌─────────────────────────────────────────────────────────────┐
│                    文档处理流程                              │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│   原始文档                                                    │
│      │                                                       │
│      ▼                                                       │
│   ┌─────────────┐                                            │
│   │  格式解析   │  ← PDF/Word/HTML/Markdown等                │
│   │  Parsing    │                                            │
│   └──────┬──────┘                                            │
│          │                                                   │
│          ▼                                                   │
│   ┌─────────────┐                                            │
│   │  内容清洗   │  ← 去除噪声、标准化                         │
│   │  Cleaning   │                                            │
│   └──────┬──────┘                                            │
│          │                                                   │
│          ▼                                                   │
│   ┌─────────────┐                                            │
│   │  文本分块   │  ← 核心步骤，多种策略                       │
│   │  Chunking   │                                            │
│   └──────┬──────┘                                            │
│          │                                                   │
│          ▼                                                   │
│   ┌─────────────┐                                            │
│   │  元数据提取 │  ← 标题、来源、时间等                       │
│   │  Metadata   │                                            │
│   └──────┬──────┘                                            │
│          │                                                   │
│          ▼                                                   │
│   分块结果 + 元数据                                            │
│                                                              │
└─────────────────────────────────────────────────────────────┘

文档解析技术

1. PDF文档处理

PDF是最常见的文档格式，但解析难度较大：

# 方案一：PyPDFLoader（简单文本）
from langchain.document_loaders import PyPDFLoader

loader = PyPDFLoader("document.pdf")
pages = loader.load()

# 方案二：Unstructured（复杂布局）
from langchain.document_loaders import UnstructuredPDFLoader

loader = UnstructuredPDFLoader(
    "document.pdf",
    mode="elements",  # 保留结构信息
    strategy="hi_res"  # 高精度OCR
)
docs = loader.load()

# 方案三：PDFPlumber（表格友好）
import pdfplumber

with pdfplumber.open("document.pdf") as pdf:
    for page in pdf.pages:
        text = page.extract_text()
        tables = page.extract_tables()

PDF解析方案对比：

2. 网页内容抓取

from langchain.document_loaders import WebBaseLoader
from bs4 import BeautifulSoup
import requests

# 基础网页加载
loader = WebBaseLoader(
    "https://example.com/article",
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("article-content", "post-content")
        )
    )
)
docs = loader.load()

# 高级：处理动态内容
from selenium import webdriver

def load_dynamic_page(url):
    driver = webdriver.Chrome()
    driver.get(url)
    # 等待JS渲染
    driver.implicitly_wait(10)
    html = driver.page_source
    driver.quit()
    return html

3. 结构化文档处理

# Markdown处理
from langchain.document_loaders import UnstructuredMarkdownLoader

loader = UnstructuredMarkdownLoader(
    "doc.md",
    mode="elements"  # 保留标题层级
)

# Word文档
from langchain.document_loaders import Docx2txtLoader

loader = Docx2txtLoader("document.docx")

分块策略详解

1. 固定字符分块（Fixed Character Chunking）

最基础的分块方式，按固定字符数分割：

from langchain.text_splitter import CharacterTextSplitter

text_splitter = CharacterTextSplitter(
    separator="\n\n",      # 分隔符
    chunk_size=1000,       # 每个块大小
    chunk_overlap=200,     # 重叠字符数
    length_function=len,   # 长度计算函数
    is_separator_regex=False
)

chunks = text_splitter.split_documents(documents)

适用场景：

• 简单文本
• 对语义完整性要求不高
• 快速原型验证

优缺点：

• ✅ 实现简单，速度快
• ❌ 可能切断句子，破坏语义

2. 递归字符分块（Recursive Character Chunking）

推荐的标准方案，按优先级尝试多种分隔符：

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=[
        "\n\n",    # 段落
        "\n",      # 换行
        "。",      # 句号（中文）
        ".",       # 句号（英文）
        "！",      # 感叹号
        "？",      # 问号
        " ",       # 空格
        ""         # 字符
    ],
    length_function=len,
)

chunks = text_splitter.split_documents(documents)

工作原理：

1. 优先按段落分割（\n\n）
2. 如果块仍过大，按换行分割（\n）
3. 继续尝试句子边界（。 .）
4. 最后按字符分割

适用场景：

• 通用文本处理
• 需要保持基本语义边界
• 大多数RAG系统的首选方案

3. 语义分块（Semantic Chunking）

基于语义相似度进行分块，保持主题一致性：

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

text_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",  # 或 "standard_deviation"
    breakpoint_threshold_amount=95
)

chunks = text_splitter.split_documents(documents)

工作原理：

1. 计算相邻句子的语义相似度
2. 当相似度低于阈值时，创建新的块
3. 确保每个块内语义连贯

适用场景：

• 长文档处理
• 主题变化明显的文本
• 对语义完整性要求高

优缺点：

• ✅ 语义边界清晰
• ❌ 计算成本高，需要调用嵌入模型

4. 文档结构分块（Document Structure Chunking）

基于文档的自然结构（标题、章节）进行分块：

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Markdown按标题分块
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)

# HTML按标签分块
from langchain.text_splitter import HTMLHeaderTextSplitter

html_splitter = HTMLHeaderTextSplitter(
    headers_to_split_on=[
        ("h1", "Header 1"),
        ("h2", "Header 2"),
        ("h3", "Header 3"),
    ]
)

适用场景：

• 结构化文档（Markdown、HTML）
• 技术文档、API文档
• 需要保留层级关系的场景

5. 智能分块（Agentic Chunking）

使用LLM智能判断分块边界：

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

chunk_prompt = PromptTemplate(
    input_variables=["text"],
    template="""
    分析以下文本，将其分割成语义完整的段落。
    每个段落应包含一个完整的主题或观点。
    
    文本：
    {text}
    
    请输出分割后的段落，用---分隔：
    """
)

llm_chain = LLMChain(llm=llm, prompt=chunk_prompt)

适用场景：

• 高度复杂的文档
• 对质量要求极高的场景
• 预算充足的商业项目

优缺点：

• ✅ 分块质量最高
• ❌ 成本极高，速度慢

分块策略选择指南

┌─────────────────────────────────────────────────────────────┐
│                    分块策略选择决策树                          │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│   开始                                                        │
│    │                                                         │
│    ▼                                                         │
│   文档是否有明确结构？ ──是──► 文档结构分块                     │
│    │ 否                                                       │
│    ▼                                                         │
│   预算是否充足？ ───是───► 语义分块 / Agentic分块               │
│    │ 否                                                       │
│    ▼                                                         │
│   需要快速原型？ ───是───► 固定字符分块                         │
│    │ 否                                                       │
│    ▼                                                         │
│   推荐：递归字符分块（RecursiveCharacterTextSplitter）          │
│                                                              │
└─────────────────────────────────────────────────────────────┘

分块参数调优

Chunk Size 选择

Chunk Overlap 选择

# 重叠率建议：10%-20%
chunk_overlap = int(chunk_size * 0.1)  # 10%重叠

# 特殊场景
# 代码分块：需要更大重叠保持逻辑连贯
chunk_overlap = int(chunk_size * 0.2)

# 问答场景：较小重叠减少冗余
chunk_overlap = int(chunk_size * 0.05)

重叠的作用

1. 保持上下文连贯 - 避免句子断裂
2. 提高检索召回 - 相关信息不会被分割到两个块
3. 增加冗余 - 重要信息可能在多个块中出现

高级技巧

1. 父子分块（Parent-Document Retrieval）

小块用于检索，大块用于生成：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 子分块（用于检索）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 父分块（用于生成）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

# 存储父文档
store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

retriever.add_documents(documents)

优势：

• 检索精度高（小块匹配）
• 生成质量好（大上下文）

2. 元数据增强

为每个块添加丰富的元数据：

def enrich_metadata(chunks, source_doc):
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "source": source_doc["path"],
            "chunk_index": i,
            "total_chunks": len(chunks),
            "title": source_doc.get("title", ""),
            "section": extract_section(chunk),
            "created_at": source_doc.get("date", ""),
            "word_count": len(chunk.page_content.split()),
        })
    return chunks

3. 多粒度分块

同时维护多种粒度的索引：

# 句子级索引（精确检索）
sentence_chunks = split_by_sentence(docs)

# 段落级索引（平衡）
paragraph_chunks = split_by_paragraph(docs)

# 章节级索引（粗粒度）
section_chunks = split_by_section(docs)

# 检索时融合
results = fuse_results([
    search(sentence_store, query, k=3),
    search(paragraph_store, query, k=3),
    search(section_store, query, k=2),
])

4. 自定义分割器

针对特定领域实现自定义分割：

class CodeSplitter:
    """代码专用分割器"""
    
    def __init__(self, language="python"):
        self.language = language
        self.separators = {
            "python": ["\nclass ", "\ndef ", "\n\n", "\n"],
            "javascript": ["\nfunction ", "\nconst ", "\nclass ", "\n\n"],
        }
    
    def split(self, code):
        # 按函数/类分割
        # 保留函数签名作为上下文
        pass

class LegalDocumentSplitter:
    """法律文档专用分割器"""
    
    def split(self, text):
        # 按条款分割
        # 保留条款编号
        # 处理引用关系
        pass

分块质量评估

评估指标

def evaluate_chunking(chunks, query_results):
    """评估分块质量"""
    
    metrics = {
        # 语义连贯性
        "semantic_coherence": calculate_coherence(chunks),
        
        # 检索准确率
        "retrieval_precision": len([r for r in query_results if r.relevant]) / len(query_results),
        
        # 上下文完整性
        "context_completeness": check_completeness(chunks),
        
        # 信息密度
        "information_density": [len(c.page_content.split()) for c in chunks],
    }
    
    return metrics

人工评估清单

• 每个块是否有明确的主题？
• 句子是否被截断？
• 关键信息是否被分割到多个块？
• 块大小是否均匀？
• 元数据是否完整？

实战案例

案例一：技术文档处理

# 处理API文档（Markdown格式）
from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("##", "Endpoint"),      # 按API端点分块
    ("###", "Parameter"),    # 参数说明
]

splitter = MarkdownHeaderTextSplitter(headers_to_split_on)
chunks = splitter.split_text(api_doc_content)

# 添加API特定元数据
for chunk in chunks:
    chunk.metadata["api_version"] = "v2"
    chunk.metadata["endpoint"] = extract_endpoint(chunk)

案例二：法律合同处理

# 处理法律合同
class LegalContractSplitter:
    def split(self, contract_text):
        # 按条款分割（第X条）
        import re
        pattern = r'(第[一二三四五六七八九十\d]+条[、.])'
        
        sections = re.split(pattern, contract_text)
        chunks = []
        
        for i in range(1, len(sections), 2):
            clause_num = sections[i]
            clause_content = sections[i+1] if i+1 < len(sections) else ""
            
            chunk = Document(
                page_content=f"{clause_num}{clause_content}",
                metadata={"clause_number": clause_num}
            )
            chunks.append(chunk)
        
        return chunks

案例三：多语言混合文档

# 处理中英文混合文档
def multilingual_splitter(text, chunk_size=500):
    """
    针对中英文混合文本的优化分割
    """
    # 中英文标点都作为分隔符
    separators = [
        "\n\n",      # 段落
        "\n",        # 换行
        "。", ".",    # 句号
        "！", "!",   # 感叹号
        "？", "?",   # 问号
        "；", ";",   # 分号
        " ",         # 空格
    ]
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=50,
        separators=separators,
        length_function=lambda x: len(x.encode('utf-8'))  # 按字节长度
    )
    
    return splitter.split_text(text)

常见问题与解决方案

Q1: 如何处理表格数据？

# 方案一：转换为文本描述
import pandas as pd

def table_to_text(table_data):
    df = pd.DataFrame(table_data)
    return df.to_markdown()

# 方案二：单独存储
for table in extracted_tables:
    table_doc = Document(
        page_content=table_to_text(table),
        metadata={"type": "table", "page": table.page}
    )

Q2: 如何处理图片中的文字？

from PIL import Image
import pytesseract

def extract_image_text(image_path):
    image = Image.open(image_path)
    text = pytesseract.image_to_string(image, lang='chi_sim+eng')
    return text

Q3: 长文档内存不足？

# 流式处理
def process_large_document(file_path, chunk_size=1000):
    with open(file_path, 'r', encoding='utf-8') as f:
        buffer = ""
        for line in f:
            buffer += line
            if len(buffer) >= chunk_size:
                yield buffer
                buffer = buffer[-200:]  # 保留重叠
        if buffer:
            yield buffer

总结

文档处理与分块是RAG系统的基石，选择合适的策略需要综合考虑：

最佳实践清单：

• 根据文档类型选择合适的解析器
• 使用递归字符分块作为默认方案
• 合理设置chunk_size和overlap
• 为每个块添加完整的元数据
• 考虑父子分块提升检索质量
• 定期评估分块效果并优化

本文最后更新于 2024-02-15，如有问题欢迎在社区讨论。