【LangGraph】ReAct构建-LangGraph简单实现
从零手写一个 ReAct 智能体:基于 LangGraph 的完整实现
本文将带你从零开始,使用 LangGraph 构建一个经典的 ReAct(Reason + Act)智能体。你将掌握如何自定义状态、节点、工具调用逻辑与控制流,真正理解智能体背后的运行机制。
什么是 ReAct 智能体?
ReAct(Reasoning + Acting)是一种结合推理(Reason)与行动(Act)的智能体框架。它通过“思考 → 调用工具 → 再思考 → 回答”的循环,显著提升了大语言模型(LLM)在复杂任务中的表现力和可靠性。
LangChain 提供了 create_react_agent
这样的高层封装,但如果你想深度定制行为逻辑(比如修改推理策略、插入中间节点、控制工具调用条件等),就需要“从零实现”一个 ReAct 智能体。这正是本文的目标。
准备工作
首先,安装必要依赖并配置 API 密钥:
import getpass
import osdef _set_env(var: str):if not os.environ.get(var):os.environ[var] = getpass.getpass(f"{var}: ")_set_env("OPENAI_API_KEY")
建议同时注册 LangSmith,它能帮助你可视化追踪智能体的每一步执行过程,极大提升调试效率。
第一步:定义智能体状态(State)
我们使用 TypedDict
定义一个最简状态结构,只包含消息历史:
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messagesclass AgentState(TypedDict):messages: Annotated[Sequence[BaseMessage], add_messages]
这里 add_messages
是一个 reducer,用于自动合并新消息到历史列表中,避免手动管理状态。
第二步:绑定模型与工具
我们使用 OpenAI 的 gpt-4o-mini
模型,并定义一个简单的天气查询工具:
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool@tool
def get_weather(location: str):"""获取指定地点的天气信息"""if any(city in location.lower() for city in ["sf", "san francisco"]):return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."else:return f"I am not sure what the weather is in {location}"tools = [get_weather]
model = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)
注意:bind_tools(tools)
会自动让模型具备调用这些工具的能力。
第三步:定义图节点(Nodes)与边(Edges)
1. 工具执行节点(tool_node)
当模型决定调用工具时,该节点负责执行并返回结果:
import json
from langchain_core.messages import ToolMessagetools_by_name = {tool.name: tool for tool in tools}def tool_node(state: AgentState):outputs = []for tool_call in state["messages"][-1].tool_calls:result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])outputs.append(ToolMessage(content=json.dumps(result),name=tool_call["name"],tool_call_id=tool_call["id"],))return {"messages": outputs}
2. 模型推理节点(call_model)
该节点负责调用 LLM,传入系统提示和完整对话历史:
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfigdef call_model(state: AgentState, config: RunnableConfig):system_prompt = SystemMessage("You are a helpful AI assistant...")response = model.invoke([system_prompt] + state["messages"], config)return {"messages": [response]}
3. 控制流判断(should_continue)
决定是否继续调用工具,还是结束对话:
def should_continue(state: AgentState):last_message = state["messages"][-1]return "continue" if last_message.tool_calls else "end"
第四步:构建状态图(Graph)
使用 StateGraph
将上述组件组装成一个循环工作流:
from langgraph.graph import StateGraph, ENDworkflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")# 条件边:agent → tools 或 END
workflow.add_conditional_edges("agent",should_continue,{"continue": "tools", "end": END}
)# 工具执行后返回 agent
workflow.add_edge("tools", "agent")graph = workflow.compile()
你可以通过 graph.get_graph().draw_mermaid_png()
可视化该流程图(需安装额外依赖)。
第五步:运行智能体
现在,让我们测试它能否正确回答天气问题:
def print_stream(stream):for s in stream:s["messages"][-1].pretty_print()inputs = {"messages": [("user", "what is the weather in sf")]}
print_stream(graph.stream(inputs, stream_mode="values"))
输出如下:
Human: what is the weather in sf
AI: [调用 get_weather 工具]
Tool: "It's sunny in San Francisco..."
AI: The weather in San Francisco is sunny! ...
完美!智能体成功完成“推理 → 调用 → 再推理 → 回答”的完整 ReAct 循环。
总结与扩展
通过本文,你已掌握如何用 LangGraph 从零构建一个 ReAct 智能体。这种低层实现方式赋予你极大的灵活性:
- 可插入日志节点、验证节点、审批节点;
- 可替换为结构化输出(如 Pydantic);
- 可集成邮件、数据库、API 等外部动作;
- 可结合记忆、多智能体协作等高级模式。
正如官方文档所强调的:“LangGraph 让你轻松定制这个基础结构,以适应你的具体用例”。
提示:虽然本文实现的是简化版 ReAct(未显式生成“Thought”文本),但它完全符合 ReAct 的核心思想——交替进行推理与行动。如需显式推理步骤,可在 prompt 中强制模型输出
Thought: ...
,并在解析时提取。
参考
- 官方教程:How to create a ReAct agent from scratch
- LangGraph 文档:https://langchain-ai.github.io/langgraph/