代理式检索增强生成

原文：Agentic RAG

一句话

使用 LangGraph 构建一个能够自主决定何时检索文档或直接回答用户问题的检索增强生成(RAG)代理系统。

什么时候翻这页

当你需要构建一个能够自主判断是否需要从向量存储检索上下文或直接响应用户的 LLM 代理系统时。

核心概念

• 检索(Retrieval)：使用文档加载器(document loaders)、文本分割器(text splitters)、嵌入(embeddings)和向量存储(vector stores)实现语义搜索
• LangGraph Graph API：包括状态(state)、节点(nodes)、边缘(edges)和条件边缘(conditional edges)
• 代理式 RAG：代理能够自主决定是否需要检索文档或直接回答问题
• 文档分级：评估检索到的文档与用户问题的相关性
• 问题重写：当检索到的文档不相关时，改进原始用户问题

怎么做

1. 文档预处理

   • 获取用于 RAG 系统的文档
   • 将文档分割成较小的块以便索引到向量存储中

2. 创建检索工具

   • 使用内存向量存储和 OpenAI 嵌入创建向量存储
   • 使用 @tool 装饰器创建检索工具
   • 测试工具功能

3. 构建节点

   • generate_query_or_respond：调用 LLM 生成响应，决定是否使用检索工具
   • grade_documents：评估检索到的文档是否与问题相关
   • rewrite_question：当文档不相关时重写问题
   • generate_answer：基于原始问题和检索到的上下文生成最终答案

4. 组装图

   • 定义节点和边缘
   • 添加条件边缘决定下一步操作
   • 编译图

5. 运行代理 RAG

   • 测试完整的图系统

命令 / API 速查

# 安装必要的包
pip install -U langgraph "langchain[openai]" langchain-text-splitters bs4 requests

# 设置 API 密钥
_set_env("OPENAI_API_KEY")

# 创建文档加载器
def load_web_page(url: str, bs_kwargs: dict | None = None) -> list[Document]:
    response = requests.get(url)
    response.raise_for_status()
    soup = bs4.BeautifulSoup(response.text, "html.parser", **(bs_kwargs or {}))
    return [Document(page_content=soup.get_text(), metadata={"source": url})]

# 文本分割
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

# 创建向量存储和检索器
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

# 创建检索工具
@tool
def retrieve_blog_posts(query: str) -> str:
    """Search and return information about Lilian Weng blog posts."""
    docs = retriever.invoke(query)
    return "\n\n".join([doc.page_content for doc in docs])

# 定义节点函数
def generate_query_or_respond(state: MessagesState):
    response = response_model.bind_tools([retriever_tool]).invoke(state["messages"])
    return {"messages": [response]}

def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:
    # 评估文档相关性的逻辑
    ...

def rewrite_question(state: MessagesState):
    # 重写问题的逻辑
    ...

def generate_answer(state: MessagesState):
    # 生成答案的逻辑
    ...

# 组装图
workflow = StateGraph(MessagesState)
workflow.add_node(generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node(rewrite_question)
workflow.add_node(generate_answer)

# 添加边缘
workflow.add_edge(START, "generate_query_or_respond")
workflow.add_conditional_edges("generate_query_or_respond", route_on_tool_calls, {"tools": "retrieve", END: END})
workflow.add_conditional_edges("retrieve", grade_documents)
workflow.add_edge("generate_answer", END)
workflow.add_edge("rewrite_question", "generate_query_or_respond")

# 编译图
graph = workflow.compile()

与 Hello-Agents / Claude Code 的联系

与 Hello-Agents 中 LangGraph 章节相比，本教程更专注于构建一个具体的代理式 RAG 系统，展示了如何使用 LangGraph 的 Graph API 构建复杂的决策流程。Hello-Agents 提供了 LangGraph 的基础概念，而本教程则展示了如何将这些概念应用于实际的 RAG 场景，包括文档处理、检索工具创建、节点设计和图组装。

初学者易错点

1. 文档分割不当：chunk_size 和 chunk_overlap 参数设置不当可能影响检索效果
2. 工具绑定错误：忘记使用 .bind_tools() 将工具绑定到模型
3. 状态管理混乱：在 MessagesState 中正确管理消息流
4. 条件边缘逻辑错误：确保路由逻辑正确处理所有可能的路径
5. 文档评估标准不明确：定义清晰的评估标准来判断文档相关性

相关词条

• StateGraph
• MessagesState
• node
• edge
• conditional edge
• tool
• retriever
• vector store
• embedding
• document loader
• text splitter