🎯 课程主题

深入理解 Wrap-style 钩子函数中的第二个核心：wrap_tool_call，掌握对工具调用的"包裹"机制，包括监控耗时、修改工具参数、重试调用等实际应用场景。

📝 核心知识点

1. wrap_tool_call 与 wrap_model_call 对比

维度wrap_model_callwrap_tool_call
包裹目标大模型的调用工具的调用
request.tool_call❌ 不包含✅ 包含工具调用的 args
handler 含义调用模型调用具体工具
核心修改对象messages / 提示词tool_call.args（工具参数）

2. wrap_tool_call 核心参数

@wrap_tool_call
def my_wrapper(request, handler):
    ...
参数类型说明
requestRequest包含 tool_call 字段，tool_call 中有 args（工具参数字典）
handlercallable工具调用的处理器，handler(request) 即执行工具
返回值Response必须返回 handler 的响应结果

3. request.tool_call.args

# tool_call 结构示例
request.tool_call = {
    "name": "get_weather",
    "args": {
        "city": "北京",
        "is_forecast": False  # ← 可修改的参数
    },
    "id": "call_abc123"
}

4. 使用场景

场景说明
监控工具耗时调用前后记录时间，计算工具执行时长
修改工具参数在工具执行前动态修改 args（如补充默认参数、强制启用某选项）
重试逻辑工具调用失败时自动重试
工具调用审计记录工具调用的完整请求和响应日志

🏗️ 架构与工作流

AI 决定调用工具 get_weather(city="北京", is_forecast=False)
    │
    ▼
┌──────────────────────────────────────────┐
│        wrap_tool_call 包裹层              │
│                                          │
│  ① 请求前处理                            │
│     读取 tool_call.args                  │
│     修改参数（如 is_forecast → True）     │
│         │                                │
│         ▼                                │
│  ② handler(request)                      │
│     执行真实的工具调用                     │
│         │                                │
│         ▼                                │
│  ③ 响应后处理                            │
│     记录结果 / 日志                       │
│                                          │
└──────────────────────────────────────────┘
    │
    ▼
工具返回结果 → Agent 继续处理

💻 代码实战

1. 基于装饰器的 wrap_tool_call

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

# ---- 定义工具 ----
@tool
def get_weather(city: str, is_forecast: bool = False) -> str:
    """查询天气，is_forecast=True 时包含明日预报"""
    if is_forecast:
        return f"{city}今天天气不错，明天天气也挺好。"
    return f"{city}今天天气不错。"

# ---- 定义 wrap_tool_call 钩子 ----
@wrap_tool_call
def weather_param_modifier(request, handler):
    """
    包裹工具调用：显示原始参数 → 修改参数 → 显示修改后结果
    """
    # ① 调用前：查看并修改工具参数
    original_args = request.tool_call["args"]
    print(f"原始参数: {original_args}")

    # 修改参数：强制查询明日天气
    request.tool_call["args"]["is_forecast"] = True
    print(f"修改后参数: {request.tool_call['args']}")

    # ② 执行原始工具调用
    result = handler(request)
    print(f"原始调用结果: {result}")

    # ③ 执行修改参数后的工具调用
    response = handler(request)
    print(f"更新参数后的调用结果: {response}")

    return response

# ---- 创建 Agent ----
agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[weather_param_modifier]
)

# ---- 调用测试：默认 is_forecast=False ----
response = agent.invoke(
    {"messages": [HumanMessage(content="帮我查询北京今天的天气如何")]}
)
for msg in response["messages"]:
    msg.pretty_print()

输出效果：

原始参数: {'city': '北京', 'is_forecast': False}
修改后参数: {'city': '北京', 'is_forecast': True}
原始调用结果: 北京今天天气不错。
更新参数后的调用结果: 北京今天天气不错，明天天气也挺好。

2. 基于类的实现

from langchain.agents.middleware import AgentMiddleware

class ToolCallMiddleware(AgentMiddleware):
    """基于类实现 wrap_tool_call"""

    def wrap_tool_call(self, state, request, handler):
        # ① 调用前：修改工具参数
        original_args = request.tool_call["args"]
        print(f"原始参数: {original_args}")

        request.tool_call["args"]["is_forecast"] = True
        print(f"修改后参数: {request.tool_call['args']}")

        # ② 执行工具调用
        response = handler(request)
        print(f"更新参数后的调用结果: {response}")

        return response

# 使用
agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[ToolCallMiddleware()]
)

3. 典型应用场景：监控工具耗时

import time

@wrap_tool_call
def tool_monitor(request, handler):
    """监控工具调用的执行耗时"""
    tool_name = request.tool_call.get("name", "unknown")
    start_time = time.time()

    response = handler(request)

    elapsed = time.time() - start_time
    print(f"[监控] 工具 '{tool_name}' 执行耗时: {elapsed:.3f}s")

    return response

4. 典型应用场景：工具调用重试

@wrap_tool_call
def tool_retry(request, handler, max_retries=3):
    """工具调用失败时自动重试"""
    last_error = None
    for attempt in range(max_retries):
        try:
            return handler(request)
        except Exception as e:
            last_error = e
            print(f"[重试 {attempt + 1}/{max_retries}] 工具调用失败: {e}")
    raise last_error

5. 典型应用场景：工具参数校验与补充

@wrap_tool_call
def tool_args_validator(request, handler):
    """校验并补充工具参数"""
    args = request.tool_call["args"]
    # 补充默认页码
    if "page" not in args:
        args["page"] = 1
    # 限制每页最大数量
    if args.get("page_size", 10) > 100:
        args["page_size"] = 100
    return handler(request)

⚠️ 常见问题与避坑指南

问题说明
直接修改 args 的风险修改 request.tool_call["args"] 会直接影响下游；确认修改意图后再操作
忘记调用 handler必须调用 handler(request)，否则工具不会执行
修改参数类型确保修改后的 args 值类型与工具函数签名一致（如字符串不改数字）
类方式方法名必须用 wrap_tool_call，不能自定义名称
监控耗时的影响time.time() 调用本身开销极小，但日志输出多时可能影响性能
重试中的幂等性工具重试需要确保工具调用是幂等的，否则可能重复创建资源

💡 个人总结与延伸

  • wrap_tool_call 是绕过大模型间接控制工具调用的"后门"——模型想查今天天气，你可以让它顺便把明天的也查了
  • 与 wrap_model_call 配合使用可形成完整的请求/响应拦截链：模型调用层 + 工具调用层双层包裹
  • 监控和审计场景在企业级 Agent 中非常重要：了解每个工具耗时才能做性能优化，记录每次调用才能做故障排查
  • 装饰器方式底层仍创建 AgentMiddleware 子类实例，两种方式在功能上完全等价
最近更新:
Contributors: yangwenguang