Parent Document Retriever 与多向量索引

[进阶选读] 本篇深入讲解父子分块的内部机制和多向量索引技术。如果你在第 02 篇中理解了父子分块的概念并希望深入实现细节，或者遇到了"检索精准但上下文不足"的问题，请阅读本篇。

1.1 检索粒度的两难困境

Parent Document Retriever工作原理
Parent Document Retriever 核心思路——小块精准匹配，父文档丰富上下文

RAG 系统的分块策略面临一个内在矛盾：

小 chunk（如 128 token） 的优势是检索精准——向量空间中，短文本的语义更集中，不同主题之间的距离更明显，Top-K 结果更少噪音。劣势是信息不完整——128 个 token 很可能只包含一个段落的中间部分，缺少前后文，模型拿到的上下文不足以生成完整答案。

大 chunk（如 1024 token） 的优势是信息完整——包含完整的论述单元，模型有足够的上下文。劣势是检索不准——向量混合了多个主题的语义，在检索时容易被其他主题的查询误命中，或者因为语义过于分散而在关键查询上排名靠后。

分块方式	检索精准度	信息完整性	适用场景
小 chunk（128 token）	高	低	精确问答，知识点密集型文档
大 chunk（512+ token）	低	高	需要上下文的推理任务
Parent Document Retriever	高	高	通用场景，推荐默认选择

解决这个矛盾的思路是：用小 chunk 做索引，用大 chunk 做上下文。这正是 Parent Document Retriever 的核心设计。

1.2 Parent Document Retriever：核心原理

Parent Document Retriever 把文档组织成两个层次：

子文档（child）：细粒度分块，128-256 token，专门用于向量索引和相似性检索
父文档（parent）：较大的文档块（或整篇文档），存储在文档存储中，供检索后返回

检索流程：

用户 query 经过向量化
在子文档的向量索引中找到最相似的子文档
根据子文档的 parent_id 找到对应的父文档
返回父文档作为上下文（而不是子文档本身）

这样，检索精准度由子文档保证，上下文完整性由父文档提供。

1.2.1 LangChain（一个构建 AI 应用的 Python 框架，提供 ParentDocumentRetriever 等现成的 RAG 组件）实现

python

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever

# 1. 加载文档
loader = TextLoader("./knowledge_base/langchain_docs.txt")
documents = loader.load()

# 2. 定义两级分块策略
# 父文档切分：较大的块，保留上下文
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=100,
)

# 子文档切分：更小的块，用于精确检索
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=20,
)

# 3. 设置存储
# 向量存储：存放子文档的向量（用于检索）
vectorstore = Chroma(
    collection_name="child_chunks",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
)

# 文档存储：存放父文档的原始内容（用于返回上下文）
docstore = InMemoryStore()

# 4. 创建 ParentDocumentRetriever
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 5. 索引文档（只需一次）
retriever.add_documents(documents)

# 6. 检索时自动返回父文档
results = retriever.invoke("LangGraph 的 Send API 如何实现并行任务？")
for doc in results:
    print(f"父文档长度：{len(doc.page_content)} 字符")
    print(doc.page_content[:300])
    print("---")

如果不指定 parent_splitter，则整篇文档作为父文档：

python

# 不分割父文档，直接用整篇文档作为上下文
retriever_full = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    # 不传 parent_splitter：命中子 chunk 后返回整篇原始文档
)

适合文档本身长度适中（1000-3000 token），不需要进一步切分的场景。

1.3 Summary Indexing：摘要向量索引

对于长文档或结构复杂的文档（如技术手册、法律合同），原始文本的向量质量较差——文档中包含大量代码、表格、列表，这些内容的语义在向量空间中表现不佳。

解决方案：为每个文档片段生成 LLM 摘要，对摘要做向量索引，检索命中摘要后返回原始文档。

摘要是对原始内容的语义提炼，向量质量更高；同时返回原始文档保证信息完整性。

python

from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.storage import InMemoryStore
from langchain_core.messages import HumanMessage
import uuid

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 存储层
vectorstore = Chroma(
    collection_name="summaries",
    embedding_function=embeddings,
)
docstore = InMemoryStore()  # 存储原始文档

