LangGraph 第10章：工具调用 ReAct Agent
ReAct（Reason + Act）模式是构建 Agent 的核心范式。LLM 进行推理、决定调用工具、观察工具执行结果，然后继续推理，直到完成任务。
ReAct 循环原理
ReAct 循环的基本流程：
1. 接收用户问题
2. LLM 推理：需要做什么？
3. LLM 决定：调用某个工具或直接回答
4. 如果是工具调用 -> 执行工具 -> 将结果反馈给 LLM -> 回到步骤 2
5. 如果是直接回答 -> 输出最终答案
用户输入
|
v
[LLM 推理] --> 需要工具? --> [调用工具] --> [观察结果]
| ^
| |
+--> 直接回答 |
| |
v |
[END] <----------------------+
定义工具函数
工具是 Agent 与外部世界交互的接口。使用 @tool 装饰器定义工具：
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索网络获取最新信息"""
# 实际实现会调用搜索 API
return f"关于'{query}'的搜索结果..."
@tool
def calculate(expression: str) -> str:
"""执行数学计算"""
try:
result = eval(expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {e}"
@tool
def get_weather(city: str) -> str:
"""查询城市天气"""
# 实际实现会调用天气 API
return f"{city}的天气: 晴, 25°C"
工具函数的关键元素：
| 元素 | 说明 |
|---|---|
@tool 装饰器 | 将函数标记为工具 |
| 函数名 | 工具名称，LLM 通过名称引用 |
| 文档字符串 | 工具描述，LLM 理解工具用途的依据 |
| 参数 | 工具的参数，LLM 会生成参数值 |
| 返回值 | 工具执行结果，会反馈给 LLM |
构建 ReAct Agent
方式一：使用 create_react_agent（推荐）
LangGraph 提供了便捷的 create_react_agent 函数：
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
# 准备工具
tools = [search_web, calculate, get_weather]
# 创建 LLM
llm = ChatOpenAI(model="gpt-4")
# 一行创建 ReAct Agent
agent = create_react_agent(
llm=llm,
tools=tools,
state_modifier="你是一个智能助手，可以使用工具回答用户问题。", # 系统提示词
checkpointer=MemorySaver() # 可选：启用持久化
)
# 执行 Agent
result = agent.invoke(
{"messages": [("human", "北京的天气怎么样？")]},
config={"configurable": {"thread_id": "session_1"}}
)
print(result["messages"][-1].content)
方式二：手动构建
对于需要精细控制的场景，可以手动构建 ReAct Agent：
from typing import TypedDict, Annotated, List, Literal
import json
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
# 定义工具
@tool
def search_web(query: str) -> str:
"""搜索网络"""
return f"关于'{query}'的搜索结果..."
tools = [search_web, calculate, get_weather]
# 创建 LLM 并绑定工具
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)
# 状态定义（使用内置 MessagesState）
class AgentState(MessagesState):
"""Agent 状态"""
pass
# 节点定义
def call_model(state: AgentState):
"""LLM 推理节点"""
system_prompt = SystemMessage(
content="你是一个智能助手。如果需要工具，请调用工具。否则直接回答。"
)
messages = [system_prompt] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
# 使用内置的 ToolNode
tool_node = ToolNode(tools)
# 路由函数：判断是否需要继续调用工具
def should_continue(state: AgentState) -> Literal["tools", END]:
"""决定下一步：调用工具还是结束"""
last_message = state["messages"][-1]
# 如果 LLM 决定调用工具，继续
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
# 否则结束
return END
# 构建图
builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent") # 工具执行完后回到 agent
app = builder.compile()
手动构建与 create_react_agent 的对比
| 方式 | 优势 | 劣势 |
|---|---|---|
create_react_agent | 代码简洁、开箱即用 | 定制化能力有限 |
| 手动构建 | 完全控制、可定制 | 代码量较大 |
手动构建的图结构：
+---------+
| agent |<----+
+----+----+ |
| |
+------+------+ |
| | |
[continue] [tools] |
| | |
v +---+
END
流式输出工具调用
在 Agent 执行过程中流式输出结果：
# 流式输出事件
for event in app.stream(
{"messages": [("human", "计算 2的10次方 并搜索 Python")]},
{"configurable": {"thread_id": "stream_demo"}}
):
for node, value in event.items():
if node == "agent":
messages = value.get("messages", [])
for msg in messages:
if hasattr(msg, "tool_calls") and msg.tool_calls:
for tc in msg.tool_calls:
print(f"\n[调用工具] {tc['name']}({tc['args']})")
elif msg.content:
print(f"[思考] {msg.content[:100]}...")
elif node == "tools":
print(f"[工具结果] {value['messages'][0].content[:100]}...")
AgentExecutor 循环控制
设置最大迭代次数
防止 Agent 陷入无限循环：
from langgraph.graph import StateGraph, MessagesState, START
class ControlledAgentState(MessagesState):
iteration_count: int
def call_model_with_limit(state: ControlledAgentState):
"""带迭代限制的推理节点"""
count = state.get("iteration_count", 0)
if count >= 5: # 最大 5 次迭代
return {
"messages": [("ai", "我已达到最大推理次数，基于当前信息给出答案...")],
"iteration_count": count + 1
}
system_prompt = SystemMessage(content="你是一个智能助手...")
messages = [system_prompt] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response], "iteration_count": count + 1}
自定义 Stopping 条件
def custom_stop_condition(state: ControlledAgentState):
"""自定义停止条件"""
last_message = state["messages"][-1]
# 条件1：达到最大迭代次数
if state["iteration_count"] >= 10:
return END
# 条件2：LLM 直接回答（无工具调用）
if not hasattr(last_message, "tool_calls") or not last_message.tool_calls:
return END
# 条件3：特定工具调用表示完成
for tc in last_message.tool_calls:
if tc["name"] == "final_answer":
return END
return "tools"
ReAct Agent 设计模式
| 模式 | 描述 | 适用场景 |
|---|---|---|
| 标准 ReAct | 推理->行动->观察 循环 | 通用 Agent |
| 带记忆的 ReAct | 保留历史执行的上下文 | 多轮 Agent 交互 |
| 并行工具 ReAct | LLM 一次调用多个工具 | 需要同时获取多种信息 |
| 分层 ReAct | Agent 可调用子 Agent | 复杂任务分解 |
| Human-in-the-Loop | Agent 在关键节点征求人类意见 | 高安全性场景 |
完整示例：多功能 Agent
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
@tool
def search_docs(query: str) -> str:
"""搜索技术文档"""
return f"文档搜索结果: {query}..."
@tool
def run_python_code(code: str) -> str:
"""运行 Python 代码"""
try:
exec_globals = {}
exec(code, exec_globals)
return "代码执行成功"
except Exception as e:
return f"执行错误: {e}"
@tool
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return f"当前时间: {datetime.now().isoformat()}"
tools = [search_docs, run_python_code, get_current_time, calculate]
llm = ChatOpenAI(model="gpt-4")
agent = create_react_agent(
llm=llm,
tools=tools,
state_modifier="你是一个全能助手，可以使用工具回答问题。",
checkpointer=MemorySaver()
)
# 测试
queries = [
"现在几点了？",
"计算 1234 * 5678",
"搜索 Python 的 list 用法",
"帮我写一个计算斐波那契数列的代码",
]
for q in queries:
result = agent.invoke(
{"messages": [("human", q)]},
config={"configurable": {"thread_id": "demo"}}
)
print(f"Q: {q}")
print(f"A: {result['messages'][-1].content}\n")
ReAct Agent 是 LangGraph 最强大的模式之一，它让 LLM 具备了使用工具、与环境交互的能力。下一章我们将介绍 Human-in-the-Loop 机制，让人类可以参与到 Agent 的执行过程中。