RAG实现技能Skill rag-implementation

---

name: rag-implementation description: 构建检索增强生成（RAG）系统，用于LLM应用，结合向量数据库和语义搜索。当实现基于知识的AI、构建文档问答系统或集成LLMs与外部知识库时使用。

RAG 实现

掌握检索增强生成（RAG）以构建LLM应用，使用外部知识源提供准确、基于上下文的响应。

何时使用此技能

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

核心组件

1. 向量数据库

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

选项：

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

2. 嵌入

目的：将文本转换为数值向量以进行相似性搜索

模型（2026年）：

模型	维度	最佳用途
voyage-3-large	1024	Claude应用（Anthropic推荐）
voyage-code-3	1024	代码搜索
text-embedding-3-large	3072	OpenAI应用、高精度
text-embedding-3-small	1536	OpenAI应用、成本效益高
bge-large-en-v1.5	1024	开源、本地部署
multilingual-e5-large	1024	多语言支持

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 = "

".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

# 小块用于精确检索，大块用于上下文
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
)

# 添加文档（分割子块，存储父块）
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=["

", "
", ". ", " ", ""]  # 按顺序尝试
)

chunks = splitter.split_documents(documents)

基于令牌的分割

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": "技术"},
    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()}

资源

• LangChain RAG教程
• LangGraph RAG示例
• Pinecone最佳实践
• Voyage AI嵌入
• RAG评估指南

最佳实践

1. 块大小：在上下文（较大）和特异性（较小）之间平衡 - 通常500-1000令牌
2. 重叠：使用10-20%重叠以在边界保留上下文
3. 元数据：包括来源、页面、时间戳以进行过滤和调试
4. 混合搜索：结合语义和关键词搜索以获得最佳召回率
5. 重排序：用于精度关键应用时使用交叉编码器重排序
6. 引用：始终返回源文档以提高透明度
7. 评估：持续测试检索质量和答案准确性
8. 监控：在生产中跟踪检索指标和延迟

常见问题

• 检索差：检查嵌入质量、块大小、查询表述
• 不相关结果：添加元数据过滤、使用混合搜索、重排序
• 缺少信息：确保文档正确索引、检查分块
• 查询慢：优化向量存储、使用缓存、减少k
• 幻觉：改进基于上下文的提示、添加验证步骤
• 上下文太长：使用压缩或父文档检索器