rag-implementation

RAG 实现

掌握检索增强生成（RAG），构建使用外部知识源提供准确、有依据回复的大模型应用。

何时使用本技能

• 基于专有文档构建问答系统
• 创建具备最新、真实信息的聊天机器人
• 实现支持自然语言查询的语义搜索
• 通过有依据的回复减少幻觉
• 让大模型能够访问特定领域的知识
• 构建文档助手
• 创建带来源引用的研究工具

核心组件

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）：多样性 + 相关性
• 基于大模型：使用大模型对相关性打分

使用 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

# Initialize components
llm = ChatAnthropic(model="claude-sonnet-5")
embeddings = VoyageAIEmbeddings(model="voyage-3-large")
vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# RAG prompt
rag_prompt = ChatPromptTemplate.from_template(
    """Answer based on the context below. If you cannot answer, say so.

    Context:
    {context}

    Question: {question}

    Answer:"""
)

async def retrieve(state: RAGState) -> RAGState:
    """Retrieve relevant documents."""
    docs = await retriever.ainvoke(state["question"])
    return {"context": docs}

async def generate(state: RAGState) -> RAGState:
    """Generate answer from context."""
    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}

# Build RAG graph
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()

# Use
result = await rag_chain.ainvoke({"question": "What are the main features?"})
print(result["answer"])

详细模式与完整示例

详细的模式文档位于 references/details.md。当上述导航层级不够用时，请阅读该文件。