RAG 实现

掌握检索增强生成 (RAG)，构建能够利用外部知识源提供准确、基于事实的响应的 LLM 应用程序。

何时使用此技能

• 在私有文档上构建问答系统
• 创建包含当前、事实信息的聊天机器人
• 使用自然语言查询实现语义搜索
• 通过基于事实的响应减少幻觉
• 使 LLM 能够访问特定领域知识
• 构建文档助手
• 创建带有来源引用的研究工具

核心组件

1. 向量数据库

目的：高效地存储和检索文档嵌入

可选方案：

• Pinecone：托管式、可扩展、无服务器
• Weaviate：开源、混合搜索、GraphQL
• Milvus：高性能、本地部署
• Chroma：轻量级、易于使用、本地开发
• Qdrant：快速、带过滤的搜索、基于 Rust
• pgvector：PostgreSQL 扩展、SQL 集成

2. 嵌入模型

目的：将文本转换为用于相似性搜索的数值向量

模型 (2026年):

3. 检索策略

方法：

• 密集检索：通过嵌入进行语义相似性搜索
• 稀疏检索：关键词匹配 (BM25, TF-IDF)
• 混合搜索：结合密集和稀疏检索，进行加权融合
• 多查询：生成多个查询变体
• HyDE：生成假设性文档以改进检索

4. 重排序

目的：通过重新排序结果来提高检索质量

方法：

• 交叉编码器：基于 BERT 的重排序 (ms-marco-MiniLM)
• Cohere Rerank：基于 API 的重排序
• 最大边际相关性 (MMR)：兼顾多样性与相关性
• 基于 LLM 的重排序：使用 LLM 对相关性进行评分

使用 LangGraph 快速开始

from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing import TypedDict, Annotated

class RAGState(TypedDict):
    question: str
    context: list[Document]
    answer: str

# 初始化组件
llm = ChatAnthropic(model="claude-sonnet-4-6")
embeddings = VoyageAIEmbeddings(model="voyage-3-large")
vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# RAG 提示词模板
rag_prompt = ChatPromptTemplate.from_template(
    """根据以下上下文回答问题。如果你无法回答，请如实说明。

    上下文：
    {context}

    问题：{question}

    答案："""
)

async def retrieve(state: RAGState) -> RAGState:
    """检索相关文档。"""
    docs = await retriever.ainvoke(state["question"])
    return {"context": docs}

async def generate(state: RAGState) -> RAGState:
    """根据上下文生成答案。"""
    context_text = "\n\n".join(doc.page_content for doc in state["context"])
    messages = rag_prompt.format_messages(
        context=context_text,
        question=state["question"]
    )
    response = await llm.ainvoke(messages)
    return {"answer": response.content}

# 构建 RAG 图
builder = StateGraph(RAGState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

rag_chain = builder.compile()

# 使用
result = await rag_chain.ainvoke({"question": "主要特性是什么？"})
print(result["answer"])

高级 RAG 模式

模式 1: 带 RRF 的混合搜索

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 稀疏检索器 (用于关键词匹配的 BM25)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10

# 密集检索器 (用于语义搜索的嵌入)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 使用倒数排名融合权重进行组合
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, dense_retriever],
    weights=[0.3, 0.7]  # 30% 关键词, 70% 语义
)

模式 2: 多查询检索

from langchain.retrievers.multi_query import MultiQueryRetriever

# 为更好的召回率生成多个查询变体
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm
)

# 单个查询 → 多个变体 → 合并结果
results = await multi_query_retriever.ainvoke("主题是什么？")

模式 3: 上下文压缩

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 压缩器仅提取相关部分
compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
)

# 返回文档中仅相关的部分
compressed_docs = await compression_retriever.ainvoke("特定查询")

模式 4: 父文档检索器

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

# 小 chunk 用于精确检索，大 chunk 用于提供上下文
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# 用于存储父文档的存储
docstore = InMemoryStore()

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

# 添加文档 (拆分 child，存储 parent)
await parent_retriever.aadd_documents(documents)

# 检索返回带有完整上下文的父文档
results = await parent_retriever.ainvoke("查询")

模式 5: HyDE (假设性文档嵌入)

from langchain_core.prompts import ChatPromptTemplate

class HyDEState(TypedDict):
    question: str
    hypothetical_doc: str
    context: list[Document]
    answer: str

hyde_prompt = ChatPromptTemplate.from_template(
    """写一个能够回答以下问题的详细段落：

    问题：{question}

    段落："""
)

async def generate_hypothetical(state: HyDEState) -> HyDEState:
    """为更好地检索生成假设性文档。"""
    messages = hyde_prompt.format_messages(question=state["question"])
    response = await llm.ainvoke(messages)
    return {"hypothetical_doc": response.content}

async def retrieve_with_hyde(state: HyDEState) -> HyDEState:
    """使用假设性文档进行检索。"""
    # 使用假设性文档而非原始查询进行检索
    docs = await retriever.ainvoke(state["hypothetical_doc"])
    return {"context": docs}

# 构建 HyDE RAG 图
builder = StateGraph(HyDEState)
builder.add_node("hypothetical", generate_hypothetical)
builder.add_node("retrieve", retrieve_with_hyde)
builder.add_node("generate", generate)
builder.add_edge(START, "hypothetical")
builder.add_edge("hypothetical", "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)

hyde_rag = builder.compile()

文档分块策略

递归字符文本分割器

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", ". ", " ", ""]  # 按顺序尝试
)

