LangGraph 06：工具调用

发表于 更新于 分类于
讲清工具的定义与挂载方式及选型，再用 bind_tools 与 ToolNode 给出最小闭环、并行调用和进阶写法。

前言

模型本身不会去查天气或调接口，它只在 AIMessage 里给出 tool_calls。
真正执行函数、把结果写成 ToolMessage 再喂回模型，是图运行时的工作。
标准「模型 + 工具」循环可直接用 LangChain 的 create_agent；需要自定义状态、并行分支或与 HITL 拼图时，才在 StateGraph 里手写绑定与路由。
本文先讲清闭环，再对比定义与挂载的几种方式及选型；默认可抄的最小图在后面，手写节点、重试包装、写入自定义状态放在进阶。
循环上限与节点 RetryPolicy 见 《LangGraph 07：循环重试与缓存》；工具内 interrupt 审批见 《LangGraph 11：HITL 人机协同》。
示例对接 火山方舟 Coding Plan，模型用 ark-code-latest。
下文需要 Python 3.12+，依赖用 uv 管理。

依赖

建议使用 Python 3.12 及以上。
用 uv 初始化工程并声明依赖（版本请按项目实际调整）。

1
2
3
4
uv init langgraph-tool-calling
cd langgraph-tool-calling
uv venv --python 3.12
uv add "langgraph>=1.0,<2.0" "langchain>=1.0,<2.0" langchain-openai python-dotenv rich

Pydantic 复杂 schema 示例一般已随 langchain 间接安装；缺失时再执行 uv add pydantic。

在项目根目录创建 .env，写入 Coding Plan 的 Key 与专用 Base URL。

1
2
OPENAI_API_KEY=你的火山方舟 API Key
OPENAI_BASE_URL=https://ark.cn-beijing.volces.com/api/coding/v3

请勿写成普通方舟 .../api/v3，以免无法抵扣 Coding Plan 额度。
后续示例用 uv run python xxx.py 运行，一般不必手动激活虚拟环境。

闭环

工具调用要拆成两步，不要混在一个节点里。

  1. 模型侧：bind_tools 只把工具 schema 交给模型，产出 tool_calls，此时函数尚未执行。
  2. 图侧：工具节点按名字执行，把每条结果写成带 tool_call_id 的 ToolMessage，再回到模型。

默认可抄的拓扑是 START → llm_node ⇄ tools → END。
tools_condition 看着最后一条消息：有 tool_calls 就进 tools，否则结束。

没有工具节点时，模型再怎么「调用」也不会跑到你的 Python 函数。
添加工具要分两层：怎么定义，以及 挂到哪套运行时。

添加方式

定义

本地工具最终都会变成带 schema 的可调用对象。
常见写法按控制力从低到高如下。

  1. 普通函数：写清类型注解和 docstring，create_agent(tools=[fn]) 可以直接收。
  2. @tool：本系列默认写法；可改名、改描述，并加 parse_docstring=True。
  3. args_schema：参数有嵌套、枚举或要给模型更细的 Field 说明时，用 Pydantic 模型。
  4. StructuredTool.from_function：函数签名不便改、或要从现成 callable 包装时用。
  5. BaseTool 子类：需要实例状态、延迟读环境变量、同步/异步两套实现时再上。
  6. 厂商内置工具：网页搜索、代码解释器等由服务端执行，通常以 dict 交给模型，不要再塞进 ToolNode。

默认用 @tool 即可。
下面把同一功能写成装饰器，并演示复杂参数怎么挂 schema。

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
from typing import Literal

from langchain.tools import tool
from pydantic import BaseModel, Field


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位",
    )


@tool(args_schema=WeatherInput)
def get_weather_ex(city: str, units: str = "celsius") -> str:
    """查询指定城市天气，可指定温度单位。"""
    temp = 22 if units == "celsius" else 72
    return f"{city} {temp}°{'C' if units == 'celsius' else 'F'}"

runtime、config 是保留参数名，不要当成业务入参。
要读状态或上下文，用 ToolRuntime，见后文「写入状态」。

挂载

定义好之后，还要告诉模型和执行器。
三套挂法覆盖绝大多数场景。

  1. create_agent：把 tools 传进去，框架同时负责 bind_tools 和执行循环。
  2. bind_tools + ToolNode：手写 StateGraph 时的默认可抄路径；两边传入同一份 tools 列表。
  3. 手写 tool_node：自己解析 tool_calls、构造 ToolMessage；图拓扑仍是模型节点 ⇄ 工具节点。

标准 Agent 循环用 create_agent 即可，不必再画图。

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
import os

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
)
agent = create_agent(model=model, tools=[get_weather])
agent.invoke({"messages": [{"role": "user", "content": "北京今天天气？"}]})