def build_summary_index(documents: list[Document]) -> None:
    """为文档列表构建摘要索引"""
    summary_docs = []
    doc_pairs = []

    for doc in documents:
        # 生成摘要
        response = llm.invoke([
            HumanMessage(content=f"""为以下文档内容生成一段简洁、信息密度高的摘要（150字以内），
保留关键概念、数字和专有名词：

{doc.page_content}

摘要：""")
        ])
        summary_text = response.content

        # 为原始文档生成唯一 ID
        doc_id = str(uuid.uuid4())

        # 构建摘要文档，metadata 中记录原始文档 ID
        summary_doc = Document(
            page_content=summary_text,
            metadata={
                "doc_id": doc_id,
                "source": doc.metadata.get("source", ""),
                "type": "summary",
            }
        )
        summary_docs.append(summary_doc)
        doc_pairs.append((doc_id, doc))

    # 将摘要存入向量库
    vectorstore.add_documents(summary_docs)

    # 将原始文档存入文档存储
    docstore.mset(doc_pairs)

def retrieve_by_summary(query: str, k: int = 4) -> list[Document]:
    """通过摘要检索，返回原始文档"""
    # 在摘要向量空间中搜索
    summary_results = vectorstore.similarity_search(query, k=k)

    # 根据 doc_id 获取原始文档
    doc_ids = [doc.metadata["doc_id"] for doc in summary_results]
    original_docs = docstore.mget(doc_ids)

    return [doc for doc in original_docs if doc is not None]

# 使用示例
from langchain_community.document_loaders import DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = DirectoryLoader("./docs", glob="**/*.md")
raw_docs = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=1500, chunk_overlap=150)
chunks = splitter.split_documents(raw_docs)

build_summary_index(chunks)

results = retrieve_by_summary("如何处理 RAG 系统中的长文档？")
for doc in results:
    print(doc.page_content[:500])

1.4 假设性问题索引：Hypothetical Questions Indexing

用户的查询通常是问句形式，而文档内容通常是陈述句形式。两者在向量空间中可能存在语义距离，直接检索效果不理想。

解决方案：为每个文档片段生成"它能回答哪些问题"，对这些假设性问题做向量索引。用户的 query 和索引中的"问题"形式一致，语义匹配效果更好。

python

from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.storage import InMemoryStore
from langchain_core.messages import HumanMessage
import uuid

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vectorstore = Chroma(
    collection_name="hypothetical_questions",
    embedding_function=embeddings,
)
docstore = InMemoryStore()

def build_hypothetical_question_index(documents: list[Document]) -> None:
    """为文档生成假设性问题并建立索引"""
    all_question_docs = []
    doc_pairs = []

    for doc in documents:
        # 为每个文档生成可能的问题
        response = llm.invoke([
            HumanMessage(content=f"""基于以下文档内容，生成3个用户可能提出的问题。
这些问题应该可以通过这段文档内容直接回答。
每行输出一个问题，不加序号或其他符号。

文档内容：
{doc.page_content}

问题：""")
        ])

        questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
        doc_id = str(uuid.uuid4())

        # 为每个问题创建一个文档，指向同一个原始文档
        for question in questions:
            question_doc = Document(
                page_content=question,
                metadata={
                    "doc_id": doc_id,
                    "source": doc.metadata.get("source", ""),
                    "type": "hypothetical_question",
                }
            )
            all_question_docs.append(question_doc)

        doc_pairs.append((doc_id, doc))

    vectorstore.add_documents(all_question_docs)
    docstore.mset(doc_pairs)

def retrieve_by_hypothetical_questions(query: str, k: int = 3) -> list[Document]:
    """通过假设性问题检索，返回原始文档，自动去重"""
    question_results = vectorstore.similarity_search(query, k=k * 3)  # 多检索一些，去重后保留 k 个

    # 去重：同一个 doc_id 只保留一次
    seen_ids = set()
    unique_doc_ids = []
    for doc in question_results:
        doc_id = doc.metadata["doc_id"]
        if doc_id not in seen_ids:
            seen_ids.add(doc_id)
            unique_doc_ids.append(doc_id)
        if len(unique_doc_ids) >= k:
            break

    original_docs = docstore.mget(unique_doc_ids)
    return [doc for doc in original_docs if doc is not None]

1.5 三种策略对比与选型

策略	核心思路	最适合的场景	主要代价
Parent Document Retriever	小 chunk 检索，大 chunk 上下文	通用场景，结构清晰的文档	存储空间翻倍，需要两层存储
Summary Indexing	摘要向量质量更高	代码、表格等非自然语言内容多的文档；长文档	索引构建时需要大量 LLM 调用，成本较高
Hypothetical Questions	问题与问题语义匹配	知识库问答；文档风格和用户查询风格差异大	索引构建成本高；问题生成质量影响检索效果

选型建议：

