代码仓库ChainReaction

模型只会生成文字。让它查天气、下单、读数据库的,是你挂在外面的一段 Python。LangChain 把这段 Python 和它的说明书一起打包成 Tool。说明书是一份 JSON Schema,模型看得到;函数体模型看不到。

这个分界决定了日常调试的方向。模型不调工具,问题多半出在描述写得含糊;模型调了但参数填错,问题在 Schema 的类型或约束;函数执行到一半炸了,模型那边只能看到你回给它的那段错误文字,它凭这段文字决定下一步。工具写得好不好,一半在函数里,一半在函数外。

这篇按顺序讲这些:工具的最小结构、@tool 能改的几件事、参数 Schema 的三种写法、校验失败后模型收到什么、ToolRuntime 从哪里取数据、工具怎么读写 agent state、异常怎么变成模型能读的消息、按上下文动态换工具集,最后用 bind_tools 直接看模型吐出来的 tool_calls。文中输出都是在 DeepSeek 上实跑的终端原文。

工具的最小结构

官方对工具的定义是两样东西的组合:一份包含名称、描述、参数定义的 Schema,一个用来执行的函数或协程。模型基于对话上下文决定调不调、传什么参数。绑定用 bind_tools,之后每次调用模型都可能挑其中一个来用。

写工具最省事的方式是 @tool 装饰器。函数名变成工具名,docstring 变成工具描述,类型注解变成参数 Schema:

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

@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气。"""
return f"{city}:晴,22 摄氏度"

模型实际拿到的是这个:

1
2
3
4
5
6
7
8
9
{
"description": "查询指定城市的当前天气。",
"properties": {
"city": { "title": "City", "type": "string" }
},
"required": ["city"],
"title": "get_weather",
"type": "object"
}

这份 JSON 来自 get_weather.tool_call_schema.model_json_schema()。绑定到模型之后,它会被翻译成对应 provider 的 function 定义,和你的函数体没有关系。函数体只在 ToolNode 里被执行。

类型注解是必需的。@tool 靠它推断 Schema,缺了注解的参数会被当成无类型。参数名里有两个是保留字,config 和 runtime,用它们当业务参数会在运行时出错,需要运行时信息就用 ToolRuntime。

docstring 的处理方式值得单独说。默认 parse_docstring=False,整个 docstring 原封不动当描述,包括里面的 Args: 段。拿官方那个 search_database 例子跑一下:

1
description = 'Search the customer database for records matching the query.\n\n    Args:\n        query: Search terms to look for\n        limit: Maximum number of results to return'

整段 Args: 连同缩进都塞进了 description,参数本身没有描述。加上 parse_docstring=True 才会拆开:

1
2
3
4
description = 'Search the customer database for records matching the query.'
properties:
query: "description": "Search terms to look for"
limit: "description": "Maximum number of results to return"

两种都行,但别混着用。要么全程 Annotated 或 Pydantic 写参数描述,要么明确打开 parse_docstring。我见过的最常见问题是描述里带一长串 Args: 文本,参数描述却是空的,模型只能靠参数名猜。

一次工具调用里发生了什么

先把整条链路跑通。三个工具,一个是普通查询,一个是算数,第三个故意抛错:

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
import os
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
temperature=0.1,
max_tokens=1000,
)

@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气。"""
fake = {"beijing": "晴,22 摄氏度", "shanghai": "小雨,19 摄氏度"}
return fake.get(city.lower(), f"{city}:暂无数据")

@tool
def calculator(expression: str) -> str:
"""计算一个纯算术表达式,例如 '12*7+3'。只支持加减乘除和括号。"""
allowed = set("0123456789+-*/(). ")
if not set(expression) <= allowed:
raise ValueError(f"表达式包含不支持的字符: {expression!r}")
return str(eval(expression))

@tool
def fetch_inventory(sku: str) -> str:
"""按 SKU 编号查询仓库库存。"""
stock = {"A100": 12, "B200": 0}
if sku not in stock:
raise ValueError(f"SKU {sku!r} 不在库存表里,可用的 SKU 只有 A100 和 B200")
return f"{sku} 剩余 {stock[sku]} 件"

agent = create_agent(
model,
tools=[get_weather, calculator, fetch_inventory],
system_prompt="你是一个助手。需要数据时必须调用工具,不要凭猜测回答。",
)

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气怎么样?顺便帮我算一下 128*7"}]})
for m in result["messages"]:
print(type(m).__name__, m.content, getattr(m, "tool_calls", None))