只 bind_tools 不配工具节点，模型会产出 tool_calls，函数不会执行。
只配 ToolNode 不 bind_tools，模型看不到 schema，通常不会去调。
厂商内置工具只 bind 给模型、由服务端跑，不要再放进 ToolNode。

选型

按问题选，不要先上最重的写法。

  1. 只要「模型 + 工具」循环：用 create_agent，工具优先 @tool。
  2. 要自定义 State、并行分支、与 HITL / 子图拼在一起：用手写图 + bind_tools + ToolNode。
  3. 要改 ToolMessage 形状或失败语义：再换手写 tool_node。
  4. 参数简单、本地函数：@tool；参数结构复杂：args_schema。
  5. 不能改原函数签名：StructuredTool.from_function；工具要带实例状态：BaseTool 子类。
  6. 搜索、跑代码等由模型厂商提供：走内置 / 服务端工具，不要包成本地函数再执行一遍。

后面示例按第 2 条展开：@tool + bind_tools + ToolNode。
create_agent 的完整用法见 《LangChain 03：1.x Agent 教程》。

示例

定义并绑定

用 @tool 声明函数，再用 bind_tools 交给模型。
parse_docstring=True 会从 docstring 解析参数说明，schema 更稳。

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
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
from rich import print as rprint

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


tools = [get_weather]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)

ai = model.invoke([HumanMessage("今天北京天气怎么样？")])
rprint(ai.tool_calls)
rprint(ai.content)

期望看到类似 [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': '...', 'type': 'tool_call'}]。
此时天气函数尚未执行，content 也常常为空，因为模型把回合交给了工具。

最小图

把绑定后的模型放进 llm_node，执行交给预构建 ToolNode。
节点名必须叫 tools，才能直接配 tools_condition，不必手写路由。

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
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


tools = [get_weather]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)


def llm_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}


builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tools", ToolNode(tools=tools))
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", tools_condition)
builder.add_edge("tools", "llm_node")
graph = builder.compile()

res = graph.invoke({"messages": [HumanMessage("今天北京天气怎么样？")]})
for msg in res["messages"]:
    msg.pretty_print()

跑完后消息列表里应有：用户问题、带 tool_calls 的模型消息、ToolMessage、最终自然语言回复。
若坚持把节点叫 tool_node，就要自己写 router，或给 add_conditional_edges 配 path_map；见后文「手写节点」。

并行调用

图拓扑与上一节完全相同，不必为每个工具单独建节点。
差别只在：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
53
54
55
56
57
58
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


@tool(parse_docstring=True)
def get_news(home_or_abroad: bool) -> str:
    """查询国内外新闻摘要。

    Args:
        home_or_abroad: True 表示国内，False 表示国外
    """
    return "国内要闻摘要" if home_or_abroad else "国际要闻摘要"


tools = [get_weather, get_news]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)


def llm_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}


builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tools", ToolNode(tools=tools))
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", tools_condition)
builder.add_edge("tools", "llm_node")
graph = builder.compile()

res = graph.invoke(
    {"messages": [HumanMessage("今天北京天气怎么样？国内有什么新闻？")]}
)
for msg in res["messages"]:
    msg.pretty_print()

同一条 AIMessage 可以带多条 tool_calls，ToolNode 会一次执行完并回写多条 ToolMessage。
模型是否合并调用取决于提示与模型本身；图结构已经允许并行。

到这里，默认可抄路径已经结束。
后面三节按需阅读：自定义消息、重试包装、把结果写入状态字段。

进阶

手写节点

需要完全自定义消息形状或失败语义时，再自己解析 tool_calls。
每条 ToolMessage 必须带上对应的 tool_call_id，否则下一轮模型会对不齐。

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
53
54
55
56
57
58
59
60
61
62
import os
from typing import Literal

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langgraph.graph import END, START, MessagesState, StateGraph

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    return f"{city} 天气晴朗，微风"


tools = [get_weather]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)


def llm_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}


def tool_node(state: MessagesState) -> MessagesState:
    last = state["messages"][-1]
    outs: list[ToolMessage] = []
    for call in last.tool_calls:
        result = tools_by_name[call["name"]].invoke(call["args"])
        outs.append(
            ToolMessage(name=call["name"], content=result, tool_call_id=call["id"])
        )
    return {"messages": outs}


def router(state: MessagesState) -> Literal["tool_node", "__end__"]:
    if state["messages"][-1].tool_calls:
        return "tool_node"
    return END


builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "llm_node")
graph = builder.compile()

