生命不息,折腾不止。前面几课搭的 Agent 只会「想」不会「做」——这一课用 @tool 装饰器 + ToolNode 给它装上手和脚,让模型能算数、查天气、调 API,跑通一个真正的 ReAct「想—做—再想」循环。

一、会说话的 Agent,为什么还是半个残废

上一课我们给 Agent 接上了记忆和「后悔药」,但它有个硬伤:只能靠脑子(模型)里的知识回答,长不出手。你问它「42×7 等于多少」「北京今天几度」,它要么硬算(还经常算错),要么直接坦承「我不知道」。很多活儿——查实时数据、调自己写的接口、读写数据库、跑一段代码——模型天生干不了。

解决办法不是换个更强的模型,而是给它工具(tool):把「查天气」「算乘法」这些外部能力包装成一个带说明的函数,让模型在需要的时候,主动发起一次「工具调用(tool call)」,工具跑完把结果塞回对话,模型再基于结果接着往下说。

这套「推理(想)→ 行动(做)→ 再推理」的动作,就是 ReAct 循环(Reasoning + Acting)。LangGraph 里搭这个循环,就两个主角:@tool 装饰器负责「造工具」,ToolNode 负责「跑工具」。

二、@tool 装饰器:一个普通函数,一秒变工具

工具的定义,本质上就是一个函数加一段「说明书」。先装依赖(前面几课已经在用的那套):

1
pip install langgraph langchain-core langchain-openai

然后写第一个工具:

1
2
3
4
5
6
from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
"""两个整数相乘,返回它们的乘积。"""
return a * b

就三句话,但每一句都有讲究:

  • @tool 来自 langchain_core.tools,不是 langgraph,别导错包;
  • docstring 是给模型看的——模型靠这段描述判断「这工具是干嘛的、什么时候该用我」。写糊了它就会乱叫;写清楚了它才用得对;
  • 参数类型注解(a: int)会自动生成输入 schema,模型照着这个结构填参数。

工具可以脱机测试一下,跟 Runnable 一样 invoke:

1
multiply.invoke({"a": 6, "b": 7})   # -> 42

如果想改默认的名称或描述,给装饰器显式传参就行:

1
2
3
@tool(name="字符串长度", description="统计一段文本的字符数")
def strlen(text: str) -> int:
return len(text)

有了工具,下一步就是把它「上交」给模型。

三、bind_tools:把工具的「简历」递给模型

模型不会自动知道有哪些工具,你得把每个工具的「名字 + 参数结构 + 说明书」打包挂到它身上。这一步用 bind_tools:

1
2
3
4
5
6
7
8
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
model="gpt-4o-mini", # 任意 OpenAI 兼容模型都行
base_url="https://ai.aklibk.com/v1", # 中转站的 OpenAI 兼容入口
api_key="sk-你的key", # 国内直连、免绑卡、人民币按量付费
)
model_with_tools = model.bind_tools([multiply, get_weather])

base_url 填 ai.aklibk.com 这种中转站时,model 想接 GPT、Claude 还是 DeepSeek 都行,全走同一套 OpenAI 兼容协议,不用为每家单独写一套连接代码——这也是前面几课反复用它的原因。

挂上工具之后,模型的行为就变了:拿到问题如果它觉得「这事得靠工具」,返回的就不再是普通文本,而是一条带着 tool_calls 字段的 AIMessage。里面写清楚了「我要调哪个工具、传哪些参数」,但它自己不会真的去执行——执行是下一节的活。

四、ToolNode:替模型把工具真跑起来

模型只负责「下单」,真正「出货」的是 ToolNode:

1
2
3
from langgraph.prebuilt import ToolNode

tool_node = ToolNode([multiply, get_weather])

ToolNode 是 LangGraph 内置的专用节点,它读入「最后一条是带 tool_calls 的 AIMessage」的状态,把每个工具调用真正执行一遍,再把结果写成 ToolMessage 塞回消息列表。它自带的三个能力,值得单独记下来:

  • 同步、异步工具通吃:async def 写的工具,它也会自动 await;
  • 多个调用并发跑:一条消息里如果模型同时要调「查天气」和「算乘法」,它会并行执行,不排队等;
  • 错误处理默认开启:工具抛异常不会整个程序崩掉,而是被接住、包成一条 status='error' 的 ToolMessage 回给模型,模型看到报错能自己琢磨着换个姿势重来。

五、搭起 ReAct 循环:想 → 做 → 再想

现在把图和模型接起来。套路和前面几课完全一致:也是在 StateGraph 上挂节点、连边,只是多了个 tools 节点和一条条件边。完整代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