完整脚本在 Tools/BasicTools.py,打印了每条消息的类型、tool_calls、tool_call_id 和 status。正常调用的输出:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
--- [0] HumanMessage ---
content='北京天气怎么样?顺便帮我算一下 128*7'
--- [1] AIMessage ---
tool_call name=get_weather args={'city': '北京'} id=call_00_ueaxfcBjKjIo77aw7fvi1211
tool_call name=calculator args={'expression': '128*7'} id=call_01_7rNqtGbTGwjQnNhlMSug2142
content="I'll check both for you."
--- [2] ToolMessage ---
tool_call_id=call_00_ueaxfcBjKjIo77aw7fvi1211 status=success name=get_weather
content='北京:暂无数据'
--- [3] ToolMessage ---
tool_call_id=call_01_7rNqtGbTGwjQnNhlMSug2142 status=success name=calculator
content='896'
--- [4] AIMessage ---
content='结果如下:... 128 × 7 = 896 ...'

注意几点。模型一次返回了两个 tool_call,两个工具被并行调用,顺序不保证。每条 ToolMessage 都带一个 tool_call_id,和 AIMessage 里的 id 一一对应,配对错了 provider 直接拒请求。status 字段是 success 还是 error,从消息层面就能看出这次执行成没成。

@tool 装饰器能改什么

装饰器接受的参数不少,常用的就这么几个:

参数 作用 例子
第一个位置参数 改工具名 @tool("web_search")
description= 覆盖 docstring 生成的描述 @tool("calc", description="做算术,任何数学题都用它")
args_schema= 换成 Pydantic 模型或 JSON Schema @tool(args_schema=OrderInput)
return_direct= 工具执行完直接结束循环 @tool(return_direct=True)
parse_docstring= 解析 docstring 的 Args 段 @tool(parse_docstring=True)
response_format= 内容与 artifact 分开返回 response_format="content_and_artifact"
async def 函数写成协程即可,装饰器不用改 见下

改名和改描述是最常用的两件事。工具名默认取函数名,函数名叫 f1 的时候模型看不出它是干嘛的。官方建议工具名用 snake_case,只用字母、数字、下划线和连字符。

异步工具不需要额外开关,把函数写成 async def,用 agent.ainvoke 调就行:

1
2
3
4
5
@tool
async def fetch_price(sku: str) -> str:
"""异步查询商品价格。"""
await asyncio.sleep(0.1)
return f"{sku} 价格 199 元"
1
2
3
4
--- [0] HumanMessage: 'A100 多少钱?'
--- [1] AIMessage: "I'll look up the price for you."
--- [2] ToolMessage: 'A100 价格 199 元'
--- [3] AIMessage: 'A100 的价格是 **199 元**。'

return_direct=True 会短路整个循环。工具执行完,结果直接当最终回答返回,不再经过模型:

1
2
3
4
--- [0] HumanMessage: '订单 #12345 什么状态?'
--- [1] AIMessage: "I'll look up the status of order #12345 for you."
--- [2] ToolMessage: '订单 12345 已发货,预计 2 天后送达。'
消息总数: 3

对比上一个例子的 4 条消息,这里少了最后那条 AIMessage。模型没有机会改写、总结、补充任何东西,输出是什么就返回什么。适合查订单、查物流这类结果拿来即用的场景。

有个细节容易踩。模型一步里并行调了多个工具时,只有当这一步所有工具都是 return_direct=True,agent 才会结束循环。只要混进一个普通工具,整批 ToolMessage 都会回到模型那里重新推理。别以为给一个工具加了 return_direct 就一定能省掉那次模型调用。

参数 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 Annotated, Literal
from pydantic import BaseModel, Field

# 写法一:纯类型注解
@tool
def style_typehint(city: str, days: int = 1) -> str:
"""查天气。"""
return f"{city} {days} 天"

# 写法二:Annotated 带描述
@tool
def style_annotated(
city: Annotated[str, "城市名,中文或拼音"],
days: Annotated[int, "要查未来几天,1 到 7"] = 1,
) -> str:
"""查天气。"""
return f"{city} {days} 天"

# 写法三:Pydantic 模型
class OrderInput(BaseModel):
"""创建订单的入参。"""
sku: str = Field(description="商品 SKU,只能是 A100 或 B200")
quantity: int = Field(gt=0, le=99, description="购买数量,1 到 99")
channel: Literal["web", "app"] = Field(default="web", description="下单渠道")

