文档处理与分块策略

为什么文档处理至关重要

RAG 系统的质量上限由数据质量决定。一个优秀的 LLM 配上糟糕的文档处理，依然会给出混乱的答案。文档处理包括三个核心环节：加载（把各种格式的文件读成文本）、清洗（去除噪声、格式化）和分块（把长文档切成适合检索的小片段）。

文档加载（Document Loading）

现实世界的文档格式五花八门：PDF、Word、网页、Markdown、数据库记录……每种格式都需要专门的加载器。

用 LangChain 加载各类文档

from langchain_community.document_loaders import (
    PyPDFLoader,          # PDF 文件
    Docx2txtLoader,       # Word .docx
    WebBaseLoader,        # 网页
    TextLoader,           # 纯文本 / Markdown
    CSVLoader,            # CSV 表格
    UnstructuredExcelLoader, # Excel
)
from langchain_core.documents import Document

# ── PDF 加载（按页分割）──────────────────
pdf_loader = PyPDFLoader("docs/annual_report.pdf")
pdf_pages = pdf_loader.load()
print(f"PDF 共 {len(pdf_pages)} 页")
print(pdf_pages[0].metadata)
# {'source': 'docs/annual_report.pdf', 'page': 0}

# ── Word 文档加载 ───────────────────────
docx_loader = Docx2txtLoader("docs/contract.docx")
docx_docs = docx_loader.load()  # 返回单个 Document

# ── 网页加载（带正文提取）──────────────
web_loader = WebBaseLoader(
    web_paths=["https://docs.qdrant.tech/concepts/collections/"],
    bs_kwargs={"features": "html.parser"}
)
web_docs = web_loader.load()

# ── 批量加载目录所有 PDF ─────────────────
from langchain_community.document_loaders import DirectoryLoader

dir_loader = DirectoryLoader(
    "./knowledge_base/",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    show_progress=True,
)
all_docs = dir_loader.load()
print(f"加载了 {len(all_docs)} 个文档块")

文档清洗：不容忽视的预处理步骤

import re
from langchain_core.documents import Document

def clean_document(doc: Document) -> Document:
    """清洗文档内容，去除噪声"""
    text = doc.page_content

    # 去除多余空白行（PDF 常见问题）
    text = re.sub(r'\n{3,}', '\n\n', text)

    # 去除页眉页脚常见模式（如页码）
    text = re.sub(r'^\d+\s*$', '', text, flags=re.MULTILINE)

    # 修复 PDF 中的连字符断行
    text = re.sub(r'(\w)-\n(\w)', r'\1\2', text)

    # 去除 HTML 标签（网页文档）
    text = re.sub(r'<[^>]+>', '', text)

    # 过滤过短的页（可能是目录页/空白页）
    if len(text.strip()) < 50:
        return None

    return Document(
        page_content=text.strip(),
        metadata=doc.metadata
    )

cleaned = [d for d in (clean_document(doc) for doc in all_docs) if d]

分块策略详解

文档分块（Chunking）是 RAG 中最需要细心调优的环节。不同场景、不同文档类型，最佳分块策略差异显著。

策略一：固定大小分块（Fixed-size Chunking）

最简单的方法：按字符数或 token 数切割，允许相邻块之间有重叠区域（overlap）以保持语义连贯。

from langchain_text_splitters import CharacterTextSplitter, TokenTextSplitter

# 按字符数分块
char_splitter = CharacterTextSplitter(
    chunk_size=800,      # 每块最多 800 字符
    chunk_overlap=100,   # 前后重叠 100 字符
    separator="\n\n"     # 优先在段落边界切割
)

# 按 Token 数分块（更精确，适合计算 API 成本）
token_splitter = TokenTextSplitter(
    encoding_name="cl100k_base",  # GPT-4 使用的 tokenizer
    chunk_size=256,               # 256 tokens ≈ 200 中文字
    chunk_overlap=30
)

chunks = char_splitter.split_documents(cleaned)
print(f"分块后：{len(chunks)} 个片段")

