父文档RAG：2026年完整实现

概述

父文档RAG是一种专门的检索模式，专注于检索小片段并附带大父上下文。本指南展示如何使用Chroma构建生产级系统。

为什么选择父文档RAG？

标准RAG在处理复杂查询、多跳推理或领域特定内容时常常力不从心。父文档RAG通过检索小片段并附带大父上下文来解决这些限制。

架构

查询 → [父文档预处理] → 向量搜索 → [上下文处理] → LLM → 响应
              ↓                                           ↑
         查询扩展                             重排序 + 过滤

实现

环境设置

bash
pip install langchain langchain-openai chroma tiktoken

python
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
初始化
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

父文档检索器

python
from langchain.retrievers import ParentDocumentRetriever
from langchain_chroma import ChromaVectorStore
构建向量存储
vectorstore = ChromaVectorStore.from_documents(
    documents=your_documents,
    embedding=embeddings,
    index_name="my-rag-index"
)
创建专门用于检索小片段并附带大父上下文的检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 6,
        "fetch_k": 25,
        "lambda_mult": 0.7
    }
)

文档处理

python
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader
def load_and_process_documents(directory: str) -> list[Document]:
    """加载并处理文档，用于父文档RAG。"""
    
    # 加载文档
    loader = DirectoryLoader(directory, glob="**/*.txt")
    raw_docs = loader.load()
    
    # 带重叠的分割以保留上下文
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=150,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    
    chunks = splitter.split_documents(raw_docs)
    
    # 添加元数据以支持检索小片段并附带大父上下文
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "chunk_id": i,
            "variant": "Parent Document",
            "chunk_length": len(chunk.page_content)
        })
    
    print(f"从 {len(raw_docs)} 个文档创建了 {len(chunks)} 个片段")
    return chunks
chunks = load_and_process_documents("./documents/")

RAG链

python
def create_parent_document_chain(retriever):
    """创建父文档RAG链，优化用于检索小片段并附带大父上下文。"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个知识渊博的AI助手。
        请使用以下检索到的上下文准确回答问题。
        
        上下文：
        {context}
        
        检索小片段并附带大父上下文的指南：
        - 引用上下文中的具体信息
        - 如果信息不在上下文中，请明确说明
        - 尽可能引用来源
        - 简洁但完整"""),
        ("human", "{question}")
    ])
    
    def format_context(docs: list[Document]) -> str:
        formatted = []
        for doc in docs:
            source = doc.metadata.get('source', 'Unknown')
            formatted.append(f"[来源: {source}]\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)
    
    chain = (
        {
            "context": retriever | format_context,
            "question": RunnablePassthrough()
        }
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain
构建并使用链
rag_chain = create_parent_document_chain(retriever)
answer = rag_chain.invoke("你的问题")

进阶：带来源的流式输出

python
from langchain_core.runnables import RunnableParallel
def create_rag_with_sources(retriever):
    """返回答案和来源文档的RAG。"""
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "根据上下文回答。要准确并引用来源。\n\n上下文：{context}"),
        ("human", "{question}")
    ])
    
    # 并行执行检索和格式化
    setup = RunnableParallel(
        context=retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
        question=RunnablePassthrough(),
        source_documents=retriever
    )
    
    chain = setup | {
        "answer": prompt | llm | StrOutputParser(),
        "sources": lambda x: [d.metadata.get('source') for d in x['source_documents']]
    }
    
    return chain
chain_with_sources = create_rag_with_sources(retriever)
result = chain_with_sources.invoke("主要主题是什么？")
print(f"答案: {result['answer']}")
print(f"来源: {result['sources']}")

评估

python
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
def evaluate_rag(test_cases: list[dict]) -> dict:
    """使用RAGAS评估父文档RAG质量。"""
    
    dataset = Dataset.from_list(test_cases)
    
    result = evaluate(
        dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )
    
    print(f"忠实度: {result['faithfulness']:.3f}")
    print(f"答案相关性: {result['answer_relevancy']:.3f}")
    print(f"上下文精确度: {result['context_precision']:.3f}")
    print(f"上下文召回率: {result['context_recall']:.3f}")
    
    return result
test_cases = [
    {
        "question": "关键特性有哪些？",
        "answer": rag_chain.invoke("关键特性有哪些？"),
        "contexts": [d.page_content for d in retriever.invoke("关键特性有哪些？")],
        "ground_truth": "预期答案..."
    }
]
evaluate_rag(test_cases)

性能优化技巧

结论

基于Chroma的父文档RAG为检索小片段并附带大父上下文提供了出色的基础。这里展示的模式已经过生产测试且可扩展。

从基本实现开始，使用RAGAS衡量质量，然后根据指标迭代。

---

*父文档RAG实现 | Chroma | 2026年5月*