graph.invoke({"messages": [HumanMessage("北京今天天气？")]})

手写循环灵活，但要自己处理并行调用、异常与字段，容易漏。
默认仍优先 ToolNode；循环上限与节点级重试见 《LangGraph 07：循环重试与缓存》。

调用包装

ToolNode(..., wrap_tool_call=fn) 在真正 execute(request) 前后插入逻辑。
适合有限次重试、统一打日志、对幂等工具做短时效缓存。

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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import os
import random
from dataclasses import dataclass

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()


@tool(parse_docstring=True)
def get_weather(city: str) -> str:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    if random.randint(1, 10) < 8:
        raise ConnectionError("网络波动")
    return f"{city} 天气晴朗，微风"


tools = [get_weather]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)


@dataclass
class UserContext:
    max_attempts: int


def llm_node(state: MessagesState) -> MessagesState:
    return {"messages": [model.invoke(state["messages"])]}


def wrap_tool_call(request, execute):
    max_attempts = request.runtime.context.max_attempts
    tool_msg = None
    for _ in range(max_attempts):
        try:
            tool_msg = execute(request)
            break
        except ConnectionError:
            continue
    if tool_msg is None:
        tool_msg = ToolMessage(
            tool_call_id=request.runtime.tool_call_id,
            content="调用次数达到上限，调用失败",
        )
    return tool_msg


builder = StateGraph(state_schema=MessagesState, context_schema=UserContext)
builder.add_node("llm_node", llm_node)
builder.add_node(
    "tools",
    ToolNode(tools=tools, wrap_tool_call=wrap_tool_call),
)
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", tools_condition)
builder.add_edge("tools", "llm_node")
graph = builder.compile()

graph.invoke(
    {"messages": [HumanMessage("今天北京天气怎么样？")]},
    context=UserContext(max_attempts=3),
)

图结构仍是最小图，只是给 ToolNode 多传了包装函数。
缓存版思路相同：用 (tool_name, 参数序列化) 做键，命中则直接构造 ToolMessage；不要缓存下单、转账等写操作。

写入状态

工具参数里声明 runtime: ToolRuntime，可读取图状态、上下文与当前 tool_call_id。
返回 Command(update=...) 时，可同时追加 ToolMessage 并改自定义字段。

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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import os

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, ToolRuntime, tools_condition
from langgraph.types import Command
from rich import print as rprint

load_dotenv()


class OverAllState(MessagesState):
    last_city: str


@tool(parse_docstring=True)
def get_weather(city: str, runtime: ToolRuntime) -> Command:
    """查询指定城市的当日天气。

    Args:
        city: 城市名称
    """
    weather = f"{city} 天气晴朗，微风"
    return Command(
        update={
            "last_city": city,
            "messages": [
                ToolMessage(
                    content=weather,
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )


tools = [get_weather]
model = init_chat_model(
    "openai:ark-code-latest",
    temperature=0,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ["OPENAI_BASE_URL"],
).bind_tools(tools)


def llm_node(state: OverAllState) -> OverAllState:
    return {"messages": [model.invoke(state["messages"])]}


builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("tools", ToolNode(tools=tools))
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", tools_condition)
builder.add_edge("tools", "llm_node")
graph = builder.compile()

res = graph.invoke(
    {"messages": [HumanMessage("今天北京天气怎么样？")], "last_city": ""}
)
rprint(res["last_city"])
for msg in res["messages"]:
    msg.pretty_print()

Command 里必须带上匹配的 tool_call_id，否则模型下一轮会对不齐工具结果。
只读状态或 runtime.context 时，工具仍可返回普通字符串。

验证

  1. 定义并绑定：单独 invoke 应打印出 get_weather 的 tool_calls，函数本身未执行。
  2. 最小图：最终消息里应有 ToolMessage 与一句自然语言天气回复。
  3. 并行调用：一次用户提问后，中间应出现两条工具结果。
  4. 调用包装：多次失败后仍可能成功，或打出「调用次数达到上限」。
  5. 写入状态：结束状态里 last_city 应为 北京。

总结

  1. 分工：bind_tools 让模型产出调用意图；工具节点负责执行并回写 ToolMessage。
  2. 添加：先选定义方式（默认 @tool），再选挂载（create_agent 或 bind_tools + ToolNode）。
  3. 选型：标准循环用 create_agent；要自定义图再用 ToolNode；只有要改消息形状时才手写节点。
  4. 默认可抄：节点名 tools + ToolNode + tools_condition 组成最小闭环。
  5. 并行与进阶：拓扑不变即可并行；重试用 wrap_tool_call；改字段用 ToolRuntime + Command。