默认从 Parent Document Retriever 开始——实现简单，效果稳定，适合大多数场景。
文档中包含大量代码块、表格、技术规范时，考虑 Summary Indexing——自然语言摘要的向量质量远好于代码/表格混合文本。
用户查询习惯与文档表述风格差异明显时（如用户问"怎么做X"，文档写的是"X的实现方案是..."`），考虑 Hypothetical Questions Indexing。
对于关键文档，可以组合使用多种策略，用多路召回融合结果（参见混合检索章节）。

1.6 端到端完整实现

以下是一个将三种策略融合为单一检索器的实现，自动选择最优策略：

python

from typing import Literal
from langchain_core.documents import Document
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.retrievers import ParentDocumentRetriever
import uuid

class MultiVectorRetriever:
    """多向量索引检索器：支持三种策略"""

    def __init__(
        self,
        strategy: Literal["parent_doc", "summary", "hypothetical_questions"] = "parent_doc",
        llm_model: str = "gpt-4o-mini",
        embedding_model: str = "text-embedding-3-small",
    ):
        self.strategy = strategy
        self.llm = ChatOpenAI(model=llm_model, temperature=0)
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.docstore = InMemoryStore()

        collection_name = f"multi_vector_{strategy}"
        self.vectorstore = Chroma(
            collection_name=collection_name,
            embedding_function=self.embeddings,
        )

        if strategy == "parent_doc":
            self._retriever = ParentDocumentRetriever(
                vectorstore=self.vectorstore,
                docstore=self.docstore,
                child_splitter=RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=20),
                parent_splitter=RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100),
            )

    def add_documents(self, documents: list[Document]) -> None:
        if self.strategy == "parent_doc":
            self._retriever.add_documents(documents)
        elif self.strategy == "summary":
            self._build_summary_index(documents)
        elif self.strategy == "hypothetical_questions":
            self._build_hq_index(documents)

    def retrieve(self, query: str, k: int = 4) -> list[Document]:
        if self.strategy == "parent_doc":
            return self._retriever.invoke(query)
        elif self.strategy == "summary":
            return self._retrieve_by_summary(query, k)
        elif self.strategy == "hypothetical_questions":
            return self._retrieve_by_hq(query, k)
        return []

    def _build_summary_index(self, documents: list[Document]) -> None:
        from langchain_core.messages import HumanMessage
        summary_docs, pairs = [], []
        for doc in documents:
            resp = self.llm.invoke([
                HumanMessage(content=f"为以下内容生成150字以内的信息密度高的摘要：\n\n{doc.page_content}\n\n摘要：")
            ])
            doc_id = str(uuid.uuid4())
            summary_docs.append(Document(
                page_content=resp.content,
                metadata={"doc_id": doc_id}
            ))
            pairs.append((doc_id, doc))
        self.vectorstore.add_documents(summary_docs)
        self.docstore.mset(pairs)

    def _retrieve_by_summary(self, query: str, k: int) -> list[Document]:
        results = self.vectorstore.similarity_search(query, k=k)
        ids = [d.metadata["doc_id"] for d in results]
        return [d for d in self.docstore.mget(ids) if d]

    def _build_hq_index(self, documents: list[Document]) -> None:
        from langchain_core.messages import HumanMessage
        all_q_docs, pairs = [], []
        for doc in documents:
            resp = self.llm.invoke([
                HumanMessage(content=f"基于以下内容，生成3个问题，每行一个：\n\n{doc.page_content}")
            ])
            doc_id = str(uuid.uuid4())
            for q in resp.content.strip().split("\n"):
                q = q.strip()
                if q:
                    all_q_docs.append(Document(
                        page_content=q,
                        metadata={"doc_id": doc_id}
                    ))
            pairs.append((doc_id, doc))
        self.vectorstore.add_documents(all_q_docs)
        self.docstore.mset(pairs)

    def _retrieve_by_hq(self, query: str, k: int) -> list[Document]:
        results = self.vectorstore.similarity_search(query, k=k * 3)
        seen, unique_ids = set(), []
        for d in results:
            did = d.metadata["doc_id"]
            if did not in seen:
                seen.add(did)
                unique_ids.append(did)
            if len(unique_ids) >= k:
                break
        return [d for d in self.docstore.mget(unique_ids) if d]


# 使用示例
from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader("./docs", glob="**/*.md")
docs = loader.load()

retriever = MultiVectorRetriever(strategy="parent_doc")
retriever.add_documents(docs)

results = retriever.retrieve("什么是 RAG 系统的评估指标？")
for doc in results:
    print(doc.page_content[:200])
    print("---")