工具调用闭环:模型开工单(tool_calls)、你干活、回执(ToolMessage)贴回去
Day13 把工具造好了,今天让模型真正用起来。关键认知:模型自己不执行任何代码——它只会在回复里"下单"。完整闭环四步:①bind_tools 把工具说明书随请求发给模型;②模型回一条带 tool_calls 的 AIMessage(工单:调哪个、参数是什么、工单号 id);③你在本地执行工具;④把结果装进 ToolMessage(带上同一个工单号)追加回消息列表,再问一次模型。今天逐环读源码,Day15 的 Agent 就是把这个循环自动化。
ToolMessage.tool_call_id 必须和 ToolCall.id 严格对上,这是闭环的"对账"机制。痛点:说明书怎么发?结果怎么还?
StructuredTool。但三个问题没答案:①模型是通过 HTTP API 调的,怎么让它"看见"我这件工具?(不能把 Python 对象塞进请求体)②模型决定要用工具时,它的"意图"以什么格式回来?各家 API 格式还不一样(OpenAI 的 function.arguments 是 JSON 字符串,Anthropic 是嵌套 dict)。③我执行完了,结果以什么身份塞回对话,模型才知道"这是刚才那个调用的结果"而不是用户新说的话?bind_tools(把任意形态的工具统一转成该厂商的 JSON 格式挂到请求上);收件用 AIMessage.tool_calls(把各家五花八门的返回统一解析成标准 ToolCall:name/args/id);回件用 ToolMessage(带 tool_call_id 的标准回执)。你的代码只面对这三个标准件,换模型厂商,闭环代码一行不改。bind_tools:把说明书翻译成厂商格式,挂在请求上
基类只定了接口(libs/core/langchain_core/language_models/chat_models.py:2338):
# libs/core/langchain_core/language_models/chat_models.py:2338
def bind_tools(
self,
tools: Sequence[dict[str, Any] | type | Callable[..., Any] | BaseTool], # 四种形态都收
*,
tool_choice: str | None = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]:
"""Bind tools to the model."""
raise NotImplementedError # 具体翻译格式,各厂商包自己实现
看真实现——OpenAI 版(libs/partners/openai/langchain_openai/chat_models/base.py:2157):
# libs/partners/openai/langchain_openai/chat_models/base.py:2157(裁剪)
def bind_tools(self, tools, *, tool_choice=None, strict=None, **kwargs):
formatted_tools = [
convert_to_openai_tool(tool, strict=strict) for tool in tools # :2203 ★统一翻译
]
...
if tool_choice: # "auto"/"any"/指定工具名/none
...
return super().bind(tools=formatted_tools, **kwargs) # :2263 ★挂参数,返回新 Runnable
tools 四种形态BaseTool、普通函数、Pydantic 类、现成的 dict schema——都行。convert_to_openai_tool(core/utils/function_calling.py:517)统一翻译成 OpenAI 的 {"type": "function", "function": {"name", "description", "parameters"}} 格式。Day13 的说明书(name/description/args_schema)在这里被序列化成 JSON Schema。super().bind(...)★注意它不发请求、不改模型!bind 是 Day03 学过的 Runnable 机制:返回一个"预先绑定了默认参数"的新 Runnable。以后每次 invoke,这份 tools JSON 都自动附在 API 请求里。模型每一次都是现场读说明书,没有任何"记住工具"这回事。tool_choice调度选项:"auto"(模型自己决定用不用)、"any"/"required"(必须用一个)、指定名字(必须用这个)。Day15 的结构化输出会用 "any" 强制模型调工具。model_with_tools = model.bind_tools([get_weather]) 读作:"复印一份模型的使用配置,在里面夹上工具说明书。"原来的 model 没被改动,model_with_tools 是个新积木——它仍然是标准 Runnable,能进任何链。AIMessage.tool_calls:模型开出的标准工单
模型决定用工具时,回复的 AIMessage(libs/core/langchain_core/messages/ai.py:160)里 tool_calls 字段非空:
# libs/core/langchain_core/messages/ai.py:160
class AIMessage(BaseMessage):
"""Message from an AI. ..."""
tool_calls: list[ToolCall] = Field(default_factory=list) # :170 ★工单列表
invalid_tool_calls: list[InvalidToolCall] = Field(default_factory=list) # :173 解析失败的工单
usage_metadata: UsageMetadata | None = None # :176 token 用量
type: Literal["ai"] = "ai"
每张工单是个 TypedDict——ToolCall(libs/core/langchain_core/messages/tool.py:206):
# libs/core/langchain_core/messages/tool.py:206
class ToolCall(TypedDict):
"""Represents an AI's request to call a tool.
Example:
{"name": "foo", "args": {"a": 1}, "id": "123"}
"""
name: str # 调哪件工具(Day13 的品名)
args: dict[str, Any] # ★参数:已解析成标准 dict(不管厂商原始格式是啥)
id: str | None # ★工单号:回执对账全靠它
type: NotRequired[Literal["tool_call"]]
args 是 dict 不是字符串OpenAI 原始返回里 arguments 是一个 JSON 字符串,LangChain 在解析阶段(messages/tool.py:350 附近的解析函数)就帮你 json.loads 好了。你拿到的永远是能直接用的 dict——跨厂商统一。id 的注释原话:232 附近写得很直白:多个并发工具调用时,需要 id 来把"调用请求"和"调用结果"对上号。这就是快递面单号。invalid_tool_calls模型偶尔会生成残缺 JSON(截断、格式错)。解析失败的不会静默丢弃,而是进 invalid_tool_calls 留着——你可以选择报错、重试或让模型自己修。诚实暴露脏数据,而不是假装没发生。tool_calls 非空时 content 常为空模型这轮回复可能一句话不说、光开工单——content="" 且 tool_calls=[...]。判断"模型是否想用工具"永远看 tool_calls,别看 content。ToolMessage:带工单号的标准回执
你执行完工具,用第四种消息还给模型——ToolMessage(libs/core/langchain_core/messages/tool.py:26):
# libs/core/langchain_core/messages/tool.py:26(裁剪)
class ToolMessage(BaseMessage, ToolOutputMixin):
"""Message for passing the result of executing a tool back to a model.
Example:
ToolMessage(content="42", tool_call_id="call_Jja7J89XsjrOLA5r...")
"""
tool_call_id: str # :67 ★必填!对应 ToolCall.id
artifact: Any = None # :73 原始产物(不发给模型)
status: Literal["success", "error"] = "success" # :81 执行成败标记
type: Literal["tool"] = "tool"
tool_call_id 必填★整个类唯一的必填新字段。模型收到消息列表时,靠它把回执贴回对应工单。tool.py:130 附近还有个校验器:uuid/int/float 类型的 id 自动转成 str——对粗心用户的宽容。content vs artifact类 docstring(:47-62)给了真实例子:图片生成工具把"生成成功"的短文本放 content(发给模型),把 base64 原图放 artifact(不进对话历史,程序自己取用)。和 Day13 的 response_format="content_and_artifact" 首尾呼应。status="error"工具执行失败也要回执!标记 status="error"、content 里写错误信息,模型看到后能换参数重试或改用别的工具——失败信息也是模型的决策输入。自动闭环:tool.invoke(ToolCall) 直接产出 ToolMessage
手工闭环要自己取 id、造 ToolMessage?不用。Day13 埋的伏笔兑现:把整个 ToolCall 直接丢给工具的 invoke。看 libs/core/langchain_core/tools/base.py:731 → _prep_run_args(base.py:1320):
# libs/core/langchain_core/tools/base.py:1320(裁剪)
def _prep_run_args(value, config, **kwargs):
if _is_tool_call(value): # ① 认出这是一张工单
tool_call_id: str | None = value["id"] # ② ★抽出工单号
tool_input = value["args"].copy() # ③ 参数就是 args dict
else:
tool_call_id = None # 普通调用:没有工单号
tool_input = value
return (tool_input, dict(..., tool_call_id=tool_call_id, **kwargs))
执行完,出口处 _format_output(base.py:1358)决定返回什么:
# libs/core/langchain_core/tools/base.py:1358(裁剪)
def _format_output(content, artifact, tool_call_id, name, status):
if isinstance(content, ToolOutputMixin) or tool_call_id is None:
return content # 普通调用 → 原样返回结果
...
return ToolMessage(content, # ★带工单号调用 → 自动包成回执
artifact=artifact,
tool_call_id=tool_call_id,
name=name,
status=status)
同一个 invoke 两种人格tool.invoke({"city":"上海"})(无工单号)返回裸结果字符串——本地调试用;tool.invoke(ai_msg.tool_calls[0])(有工单号)返回现成的 ToolMessage,id 已自动贴好——闭环用。判断依据就是输入是不是 ToolCall 形状。闭环代码就三行for tc in ai_msg.tool_calls: messages.append(tool_map[tc["name"]].invoke(tc))——取工单、执行、回执入列,一气呵成。Day15 的 Agent 里那个 tools 节点,核心干的就是这个。status 也自动填执行抛异常且配置了 handle_tool_error 时,回执自动 status="error"——错误也走标准回执通道。串起来 + 今日小结
model_with_tools = model.bind_tools([get_weather]);第一轮 ai_msg = model_with_tools.invoke([HumanMessage("上海现在多少度?")]) → 返回 AIMessage(content="", tool_calls=[{"name": "get_weather", "args": {"city": "上海"}, "id": "call_ab12", "type": "tool_call"}])。执行:tool_msg = get_weather.invoke(ai_msg.tool_calls[0]) → _prep_run_args 抽出 id="call_ab12",跑 Day13 的 run,出口 _format_output 包成 ToolMessage(content="上海 26 度,晴", tool_call_id="call_ab12", status="success")。第二轮 model_with_tools.invoke([human, ai_msg, tool_msg]) → AIMessage(content="上海现在 26 度,天气晴朗。", tool_calls=[])——工单空了,闭环结束。👶 小白:bind_tools 之后,模型是不是就"拥有"这些工具了?它会不会自己偷偷调用?
👨🏫 老师:完全不会,这是今天最重要的安全认知。模型从头到尾只输出文本——bind_tools 只是每次请求时附一份 JSON 说明书;所谓 tool_calls 也只是"结构化地表达了想调用的意愿"。执行发生在你的 Python 进程里、由你的代码触发,你完全可以在执行前审查、过滤、要求人工确认(Day15 的 middleware 里真有 human_in_the_loop 这一层)。模型是下单的客户,你才是掌勺的厨房。
🧠 今天你应该能回答
- bind_tools 干了什么、没干什么?(把工具翻译成厂商 JSON 挂到请求参数上;不发请求、不执行、不改原模型)
- ToolCall 三要素?(name 调谁 / args 标准 dict 参数 / id 工单号)
- ToolMessage 为什么必须带 tool_call_id?(多工单并发时对账,API 层面强制"每单必回")
- tool.invoke 传 ToolCall 和传 dict 有什么区别?(前者自动抽 id、返回现成 ToolMessage;后者返回裸结果)
- 工具执行失败怎么办?(回 status="error" 的回执,让模型自己决定重试还是换路)
- 整个闭环里代码真正执行在哪?(永远在你本地进程,模型只出文本)
✋ 10 分钟动手
cd /Users/bitmart/work/codes/github/AI_WORK/langchain
# 1. 接口与实现
sed -n '2338,2356p' libs/core/langchain_core/language_models/chat_models.py # 基类接口
grep -n "convert_to_openai_tool\|return super().bind" \
libs/partners/openai/langchain_openai/chat_models/base.py | head -5 # OpenAI 翻译+bind
# 2. 工单与回执
sed -n '160,180p' libs/core/langchain_core/messages/ai.py # AIMessage.tool_calls
sed -n '206,240p' libs/core/langchain_core/messages/tool.py # ToolCall
sed -n '26,82p' libs/core/langchain_core/messages/tool.py # ToolMessage
# 3. 自动闭环两端
sed -n '1320,1356p' libs/core/langchain_core/tools/base.py # _prep_run_args 抽工单号
sed -n '1358,1395p' libs/core/langchain_core/tools/base.py # _format_output 产回执
libs/langchain_v1/langchain/agents/factory.py 的 create_agent——它把这个循环编译成一张 LangGraph 状态图(model 节点 ⇄ tools 节点),这就是 ReAct Agent 的真身,也顺便讲清 LangChain 和 LangGraph 的关系。