> ## Documentation Index
> Fetch the complete documentation index at: https://nvd-54.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# 使用 LangGraph 构建自定义 RAG 智能体
## 概述
在本教程中，我们将使用 LangGraph 构建一个[检索](/oss/python/langchain/retrieval)智能体。
LangChain 提供了内置的[智能体](/oss/python/langchain/agents)实现，基于 [LangGraph](/oss/python/langgraph/overview) 原语实现。如果需要更深度的自定义，可以直接在 LangGraph 中实现智能体。本指南演示了一个检索智能体的示例实现。[检索](/oss/python/langchain/retrieval)智能体在你希望 LLM 决定是从向量存储中检索上下文还是直接回复用户时非常有用。
在本教程结束时，我们将完成以下工作：
1. 获取并预处理用于检索的文档。
2. 将这些文档索引到语义搜索中，并为智能体创建检索工具。
3. 构建一个能够决定何时使用检索工具的智能体 RAG 系统。
### 概念
我们将涵盖以下概念：
* 使用[文档加载器](/oss/python/integrations/document_loaders)、[文本分割器](/oss/python/integrations/splitters)、[嵌入](/oss/python/integrations/embeddings)和[向量存储](/oss/python/integrations/vectorstores)进行[检索](/oss/python/langchain/retrieval)
* LangGraph [图 API](/oss/python/langgraph/graph-api)，包括状态、节点、边和条件边。
## 设置
让我们下载所需的包并设置 API 密钥：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
pip install -U langgraph "langchain[openai]" langchain-community langchain-text-splitters bs4
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import getpass
import os
def _set_env(key: str):
if key not in os.environ:
os.environ[key] = getpass.getpass(f"{key}:")
_set_env("OPENAI_API_KEY")
```
注册 LangSmith 以快速发现问题并提升 LangGraph 项目的性能。[LangSmith](https://docs.smith.langchain.com) 让你使用追踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用。
## 1. 预处理文档
1. 获取用于 RAG 系统的文档。我们将使用 [Lilian Weng 优秀博客](https://lilianweng.github.io/)中最近的三个页面。首先使用 `WebBaseLoader` 工具获取页面内容：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_community.document_loaders import WebBaseLoader
urls = [
"https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",
"https://lilianweng.github.io/posts/2024-07-07-hallucination/",
"https://lilianweng.github.io/posts/2024-04-12-diffusion-video/",
]
docs = [WebBaseLoader(url).load() for url in urls]
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
docs[0][0].page_content.strip()[:1000]
```
2. 将获取的文档分割成较小的块，以便索引到向量存储中：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_text_splitters import RecursiveCharacterTextSplitter
docs_list = [item for sublist in docs for item in sublist]
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)
```
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
doc_splits[0].page_content.strip()
```
## 2. 创建检索工具
现在我们有了分割后的文档，可以将它们索引到向量存储中，用于语义搜索。
1. 使用内存向量存储和 OpenAI 嵌入：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
vectorstore = InMemoryVectorStore.from_documents(
documents=doc_splits, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
```
2. 使用 `@tool` 装饰器创建检索工具：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.tools import tool
@tool
def retrieve_blog_posts(query: str) -> str:
"""搜索并返回 Lilian Weng 博客文章的相关信息。"""
docs = retriever.invoke(query)
return "\n\n".join([doc.page_content for doc in docs])
retriever_tool = retrieve_blog_posts
```
3. 测试工具：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
retriever_tool.invoke({"query": "types of reward hacking"})
```
## 3. 生成查询
现在我们开始构建智能体 RAG 图的组件（[节点](/oss/python/langgraph/graph-api#nodes)和[边](/oss/python/langgraph/graph-api#edges)）。
注意这些组件将在 [`MessagesState`](/oss/python/langgraph/graph-api#messagesstate) 上操作——一个包含 `messages` 键（[聊天消息](https://python.langchain.com/docs/concepts/messages/)列表）的图状态。
1. 构建 `generate_query_or_respond` 节点。它将调用 LLM 基于当前图状态（消息列表）生成响应。给定输入消息，它会决定是使用检索工具进行检索，还是直接回复用户。注意我们通过 `.bind_tools` 让聊天模型可以访问之前创建的 `retriever_tool`：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import MessagesState
from langchain.chat_models import init_chat_model
response_model = init_chat_model("gpt-5.4", temperature=0)
def generate_query_or_respond(state: MessagesState):
"""调用模型基于当前状态生成响应。给定问题，
它会决定使用检索工具进行检索，还是直接回复用户。
"""
response = (
response_model
.bind_tools([retriever_tool]).invoke(state["messages"]) # [!code highlight]
)
return {"messages": [response]}
```
2. 对随机输入进行测试：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
input = {"messages": [{"role": "user", "content": "hello!"}]}
generate_query_or_respond(input)["messages"][-1].pretty_print()
```
**输出：**
```
================================== Ai Message ==================================
Hello! How can I help you today?
```
3. 提出一个需要语义搜索的问题：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
input = {
"messages": [
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
}
]
}
generate_query_or_respond(input)["messages"][-1].pretty_print()
```
**输出：**
```
================================== Ai Message ==================================
Tool Calls:
retrieve_blog_posts (call_tYQxgfIlnQUDMdtAhdbXNwIM)
Call ID: call_tYQxgfIlnQUDMdtAhdbXNwIM
Args:
query: types of reward hacking
```
## 4. 评估文档
1. 添加一个[条件边](/oss/python/langgraph/graph-api#conditional-edges) `grade_documents`，用于判断检索到的文档是否与问题相关。我们将使用一个带有结构化输出模式 `GradeDocuments` 的模型来进行文档评分。`grade_documents` 函数将根据评分决策返回下一个要跳转到的节点名称（`generate_answer` 或 `rewrite_question`）：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from pydantic import BaseModel, Field
from typing import Literal
GRADE_PROMPT = (
"You are a grader assessing relevance of a retrieved document to a user question. \n "
"Here is the retrieved document: \n\n {context} \n\n"
"Here is the user question: {question} \n"
"If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n"
"Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question."
)
class GradeDocuments(BaseModel): # [!code highlight]
"""使用二元评分进行文档相关性检查。"""
binary_score: str = Field(
description="Relevance score: 'yes' if relevant, or 'no' if not relevant"
)
grader_model = init_chat_model("gpt-5.4", temperature=0)
def grade_documents(
state: MessagesState,
) -> Literal["generate_answer", "rewrite_question"]:
"""判断检索到的文档是否与问题相关。"""
question = state["messages"][0].content
context = state["messages"][-1].content
prompt = GRADE_PROMPT.format(question=question, context=context)
response = (
grader_model
.with_structured_output(GradeDocuments).invoke( # [!code highlight]
[{"role": "user", "content": prompt}]
)
)
score = response.binary_score
if score == "yes":
return "generate_answer"
else:
return "rewrite_question"
```
2. 使用不相关的文档在工具响应中运行：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain_core.messages import convert_to_messages
input = {
"messages": convert_to_messages(
[
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "1",
"name": "retrieve_blog_posts",
"args": {"query": "types of reward hacking"},
}
],
},
{"role": "tool", "content": "meow", "tool_call_id": "1"},
]
)
}
grade_documents(input)
```
3. 确认相关文档被正确分类：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
input = {
"messages": convert_to_messages(
[
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "1",
"name": "retrieve_blog_posts",
"args": {"query": "types of reward hacking"},
}
],
},
{
"role": "tool",
"content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
"tool_call_id": "1",
},
]
)
}
grade_documents(input)
```
## 5. 重写问题
1. 构建 `rewrite_question` 节点。检索工具可能会返回不相关的文档，这表明需要改进原始用户问题。为此，我们将调用 `rewrite_question` 节点：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langchain.messages import HumanMessage
REWRITE_PROMPT = (
"Look at the input and try to reason about the underlying semantic intent / meaning.\n"
"Here is the initial question:"
"\n ------- \n"
"{question}"
"\n ------- \n"
"Formulate an improved question:"
)
def rewrite_question(state: MessagesState):
"""重写原始用户问题。"""
messages = state["messages"]
question = messages[0].content
prompt = REWRITE_PROMPT.format(question=question)
response = response_model.invoke([{"role": "user", "content": prompt}])
return {"messages": [HumanMessage(content=response.content)]}
```
2. 试一下：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
input = {
"messages": convert_to_messages(
[
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "1",
"name": "retrieve_blog_posts",
"args": {"query": "types of reward hacking"},
}
],
},
{"role": "tool", "content": "meow", "tool_call_id": "1"},
]
)
}
response = rewrite_question(input)
print(response["messages"][-1].content)
```
**输出：**
```
What are the different types of reward hacking described by Lilian Weng, and how does she explain them?
```
## 6. 生成答案
1. 构建 `generate_answer` 节点：如果通过了评分检查，我们可以基于原始问题和检索到的上下文生成最终答案：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
GENERATE_PROMPT = (
"You are an assistant for question-answering tasks. "
"Use the following pieces of retrieved context to answer the question. "
"If you don't know the answer, just say that you don't know. "
"Use three sentences maximum and keep the answer concise.\n"
"Question: {question} \n"
"Context: {context}"
)
def generate_answer(state: MessagesState):
"""生成答案。"""
question = state["messages"][0].content
context = state["messages"][-1].content
prompt = GENERATE_PROMPT.format(question=question, context=context)
response = response_model.invoke([{"role": "user", "content": prompt}])
return {"messages": [response]}
```
2. 试一下：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
input = {
"messages": convert_to_messages(
[
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "1",
"name": "retrieve_blog_posts",
"args": {"query": "types of reward hacking"},
}
],
},
{
"role": "tool",
"content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
"tool_call_id": "1",
},
]
)
}
response = generate_answer(input)
response["messages"][-1].pretty_print()
```
**输出：**
```
================================== Ai Message ==================================
Lilian Weng categorizes reward hacking into two types: environment or goal misspecification, and reward tampering. She considers reward hacking as a broad concept that includes both of these categories. Reward hacking occurs when an agent exploits flaws or ambiguities in the reward function to achieve high rewards without performing the intended behaviors.
```
## 7. 组装图
现在我们将所有节点和边组装成一个完整的图：
* 从 `generate_query_or_respond` 开始，判断是否需要调用 `retriever_tool`
* 根据模型是否进行了工具调用来路由到下一步：
* 如果 `generate_query_or_respond` 返回了 `tool_calls`，则调用 `retriever_tool` 检索上下文
* 否则，直接回复用户
* 对检索到的文档内容评估其与问题的相关性（`grade_documents`）并路由到下一步：
* 如果不相关，使用 `rewrite_question` 重写问题，然后再次调用 `generate_query_or_respond`
* 如果相关，继续到 `generate_answer`，使用包含检索到的文档上下文的 [`ToolMessage`](https://reference.langchain.com/python/langchain-core/messages/tool/ToolMessage) 生成最终响应
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
workflow = StateGraph(MessagesState)
# Define the nodes we will cycle between
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")
# Route based on whether the model requested tool calls.
def route_on_tool_calls(state: MessagesState):
last_message = state["messages"][-1]
if getattr(last_message, "tool_calls", None):
return "tools"
return END
# Decide whether to retrieve
workflow.add_conditional_edges(
"generate_query_or_respond",
# Assess LLM decision (call `retriever_tool` tool or respond to the user)
route_on_tool_calls,
{
# Translate the condition outputs to nodes in our graph
"tools": "retrieve",
END: END,
},
)
# Edges taken after the `action` node is called.
workflow.add_conditional_edges(
"retrieve",
# Assess agent decision
grade_documents,
)
workflow.add_edge("generate_answer", END)
workflow.add_edge("rewrite_question", "generate_query_or_respond")
# Compile
graph = workflow.compile()
```
可视化图：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))
```
## 8. 运行智能体 RAG
现在让我们通过一个问题来测试完整的图：
```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
for chunk in graph.stream(
{
"messages": [
{
"role": "user",
"content": "What does Lilian Weng say about types of reward hacking?",
}
]
}
):
for node, update in chunk.items():
print("来自节点的更新", node)
update["messages"][-1].pretty_print()
print("\n\n")
```
**输出：**
```
来自节点的更新 generate_query_or_respond
================================== Ai Message ==================================
Tool Calls:
retrieve_blog_posts (call_NYu2vq4km9nNNEFqJwefWKu1)
Call ID: call_NYu2vq4km9nNNEFqJwefWKu1
Args:
query: types of reward hacking
来自节点的更新 retrieve
================================= Tool Message ==================================
Name: retrieve_blog_posts
(Note: Some work defines reward tampering as a distinct category of misalignment behavior from reward hacking. But I consider reward hacking as a broader concept here.)
At a high level, reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering.
Why does Reward Hacking Exist?#
Pan et al. (2022) investigated reward hacking as a function of agent capabilities, including (1) model size, (2) action space resolution, (3) observation space noise, and (4) training time. They also proposed a taxonomy of three types of misspecified proxy rewards:
Let's Define Reward Hacking#
Reward shaping in RL is challenging. Reward hacking occurs when an RL agent exploits flaws or ambiguities in the reward function to obtain high rewards without genuinely learning the intended behaviors or completing the task as designed. In recent years, several related concepts have been proposed, all referring to some form of reward hacking:
来自节点的更新 generate_answer
================================== Ai Message ==================================
Lilian Weng categorizes reward hacking into two types: environment or goal misspecification, and reward tampering. She considers reward hacking as a broad concept that includes both of these categories. Reward hacking occurs when an agent exploits flaws or ambiguities in the reward function to achieve high rewards without performing the intended behaviors.
```
***
[将这些文档连接](/use-these-docs)到 Claude、VSCode 等，通过 MCP 获取实时答案。
[在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/agentic-rag.mdx)或[提交 issue](https://github.com/langchain-ai/docs/issues/new/choose)。