@tool(args_schema=OrderInput)
def create_order(sku: str, quantity: int, channel: str = "web") -> str:
"""创建一个订单。"""
return f"已下单 {sku} x{quantity}(渠道 {channel})"

三者生成的 Schema 对比(Tools/PydanticSchema.py 里的实跑输出,删掉了 title 之类的样板字段):

写法 参数描述来自哪 能不能表达约束 适合什么
类型注解 没有,模型只看到类型和参数名 不能 一两个参数、语义自明
Annotated 注解的第二个参数 不能,除非用 Field 参数少但需要一句解释
Pydantic 模型 Field(description=...) 能,gt/le、Literal、自定义 validator 都行 参数多、有范围或枚举、有跨字段规则
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
--- style_typehint ---
"properties": {
"city": { "title": "City", "type": "string" },
"days": { "default": 1, "title": "Days", "type": "integer" }
}
--- style_annotated ---
"properties": {
"city": { "description": "城市名,中文或拼音", "title": "City", "type": "string" },
"days": { "default": 1, "description": "要查未来几天,1 到 7", "title": "Days", "type": "integer" }
}
--- create_order ---
"properties": {
"sku": { "description": "商品 SKU,只能是 A100 或 B200", "type": "string" },
"quantity": { "description": "购买数量,1 到 99", "exclusiveMinimum": 0, "maximum": 99, "type": "integer" },
"channel": { "default": "web", "description": "下单渠道", "enum": ["web", "app"], "type": "string" }
},
"required": ["sku", "quantity"]

gt=0 变成了 exclusiveMinimum: 0,Literal 变成了 enum。这些约束不只是给模型看的提示,同时也是真正会执行的校验。

我平时的选择:参数只有一两个且名字自明,用类型注解;参数需要解释,用 Annotated;出现范围、枚举、或者多个参数互相牵制,直接上 Pydantic。Annotated 写到第三个参数就该换成 Pydantic 了。

校验失败时模型看到什么

Pydantic 模型不只过滤,它是真的会拦。直接给工具喂错类型:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
输入 {'sku': 'A100', 'quantity': 'abc'}
异常类型: ValidationError
异常消息: 1 validation error for OrderInput / quantity / Input should be a valid integer,
unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]

输入 {'sku': 'A100', 'quantity': 0}
异常类型: ValidationError
异常消息: 1 validation error for OrderInput / quantity / Input should be greater than 0
[type=greater_than, input_value=0, input_type=int]

输入 {'sku': 'A100', 'quantity': 3, 'channel': 'fax'}
异常类型: ValidationError
异常消息: 2 validation errors for OrderInput / quantity / Value error, 3 不是整箱数量...
/ channel / Input should be 'web' or 'app' [type=literal_error]

这些是 pydantic 的原始报错,带英文类型标签和一串文档链接。直接在 agent 里跑,模型收到的不是这坨东西。ToolNode 会把 ValidationError 包一层,用更短的模板重新排版:

1
2
3
4
5
6
7
8
9
10
11
12
13
--- [1] AIMessage ---
tool_call name=create_order args={'sku': 'A100', 'quantity': 10} id=call_00_mejoEFCMJNH6jugoEueB2742
--- [2] ToolMessage ---
tool_call_id=call_00_mejoEFCMJNH6jugoEueB2742 status=error name=create_order
content="Error invoking tool 'create_order' with kwargs {'sku': 'A100', 'quantity': 10} with error:
quantity: Value error, 10 不是整箱数量,这个 SKU 只能整箱卖,数量必须是 6 的倍数
Please fix the error and try again."
--- [3] AIMessage ---
tool_call name=create_order args={'sku': 'A100', 'quantity': 12} id=call_00_rhXqOEgZIbN1HrukNvQN1581
content='A100 只能整箱卖(6 的倍数),10 不符合。我改成 12 个(最接近 10 的整箱数量)再试一次。'
--- [4] ToolMessage ---
tool_call_id=call_00_rhXqOEgZIbN1HrukNvQN1581 status=success name=create_order
content='已下单 A100 x12(渠道 web)'

这个例子里,quantity 上挂了一个 validator,要求数量必须是 6 的倍数。描述里只写了「1 到 99」,模型不知道整箱规则,于是传了 10,被拦下,读到错误文本,改成 12,第二次成功。整条链路没有抛异常,agent.invoke 正常返回。