@tool
def get_weather(location: str) -> str:
"""查询指定城市的当前天气。"""
return f"{location} 现在是 20℃,晴。"

@tool
def multiply(a: int, b: int) -> int:
"""两个整数相乘。"""
return a * b

tools = [get_weather, multiply]
tool_node = ToolNode(tools)

model = ChatOpenAI(
model="gpt-4o-mini",
base_url="https://ai.aklibk.com/v1",
api_key="sk-你的key",
)
model_with_tools = model.bind_tools(tools)


def call_model(state: MessagesState):
# 把当前消息交给带工具的模型,它会决定「直接回答」还是「发起工具调用」
response = model_with_tools.invoke(state["messages"])
return {"messages": [response]}


def should_continue(state: MessagesState):
last = state["messages"][-1]
# 模型发了 tool_calls(非空)→ 去 tools 节点真正执行
if last.tool_calls:
return "tools"
# 没有工具调用 → 直接收工
return END


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue, ["tools", END])
builder.add_edge("tools", "call_model") # 工具跑完,结果带回给模型接着想
graph = builder.compile()

graph.invoke({
"messages": [{"role": "user", "content": "帮我算 42×7 等于多少,再查一下北京的天气"}]
})

这段代码没几行,但那个「想—做—再想」的循环是精髓:

  1. START → call_model:模型接到问题,发现两个活儿都得靠工具,返回一条带两个 tool_calls 的 AIMessage;
  2. should_continue 看到 last.tool_calls 非空,路由到 tools;
  3. ToolNode 并行跑掉「算 42×7」和「查北京天气」,产出两条 ToolMessage;
  4. tools 有一条固定边回到 call_model,模型拿着工具结果,组织成最终回答:「42×7=294,北京现在 20℃ 晴」;
  5. 这一次它不再发起工具调用,should_continue 返回 END,循环结束。

这个循环可以转不止一圈:模型完全可能「先查天气 → 再根据天气决定要不要再调个『穿衣建议』工具 → 最后回答」。last.tool_calls 的这种判断,就是标准的 ReAct 路由逻辑——(tools_condition 是 LangGraph 内置的同款简写,这里手写出来是为了把判断逻辑摆到台面上看得清)。

六、进阶一:错误处理,给模型留条后路

模型迟早会传错参数。默认 ToolNode 会把异常消化成 status='error' 的 ToolMessage,模型看了能自己重试;你也可以把报错文案换成更友好的提示:

1
2
# 出错了不用原样抛出,改成一句它能看懂的人话
tool_node = ToolNode(tools, handle_tool_errors="参数不对,请换一组再试")

或者干脆 handle_tool_errors=False,让异常原样往上抛,自己在外面 try/except 兜底——想完全掌控时用这个。

异步工具也一样顺手,直接 async def 包装即可:

1
2
3
4
5
6
7
8
9
from langchain_core.tools import tool
import httpx

@tool
async def fetch_price(symbol: str) -> str:
"""查询某个交易对的最新价格。"""
async with httpx.AsyncClient() as c:
r = await c.get(f"https://api.example.com/price/{symbol}")
return r.json()["price"]

ToolNode 会替你 await,图上别的代码一行都不用改。

七、三个坑 + 下一步

坑一:忘了 bind_tools。 工具定义得再好,只要没挂到模型上,模型压根不知道有这回事,永远不发 tool_calls。@tool、bind_tools、ToolNode 三件套缺一不可。

坑二:docstring 太糊。 模型全靠描述挑工具、填参数。像 """处理数据。""" 这种描述等于没写,模型会乱叫工具、乱传参。写清楚「输入什么、返回什么、什么时候用」。

坑三:把 create_react_agent 当唯一解。 在 LangGraph v1 里,create_react_agent 这个预置接口已经被标记弃用,官方引导迁移到 LangChain 的 create_agent。如果你想快速起一个工具型 Agent,用它没问题;但要接上前几课讲的中断审批、checkpointer 持久化这些自定义编排,还是得像本文这样自己手搭图——用到的 StateGraph、ToolNode、条件边都是稳定的底层 API,不过时。

到这里,「会想」的 Agent 终于长出了「会做」的手。但会干活还不行,得有人看着——把这课的工具,跟前两课的「人工审批 + 持久化」拼在一起,才是一个能落地的完整应用。

生命不息,折腾不止。下一篇「LangGraph 实战第 5 课」:把这三课的成果拼起来,搭一个「自动查订单 + 算优惠 + 人工确认发货」的工单 Agent——既能调工具,调完又卡在 interrupt 等人拍板,状态还落盘能倒带,把工具、审批、持久化一次性串成真能用的产品。