# 查看分块统计
lengths = [len(c.page_content) for c in chunks]
print(f"平均长度：{sum(lengths)/len(lengths):.0f} 字符")
print(f"最短：{min(lengths)}，最长：{max(lengths)}")

策略二：递归字符分块（Recursive Character Splitting）

LangChain 最推荐的通用分块方式。它按分隔符优先级递归尝试：先按段落（\n\n），再按句子（\n），再按词（空格），最后按字符。尽可能保持语义完整性。

from langchain_text_splitters import RecursiveCharacterTextSplitter

recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
    separators=[
        "\n\n",   # 段落
        "\n",     # 换行
        "。",     # 中文句号
        "；",     # 中文分号
        "，",     # 中文逗号
        " ",      # 空格
        ""        # 字符级（最后手段）
    ]
)

chunks = recursive_splitter.split_documents(docs)

# 中文文档专用（按语义标点）
chinese_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    model_name="gpt-4",
    chunk_size=300,   # tokens
    chunk_overlap=50
)

策略三：语义分块（Semantic Chunking）

最智能的分块方式：计算相邻句子之间的语义相似度，在语义跳变点切割，确保每个 chunk 内部语义连贯。

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    # breakpoint_threshold_type 控制切割灵敏度
    breakpoint_threshold_type="percentile",  # 或 "standard_deviation"
    breakpoint_threshold_amount=85,           # 相似度降低超过 85 百分位时切割
)

# 语义分块会调用 Embedding API，成本较高
semantic_chunks = semantic_splitter.create_documents([long_text])

策略四：父子分块（Parent-Child Chunking）

这是 Advanced RAG 的关键技术：用小块（Child）检索，用大块（Parent）生成。小块语义聚焦，检索精准；大块上下文丰富，生成质量高。

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 父块：大，用于最终生成上下文
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2000, chunk_overlap=200
)

# 子块：小，用于精确向量检索
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=400, chunk_overlap=50
)

vectorstore = Chroma(embedding_function=embeddings)
docstore = InMemoryStore()

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

retriever.add_documents(docs)

# 检索时：用小块匹配 → 返回对应的大块
relevant_docs = retriever.invoke("RAG 的核心流程是什么？")

Chunk Size 与 Overlap 的选择指南

分块质量评估

分块质量直接决定检索精度。评估方法：构建一组"问题-答案"标注对，测试检索系统能否找到包含答案的 chunk。

def evaluate_chunking(chunks, qa_pairs):
    """
    评估分块质量：答案命中率
    qa_pairs: [{"question": ..., "answer_keyword": ...}, ...]
    """
    hits = 0
    for qa in qa_pairs:
        # 检查答案关键词是否出现在 Top-3 检索结果中
        results = vectorstore.similarity_search(qa["question"], k=3)
        found = any(
            qa["answer_keyword"] in r.page_content
            for r in results
        )
        if found:
            hits += 1

    recall = hits / len(qa_pairs)
    print(f"答案命中率（Recall@3）: {recall:.1%}")
    return recall

# 实际使用 RAGAS 框架可以更系统地评估（第9章详述）

代码文档的专用分块

from langchain_text_splitters import Language, RecursiveCharacterTextSplitter

# LangChain 内置编程语言感知分割器
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=2000,
    chunk_overlap=100,
)

# 支持：PYTHON, JS, TS, JAVA, GO, RUST, CPP, RUBY, MARKDOWN, HTML, LATEX ...

python_code = """
class VectorStore:
    def __init__(self, dimension: int):
        self.vectors = []
        self.dimension = dimension

    def add(self, vector: list, metadata: dict):
        self.vectors.append({"vector": vector, "metadata": metadata})

    def search(self, query: list, top_k: int = 5):
        # 计算余弦相似度
        scores = [cosine_sim(query, v["vector"]) for v in self.vectors]
        sorted_idx = sorted(range(len(scores)), key=lambda i: -scores[i])
        return [self.vectors[i] for i in sorted_idx[:top_k]]
"""

code_chunks = python_splitter.create_documents([python_code])
# Python 分割器会在函数/类边界切割，保持代码完整性