这里有个前提,业务规则藏在 validator 里,描述里不写。反过来说,如果范围写进描述,模型通常就不会越界。我用 Field(gt=0, le=100) 加描述「0 到 100 之间」试了「打 -20% 折扣」和「打 150% 折扣」,模型一次都没调用工具,直接回复用户说超出范围。描述约束住了它。

两件事都要做:描述里写清范围,validator 里守住底线。描述降低出错概率,validator 保证出错时不会真的执行。

ToolRuntime:工具能从哪拿数据

到这一步,工具还只是「输入参数、返回字符串」。真实场景里它常常需要知道当前是谁在调用、这轮对话进行到哪了、用户上次存过什么偏好。这些信息不走参数,走 ToolRuntime。

在函数签名里加一个 runtime: ToolRuntime,ToolNode 在执行时注入实例。这个参数不会出现在给模型的 Schema 里,实测:

1
2
3
4
whoami -> []
remember_preference -> ['key', 'value']
recall_preference -> ['key']
set_user_name -> ['new_name']

whoami 只有一个 runtime 参数,模型看到的参数列表是空的。模型不知道 runtime 的存在,也无法伪造里面的值。

ToolRuntime 上挂着这些:

属性 内容 生命周期
runtime.state 当前会话的 graph state,含 messages 和自定义字段 一次对话
runtime.context 调用时传入的不可变配置,比如 user_id、租户、角色 单次 invoke
runtime.store BaseStore 长期记忆,namespace + key 存取 跨对话
runtime.tool_call_id 当前工具调用的 id 单次调用
runtime.stream_writer 往流里写实时进度 单次调用
runtime.execution_info thread_id、run_id、重试次数 单次执行
runtime.server_info LangGraph Server 上的 assistant、graph、用户信息,本地跑是 None 部署环境
runtime.config 本次执行的 RunnableConfig 单次执行

三个数据来源要分清。context 是你在 invoke 时传进去的,一次调用一份,工具只读不写;state 是当前会话的,可以读也可以改;store 活在会话之外,同一个 user_id 下次再来还能读到。

Tools/ToolRuntime.py 里四个工具分别读 context、写 store、读 store、用 Command 改 state。第一次调用的输出(工具内部的打印):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
==================== 模型看到的 schema 里没有 runtime ====================
whoami -> []
remember_preference -> ['key', 'value']
recall_preference -> ['key']
set_user_name -> ['new_name']
==================== 第一次调用 ====================
[whoami] type(runtime.context) = UserContext
[whoami] runtime.context = UserContext(user_id='user123', tenant='acme', plan='premium')
[whoami] runtime.context.user_id = user123
[whoami] runtime.context.tenant = acme
[whoami] runtime.context.plan = premium
[whoami] runtime.state['user_name'] = '未知'
[whoami] runtime.state['ticket_count'] = 0
[whoami] len(runtime.state['messages']) = 2
[whoami] runtime.tool_call_id = call_00_NYCkxPe9KtBNq6357Fh55147
[whoami] runtime.store is None = False
[remember] store.put namespace=('preferences', 'user123') key=theme value=dark
[set_user_name] Command update user_name='张三' ticket_count=1
[msg 2] ToolMessage name=whoami status=success content='user123 / acme / premium / 姓名=未知'
[msg 3] ToolMessage name=remember_preference status=success content='已记住 theme=dark'
[msg 4] ToolMessage name=set_user_name status=success content='已把用户姓名记为 张三,工单计数 1。'
state: user_name='张三' ticket_count=1

runtime.context 拿到的就是 invoke 时传进去的那个 dataclass 实例,类型和字段都对得上。runtime.state 是一个 dict,messages 里已经有两条消息,说明模型这条 AIMessage 也计入了。runtime.store 不是 None,因为 create_agent 时传了 store=InMemoryStore()。

对应的 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
29
30
from dataclasses import dataclass
from langchain.agents import AgentState, create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

@dataclass
class UserContext:
user_id: str
tenant: str
plan: str

class SupportState(AgentState):
user_name: str
ticket_count: int

agent = create_agent(
model,
tools=[whoami, remember_preference, recall_preference, set_user_name],
context_schema=UserContext,
state_schema=SupportState,
checkpointer=InMemorySaver(),
store=store,
system_prompt="你是客服助手。需要用户信息或记忆时调用工具,不要编造。",
)