chunks = splitter.split_documents(documents)

基于 Token 的分割

from langchain_text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    encoding_name="cl100k_base"  # OpenAI tiktoken 编码
)

语义分块

from langchain_experimental.text_splitter import SemanticChunker

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95
)

Markdown 标题分割器

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "标题 1"),
    ("##", "标题 2"),
    ("###", "标题 3"),
]

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False
)

向量存储配置

Pinecone (无服务器)

from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore

# 初始化 Pinecone 客户端
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

# 如果需要，创建索引
if "my-index" not in pc.list_indexes().names():
    pc.create_index(
        name="my-index",
        dimension=1024,  # voyage-3-large 的维度
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

# 创建向量存储
index = pc.Index("my-index")
vectorstore = PineconeVectorStore(index=index, embedding=embeddings)

Weaviate

import weaviate
from langchain_weaviate import WeaviateVectorStore

client = weaviate.connect_to_local()  # 或 connect_to_weaviate_cloud()

vectorstore = WeaviateVectorStore(
    client=client,
    index_name="Documents",
    text_key="content",
    embedding=embeddings
)

Chroma (本地开发)

from langchain_chroma import Chroma

vectorstore = Chroma(
    collection_name="my_collection",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

pgvector (PostgreSQL)

from langchain_postgres.vectorstores import PGVector

connection_string = "postgresql+psycopg://user:pass@localhost:5432/vectordb"

vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="documents",
    connection=connection_string,
)

检索优化

1. 元数据过滤

from langchain_core.documents import Document

# 索引时添加元数据
docs_with_metadata = []
for doc in documents:
    doc.metadata.update({
        "source": doc.metadata.get("source", "unknown"),
        "category": determine_category(doc.page_content),
        "date": datetime.now().isoformat()
    })
    docs_with_metadata.append(doc)

# 检索时进行过滤
results = await vectorstore.asimilarity_search(
    "查询",
    filter={"category": "technical"},
    k=5
)

2. 最大边际相关性 (MMR)

# 平衡相关性与多样性
results = await vectorstore.amax_marginal_relevance_search(
    "查询",
    k=5,
    fetch_k=20,  # 获取 20 个，返回 5 个多样化的结果
    lambda_mult=0.5  # 0=最大多样性，1=最大相关性
)

3. 使用交叉编码器重排序

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

async def retrieve_and_rerank(query: str, k: int = 5) -> list[Document]:
    # 获取初始结果
    candidates = await vectorstore.asimilarity_search(query, k=20)

    # 重排序
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = reranker.predict(pairs)

    # 按分数排序并取前 k 个
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:k]]

4. Cohere 重排序

from langchain.retrievers import CohereRerank
from langchain_cohere import CohereRerank

reranker = CohereRerank(model="rerank-english-v3.0", top_n=5)

# 用重排序包装检索器
reranked_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

RAG 的提示词工程

带引用的上下文提示词

rag_prompt = ChatPromptTemplate.from_template(
    """根据以下上下文回答问题。请使用 [1]，[2] 等格式包含引用。

    如果你无法根据上下文回答，请说“我没有足够的信息”。

    上下文：
    {context}

    问题：{question}

    说明：
    1. 仅使用上下文中的信息
    2. 使用 [1]，[2] 格式引用来源
    3. 如果不确定，请表达不确定性

    答案（带引用）："""
)

RAG 的结构化输出

from pydantic import BaseModel, Field

class RAGResponse(BaseModel):
    answer: str = Field(description="基于上下文的答案")
    confidence: float = Field(description="置信度分数 0-1")
    sources: list[str] = Field(description="使用的源文档 ID")
    reasoning: str = Field(description="答案的简要推理过程")

# 配合结构化输出使用
structured_llm = llm.with_structured_output(RAGResponse)

评估指标

from typing import TypedDict

class RAGEvalMetrics(TypedDict):
    retrieval_precision: float  # 相关文档数 / 检索文档数
    retrieval_recall: float     # 检索到的相关文档 / 总相关文档
    answer_relevance: float     # 答案是否切题
    faithfulness: float         # 答案是否基于上下文
    context_relevance: float    # 上下文是否与问题相关

async def evaluate_rag_system(
    rag_chain,
    test_cases: list[dict]
) -> RAGEvalMetrics:
    """在测试用例上评估 RAG 系统。"""
    metrics = {k: [] for k in RAGEvalMetrics.__annotations__}

    for test in test_cases:
        result = await rag_chain.ainvoke({"question": test["question"]})

        # 检索指标
        retrieved_ids = {doc.metadata["id"] for doc in result["context"]}
        relevant_ids = set(test["relevant_doc_ids"])

        precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)
        recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)

        metrics["retrieval_precision"].append(precision)
        metrics["retrieval_recall"].append(recall)

        # 使用 LLM 作为评判者来评估质量指标
        quality = await evaluate_answer_quality(
            question=test["question"],
            answer=result["answer"],
            context=result["context"],
            expected=test.get("expected_answer")
        )
        metrics["answer_relevance"].append(quality["relevance"])
        metrics["faithfulness"].append(quality["faithfulness"])
        metrics["context_relevance"].append(quality["context_relevance"])

    return {k: sum(v) / len(v) for k, v in metrics.items()}