本页目录
Middleware / 中间件Note 89

第三章:`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 的关系

ToolRetryMiddlewareToolErrorMiddleware 本质上也工作在工具层。区别是:前者负责失败后重试,后者负责把异常转换成模型可见的工具错误消息。自定义 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()