agent.invoke(
{"messages": [...], "user_name": "未知", "ticket_count": 0},
config={"configurable": {"thread_id": "thread-1"}},
context=UserContext(user_id="user123", tenant="acme", plan="premium"),
)

thread_id 和 context 是两条独立的轴。thread_id 配合 checkpointer 决定「这轮对话接着哪段历史」,context 决定「这次调用以什么身份执行」。同一个 thread 换 context 是合法的,多租户场景里常见。

第二次调用用了同一个 thread 和同一个 context,工具从 store 里读回了偏好:

1
2
3
4
[recall] store.get namespace=('preferences', 'user123') key=theme
-> Item(namespace=['preferences', 'user123'], key='theme', value={'value': 'dark'}, ...)
[msg 8] ToolMessage name=recall_preference status=success content='theme=dark'
最终回答: 你上次让我记的 theme 是 **dark**(深色主题)。

第三次换了 user_id="user999",namespace 变了,读不到:

1
2
3
[recall] store.get namespace=('preferences', 'user999') key=theme -> None
[msg 2] ToolMessage name=recall_preference status=success content='没有记录 theme'
最终回答: 我查了一下长期记忆,里面没有关于 theme 的记录...

namespace 用 ("preferences", user_id) 而不是单一的 ("preferences",),隔离是自动的。所有用户共用一个大 namespace 再靠 key 前缀区分,早晚会串。

用 Command 改 agent state

工具想改 state,返回 Command 而不是普通值:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain.messages import ToolMessage
from langgraph.types import Command

@tool
def set_user_name(new_name: str, runtime: ToolRuntime[None, SupportState]) -> Command:
"""把用户姓名写进对话状态,并给工单计数加一。"""
new_count = (runtime.state.get("ticket_count") or 0) + 1
return Command(
update={
"user_name": new_name,
"ticket_count": new_count,
"messages": [
ToolMessage(
content=f"已把用户姓名记为 {new_name},工单计数 {new_count}。",
tool_call_id=runtime.tool_call_id,
)
],
}
)

两条规则不能省。第一,messages 里必须补一条 ToolMessage,tool_call_id 用 runtime.tool_call_id。模型发出的每个 tool_call 都必须在历史里有对应的 ToolMessage,少一条整个请求就不合法。第二,并行工具可能同时改同一个字段,涉及计数字段时按 LangGraph 的 reducer 规则处理冲突。

跑完第一次调用,state 里 user_name 变成了 '张三',ticket_count 从 0 变成 1,都在最终返回值里能读到。这一步比返回字符串多花不了几行,换来的是工具之间可以靠 state 传递信息。

工具内部报错怎么处理

前面说过,参数校验失败会被自动兜住。函数体里抛的异常是另一回事。同样三个工具,不加任何处理直接让模型调那个会抛错的:

1
2
3
==================== B. 工具抛错且未处理 ====================
抛出异常类型: ValueError
异常内容: SKU 'C300' 不在库存表里,可用的 SKU 只有 A100 和 B200

agent.invoke 直接崩了。ToolNode 的默认错误处理只认参数校验那一类,其他异常原样往上抛。模型根本没机会看到这个错误。

官方推荐的处理方式是中间件 wrap_tool_call,把异常转成 ToolMessage:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from collections.abc import Callable
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest

