用 LangGraph 写 ReAct

第 7 章我们手写过一个 ReAct 智能体：循环里"模型想 → 执行工具 → 把结果喂回去"，直到模型不再要求调工具。用 LangGraph 重写时，这个循环变成一张只有两个节点的图：agent 节点负责思考并决定是否调工具，tools 节点负责执行，中间用一条条件边判断"还要不要继续"。本章给出完整可跑代码与一次真实运行轨迹。

绑定工具：model.bind_tools

让模型"知道有哪些工具可用、并返回结构化调用请求"，用 bind_tools 即可：

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

model = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),
)
model_with_tools = model.bind_tools([add])

说明：bind_tools 走的是模型原生 function calling，回复里带结构化 tool_calls 字段，框架解析可靠；另一条路是第 5~7 章的做法——不用 bind_tools，靠提示词让模型输出固定 JSON 再自行解析，灵活但对提示与容错要求高。本系列统一用 bind_tools。

图结构：agent -> tools -> agent ... 直到 END

messages 字段用 add_messages 做 Reducer，保证每轮产生的消息都追加进历史，模型始终能看到完整对话。

from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

class State(TypedDict):
    messages: Annotated[list, add_messages]

def agent(state):
    reply = model_with_tools.invoke(state["messages"])
    return {"messages": [reply]}

def should_continue(state):
    last = state["messages"][-1]
    # 最后一条消息还带着工具调用请求 -> 去执行工具；否则回答完毕，结束
    return "tools" if getattr(last, "tool_calls", None) else END

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode([add]))     # 预置节点：执行工具并回填结果
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")             # 工具结果送回 agent，形成循环
graph = builder.compile()

ToolNode 是 langgraph.prebuilt 提供的现成工具执行节点：它读出消息里的 tool_calls，逐个调用对应函数，再把 ToolMessage 追加回 messages。工具结果回到 agent 后，模型基于新信息继续推理，可能再次要求调工具，也可能给出最终回答走向 END。

完整运行与轨迹输出

def run(question):
    out = graph.invoke({"messages": [HumanMessage(question)]})
    for msg in out["messages"]:
        who = type(msg).__name__
        content = msg.content or str(getattr(msg, "tool_calls", ""))
        print(f"[{who}] {content}")
        if who == "ToolMessage":
            print(f"       工具返回：{msg.content}")

run("25 加 17 等于多少？请用工具计算")

运行一次的真实轨迹（模型措辞可能不同，结构一致）：

[HumanMessage] 25 加 17 等于多少？请用工具计算
[AIMessage] [{'name': 'add', 'args': {'a': 25, 'b': 17}, 'id': 'call_xxx'}]   <- 模型请求调工具
[ToolMessage] 42
[AIMessage] 25 加 17 等于 42。

注意中间并没有出现第三个节点：ToolMessage 回到 agent 后，条件边发现最后一条消息不再含 tool_calls，于是走向 END。如果换成"先算价格再算税费"这类问题，轨迹会变成 agent→tools→agent→tools→agent→END 的多轮循环，图本身不用改——这就是把循环交给框架的好处。

多工具与循环上限

要挂更多工具，直接扩展列表即可：

tools = [add, subtract, multiply_by]     # 每个都是 @tool 函数
model_with_tools = model.bind_tools(tools)

循环由条件边天然承载，但模型可能"停不下来"，生产上建议加保护：在 should_continue 里统计轮数（例如读 messages 中 AIMessage 个数），超过上限就返回 END，避免无限调用消耗额度。

防死循环：给条件边加上限

模型偶尔会"停不下来"，在 should_continue 里加轮次上限更稳妥：

MAX_TOOL_ROUNDS = 5

def should_continue(state):
    last = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return END                     # 不再请求工具，收尾
    # 从历史消息里数已经发生过的工具调用轮次
    used = sum(1 for m in state["messages"] if getattr(m, "tool_calls", None))
    return "tools" if used < MAX_TOOL_ROUNDS else END

计数直接从消息历史推导，无需新增状态字段；超过上限强制走 END，保住预算。

工具执行出错怎么办

ToolNode 不会让图崩溃：函数抛出的异常会被包装成 ToolMessage 回填给模型，模型看到报错文本后往往自己修正参数重试。工具内部也可自行兜底：

@tool
def divide(a: float, b: float) -> str:
    """两个数相除"""
    if b == 0:
        return "错误：除数不能为 0"
    return str(a / b)

常见问题

模型一直不调工具？检查 bind_tools 是否生效、函数 docstring 是否写清用途；必要时在系统提示里点明"涉及计算请使用工具"。
想实时展示"正在调哪个工具"？把 run() 的事后打印换成第 21 章的 stream 逐节点输出，边执行边展示。
与第 7 章手写 ReAct 的差别？手写版要自己维护 while 循环、解析 tool_calls、拼接消息；图版交给条件边与 ToolNode，逻辑等价但更可读、可持久化、可人工打断（见第 19~20 章）。

小结：用 LangGraph 写 ReAct = 两个节点（agent/tools）+ 一条条件边（还有 tool_calls 就回 agent，否则 END）；model.bind_tools 声明工具、ToolNode 负责执行，历史消息靠 add_messages 自动累积，多轮工具循环不用自己写 while。

笔记加载中…

标签：#Agent实战

返回知识库