代码仓库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 osfrom langchain.agents import create_agentfrom langchain.tools import toolfrom langchain_openai import ChatOpenAImodel = 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,从消息层面就能看出这次执行成没成。
sequenceDiagram
participant U as 用户
participant A as Agent 循环
participant M as 模型
participant T as ToolNode
U->>A: 北京天气怎么样?算一下 128*7
A->>M: 消息历史 + 工具 JSON Schema
M-->>A: AIMessage, tool_calls=[get_weather, calculator]
A->>T: 按 id 逐个执行
T-->>A: ToolMessage(tool_call_id=..., status=success)
A->>M: 历史 + 两条 ToolMessage
M-->>A: AIMessage 文本
A-->>U: 最终回答
装饰器接受的参数不少,常用的就这么几个:
参数
作用
例子
第一个位置参数
改工具名
@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} 天" @tool def style_annotated ( city: Annotated[str , "城市名,中文或拼音" ], days: Annotated[int , "要查未来几天,1 到 7" ] = 1 , ) -> str : """查天气。""" return f"{city} {days} 天" 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 正常返回。
flowchart TD
M[模型给出 args] --> P[Pydantic 按 args_schema 校验]
P -->|通过| F[执行函数体]
P -->|失败| V[ValidationError]
V --> TI[ToolNode 包成 ToolInvocationError]
TI --> TM[ToolMessage status=error]
TM --> M2[模型读到错误文本]
F --> OK[ToolMessage status=success]
OK --> M2
M2 --> M3[改正参数后重试]
这里有个前提,业务规则藏在 validator 里,描述里不写。反过来说,如果范围写进描述,模型通常就不会越界。我用 Field(gt=0, le=100) 加描述「0 到 100 之间」试了「打 -20% 折扣」和「打 150% 折扣」,模型一次都没调用工具,直接回复用户说超出范围。描述约束住了它。
两件事都要做:描述里写清范围,validator 里守住底线。描述降低出错概率,validator 保证出错时不会真的执行。
到这一步,工具还只是「输入参数、返回字符串」。真实场景里它常常需要知道当前是谁在调用、这轮对话进行到哪了、用户上次存过什么偏好。这些信息不走参数,走 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 下次再来还能读到。
flowchart LR
subgraph 创建时
CS[context_schema] --> CTX[context 对象]
SS[state_schema] --> ST[graph state]
CK[checkpointer] --> ST
SR[store] --> STR[BaseStore]
end
INV[agent.invoke 传入 context] --> CTX
TN[ToolNode 执行工具] --> RT[ToolRuntime]
CTX --> RT
ST --> RT
STR --> RT
RT --> R1[runtime.context 只读]
RT --> R2[runtime.state 当前会话可读写]
RT --> R3[runtime.store 跨会话]
RT --> R4[runtime.tool_call_id / config / stream_writer]
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 dataclassfrom langchain.agents import AgentState, create_agentfrom langgraph.checkpoint.memory import InMemorySaverfrom 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 ToolMessagefrom 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_callfrom langchain.messages import ToolMessagefrom 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 dataclassfrom 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,也没尝试调用它,直接说明自己没有删除能力。权限判断放在中间件里,模型无法绕过。
flowchart LR
C[runtime.context.user_role] --> MW[wrap_model_call 中间件]
MW -->|viewer| T1[read_data]
MW -->|editor| T2[read_data + write_data]
MW -->|admin| T3[read_data + write_data + delete_data]
T1 --> M[模型]
T2 --> M
T3 --> M
如果工具本身是运行时才知道的,比如从 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。
前面都在用 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 就够了。