@wrap_tool_call
def handle_tool_errors(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
"""把工具异常转成模型能读的 ToolMessage。"""
try:
return handler(request)
except Exception as e:
return ToolMessage(
content=f"工具执行失败:{e}。请换一个参数重试,或直接告诉用户查不到。",
tool_call_id=request.tool_call["id"],
name=request.tool_call["name"],
status="error",
)

同一个问题再问一次:

1
2
3
4
5
6
7
8
==================== C. 用 wrap_tool_call 兜住错误 ====================
--- [1] AIMessage ---
tool_call name=fetch_inventory args={'sku': 'C300'} id=call_00_0B8eEz7mgEAT5IiZvKad8381
--- [2] ToolMessage ---
tool_call_id=call_00_0B8eEz7mgEAT5IiZvKad8381 status=error name=fetch_inventory
content="工具执行失败:SKU 'C300' 不在库存表里,可用的 SKU 只有 A100 和 B200。请换一个参数重试,或直接告诉用户查不到。"
--- [3] AIMessage ---
content='查不到 C300 的库存。库存表里没有这个 SKU,目前只有 A100 和 B200 两个 SKU 可查。...'

agent 正常结束,模型读到错误,转而把可用选项告诉了用户。

写错误消息时,把「怎么改」写进去。我上面那句「请换一个参数重试,或直接告诉用户查不到」是给模型的指令,模型基本会照做。只写 str(e) 也行,但模型更容易反复重试同一个错参数,白白烧几轮 token。原始异常信息别丢,调试时还要看。

另外,wrap_tool_call 只在异常真的冒到 ToolNode 外层时才触发。参数校验错误在 ToolNode 内部就被转成了 ToolMessage,中间件拿到的已经是正常返回值,不会走到 except。这一点我一开始搞错了,以为中间件能统一处理所有错误。实际是两套机制:校验错误自动处理,业务异常靠中间件。

按上下文动态选工具

工具多了模型会挑错,或者干脆不用。官方那句话说得很直白:工具太多会让模型不堪重负、错误变多,工具太少能力受限。按用户权限、功能开关、对话阶段动态调整工具集是常规做法。

已经注册的工具,用 wrap_model_call 中间件在每次模型调用前过滤:

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
from dataclasses import dataclass
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call

@dataclass
class UserContext:
user_role: str

@wrap_model_call
def context_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""按 runtime.context 里的角色决定这次给模型哪些工具。"""
if request.runtime is None or request.runtime.context is None:
user_role = "viewer"
else:
user_role = request.runtime.context.user_role

if user_role == "admin":
tools = list(request.tools)
elif user_role == "editor":
tools = [t for t in request.tools if t.name != "delete_data"]
else:
tools = [t for t in request.tools if t.name.startswith("read_")]

return handler(request.override(tools=tools))

注册了 read_data、write_data、delete_data 三个工具,三种角色各问一次「把 orders 表删掉,然后读一下 users 表」,中间件打印的实际绑定:

1
2
3
4
5
6
7
8
9
==================== role = viewer ====================
[middleware] role=viewer -> 实际绑定给模型的工具 = ['read_data']
[msg 1] 模型请求调用 read_data args={'table': 'users'}
[msg 2] ToolMessage read_data -> 'users 的 3 行数据'
最终回答: users 表已读取... 关于删除 orders 表:我无法执行。当前可用的工具只有 read_data(只读)...
==================== role = editor ====================
[middleware] role=editor -> 实际绑定给模型的工具 = ['read_data', 'write_data']
==================== role = admin ====================
[middleware] role=admin -> 实际绑定给模型的工具 = ['read_data', 'write_data', 'delete_data']

viewer 那次,模型看不到 delete_data,也没尝试调用它,直接说明自己没有删除能力。权限判断放在中间件里,模型无法绕过。

如果工具本身是运行时才知道的,比如从 MCP server 拉回来的,光过滤不够。这种要同时用 wrap_model_call 把工具加进 request,再用 wrap_tool_call 处理它的执行,两个钩子缺一不可。

几个容易踩的地方

工具重名不会报错,后注册的赢。两个工具都叫 search,注册进同一个 agent,我原以为会冲突报错,实测是静默覆盖:

1
2
search_a.name = search | search_b.name = search
ToolMessage search -> '[B] LangChain'

执行的是后注册的 search_b。search_a 从此不可达,没有任何提示。工具名冲突要么在写的时候就避开,要么在启动时自己查一遍重名。

工具名带空格或大写,DeepSeek 直接 400。官方警告过这一点,我试了 @tool("Web Search"):

1
2
3
4
web_search.name = 'Web Search'
报错: OpenAIInvalidRequestError
Error code: 400 - {'error': {'message': "Invalid 'tools[0].function.name': string does not
match pattern. Expected a string that matches the pattern '^[a-zA-Z0-9_-]+$'..."}}

错误在绑定后第一次调用时才出现,报错信息指向 tools[0].function.name,不太好定位到具体是哪个工具。命名就守 [a-zA-Z0-9_-] 这个范围,最省事。

返回大对象会原样塞进上下文。工具返回 dict 时,内容会被序列化成字符串放进 ToolMessage:

1
2
3
4
tool.invoke 直接调用返回类型: ToolMessage
ToolMessage.content 类型: str
ToolMessage.content 长度: 44708 字符
开头 160 字符: {"total": 500, "orders": [{"id": "ORD0000", "amount": 0.0, "status": "shipped", ...

500 条订单变成了 44708 个字符,全部进模型上下文。多来几次这种调用,窗口就满了。工具返回之前先裁,只给模型需要的字段和条数;完整数据放 artifact,那个字段不会发给模型,但程序里还能读到。

config 和 runtime 是保留参数名。拿它们当业务参数会在运行时出错,得换名字。需要运行时信息就用 ToolRuntime。

用 bind_tools 直接看 tool_calls

前面都在用 create_agent,它把循环藏起来了。想看清循环本身,绕开 agent,直接绑定工具调模型:

1
2
3
4
5
tools = [get_weather, calculator]
model_with_tools = model.bind_tools(tools)

response = model_with_tools.invoke("北京和上海天气怎么样?顺便算一下 128*7")
print(response.tool_calls)
1
2
3
4
5
6
7
8
9
返回类型: AIMessage
content: "I'll check the weather for both cities and do the calculation."
tool_calls:
[
{ "name": "get_weather", "args": { "city": "北京" }, "id": "call_00_Qa0Vc5oCopOxrrsasr0w2998", "type": "tool_call" },
{ "name": "get_weather", "args": { "city": "上海" }, "id": "call_01_BN1vQMKxUmTPbpcPNItq4821", "type": "tool_call" },
{ "name": "calculator", "args": { "expression": "128*7" }, "id": "call_02_BqUNX7iOGSSStbmIIfrt7773", "type": "tool_call" }
]
usage: {'input_tokens': 333, 'output_tokens': 108, 'total_tokens': 441, ...}

tool_calls 就是一个列表,每项有 name、args、id。模型只负责生成这个列表,执行是下一步的事。把工具对象直接 invoke 这个 dict,它返回的就已经是 ToolMessage:

1
2
tool.invoke 返回: ToolMessage | name = get_weather | tool_call_id = call_00_6vFTRZJqsSUiLDy7AjTL9661
| status = success | content = '北京:暂无数据'

手写一遍循环,就是把这几步串起来:

1
2
3
4
5
6
7
messages = [{"role": "user", "content": "北京天气怎么样?"}]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)
for tool_call in ai_msg.tool_calls:
messages.append(tools_by_name[tool_call["name"]].invoke(tool_call))
final = model_with_tools.invoke(messages)
print(final.text)

create_agent 内部跑的就是这个循环,多了几层东西:路由判断、并行执行、状态合并、错误处理、中间件。模型不需要工具时,tool_calls 是空列表,这次调用直接产出最终文本。想强制它调,用 tool_choice:

1
2
3
4
5
model.bind_tools(tools, tool_choice="any").invoke("今天天气不错")
tool_calls = [{"name": "get_weather", "args": {"city": "北京"}, ...}]

model.bind_tools(tools, tool_choice="calculator").invoke("你好")
tool_calls = [{"name": "calculator", "args": {"expression": "1+1"}, ...}]

tool_choice="any" 允许模型任选一个,tool_choice="calculator" 指定必须用它。第二次调用里模型把「你好」硬凑成了 1+1,这种强制用法只在确实需要时用,平时别动。

小结

  • 工具是「JSON Schema + 可执行函数」。模型只看得到 Schema,所以 docstring 和参数描述就是全部的沟通渠道。默认整个 docstring 都进描述,包括 Args: 段,要么打开 parse_docstring=True,要么用 Annotated / Pydantic 写参数描述。
  • 参数 Schema 三种写法按复杂度递进:类型注解、Annotated、Pydantic。有范围、枚举、业务规则就用 Pydantic。
  • 参数校验失败不会炸掉 agent。ToolNode 把 ValidationError 转成 status="error" 的 ToolMessage 回给模型,模型会自己改正。函数体里抛的异常默认会中断整个 invoke,要用 wrap_tool_call 中间件转成 ToolMessage。
  • ToolRuntime 不占 Schema。context 是一次调用的只读配置,state 是当前会话,store 跨会话。改 state 要返回 Command,并补上带 runtime.tool_call_id 的 ToolMessage。
  • 工具集可以按 runtime.context 动态过滤,权限判断放在 wrap_model_call 里,模型绕不过去。
  • 工具名只用字母、数字、下划线和连字符,不要重名,返回结果先裁再给模型。想看清 agent 内部,用 bind_tools 打印 tool_calls 就够了。