第三章:`wrap_tool_call`
wraptoolcall 是包住工具执行的 middleware:它拿到模型发出的 tool call,调用 handler(request) 后工具才会真正执行。它解决的是工具层控制问题,例如审计、重试、错误转译、结果改写和敏感信息过滤。
wrap_tool_call 是包住工具执行的 middleware:它拿到模型发出的 tool call,调用 handler(request) 后工具才会真正执行。它解决的是工具层控制问题,例如审计、重试、错误转译、结果改写和敏感信息过滤。
最小代码
代码在 deepagent_src/middleware_teach/03_wrap_tool_call.py。
@wrap_tool_call
def audit_tool_call(request, handler):
tool_events.append(f"before:{request.tool_call['name']}")
result = handler(request)
tool_events.append(f"after:{result.content}")
return ToolMessage(
content=f"AUDITED:{result.content}",
tool_call_id=request.tool_call["id"],
name=request.tool_call["name"],
)
handler(request) 是工具真正执行的位置;不调用它,就不会调用工具。示例先让模型发出 add_one(value=2),工具返回 3,middleware 再把工具消息改成 AUDITED:3。
运行命令
uv run python -m deepagent_src.middleware_teach.03_wrap_tool_call
预期现象
tool_events: ["before:add_one:{'value': 2}", 'after:3']
tool_message: AUDITED:3
final: TOOL_DONE
wrap_tool_call local check ok
这说明工具节点真的执行了:before 记录模型请求的工具名和参数,after 记录原始工具结果,最终进入对话历史的是被 middleware 改写后的 ToolMessage。
常见误区
wrap_tool_call 管的是工具执行,不管模型是否会选择工具。模型要不要发 tool call,是模型层和 prompt/tool schema 的问题;工具一旦被选中,才进入 wrap_tool_call。艹,别指望它替你解决“模型不调用工具”的问题。
和内置 middleware 的关系
ToolRetryMiddleware 和 ToolErrorMiddleware 本质上也工作在工具层。区别是:前者负责失败后重试,后者负责把异常转换成模型可见的工具错误消息。自定义 wrap_tool_call 更适合做审计、权限检查、结果脱敏这类业务逻辑。
为什么用了 ToolCallingFakeModel
当前 FakeMessagesListChatModel 不实现 bind_tools(),直接挂工具会报 NotImplementedError。示例里的 ToolCallingFakeModel 只是给本地教学模型补一个最小 bind_tools(),让 LangChain agent loop 能走到工具节点;真实 OpenAI/Anthropic/DeepSeek 等工具调用模型不需要这层补丁。
相关资源
查看示例代码:deepagent_src/middleware_teach/03_wrap_tool_call.py
from __future__ import annotations from collections.abc import Callable from typing import Any from langchain.agents import create_agent from langchain.agents.middleware import ToolCallRequest, wrap_tool_call from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel from langchain_core.messages import AIMessage, ToolMessage from langchain_core.tools import tool tool_events: list[str] = [] class ToolCallingFakeModel(FakeMessagesListChatModel): def bind_tools( self, tools: Any, *, tool_choice: Any = None, **kwargs: Any, ) -> ToolCallingFakeModel: return self @tool def add_one(value: int) -> str: """Add one to a number.""" return str(value + 1) @wrap_tool_call(name="audit_tool_call") def audit_tool_call( request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage], ) -> ToolMessage: tool_events.append(f"before:{request.tool_call['name']}:{request.tool_call['args']}") result = handler(request) tool_events.append(f"after:{result.content}") return ToolMessage( content=f"AUDITED:{result.content}", tool_call_id=request.tool_call["id"], name=request.tool_call["name"], ) def main() -> None: tool_events.clear() model = ToolCallingFakeModel( responses=[ AIMessage( content="", tool_calls=[ { "name": "add_one", "args": {"value": 2}, "id": "call_1", } ], ), AIMessage(content="TOOL_DONE"), ] ) agent = create_agent( model=model, tools=[add_one], middleware=[audit_tool_call], ) state = agent.invoke({"messages": [{"role": "user", "content": "run tool"}]}) tool_messages = [message for message in state["messages"] if message.type == "tool"] final_text = state["messages"][-1].text print("tool_events:", tool_events) print("tool_message:", tool_messages[0].text) print("final:", final_text) assert tool_events == ["before:add_one:{'value': 2}", "after:3"], tool_events assert tool_messages[0].text == "AUDITED:3", tool_messages[0].text assert final_text == "TOOL_DONE", final_text print("wrap_tool_call local check ok") if __name__ == "__main__": main()