HarmonyOS 鸿蒙Next中@wrap_model_call 和 @wrap_tool_call 的作用

HarmonyOS 鸿蒙Next中@wrap_model_call 和 @wrap_tool_call 的作用 在 LangChain V1.0+（特别是基于 langgraph 和新版 create_agent 的架构）中，[@wrap_model_call](/user/wrap_model_call) 和 [@wrap_tool_call](/user/wrap_tool_call) 是用于定义 Agent 中间件（Middleware） 的两个核心装饰器。

它们的作用是允许开发者在不修改 Agent 核心逻辑的情况下，拦截、增强或修改 Agent 运行过程中的关键步骤。这类似于 Web 框架（如 Express.js 或 FastAPI）中的中间件概念，但专门针对 LLM Agent 的“思考 - 行动”循环。

以下是它们的详细作用、区别及代码示例：

1. [@wrap_model_call](/user/wrap_model_call)：包裹模型调用

作用位置：在 Agent 调用大语言模型（LLM）生成回复或决定下一步动作之前和之后。

主要用途：

• 自动重试 (Retry)：当模型调用因网络错误或速率限制（Rate Limit）失败时，自动重试。
• 缓存 (Caching)：检查相同的 Prompt 是否已有缓存结果，避免重复调用模型以节省 Token。
• 模型回退 (Fallback)：如果主模型（如 gpt-4）不可用，自动切换到备用模型（如 gpt-3.5）。
• 日志与追踪 (Logging/Tracing)：记录输入 Prompt、输出内容、耗时和 Token 用量，用于调试或接入 LangSmith。
• 提示词注入/修改：在发送给模型前动态修改 System Prompt（例如添加当前的时间或用户上下文）。

函数签名逻辑： 它通常接收一个 handler（下一个处理步骤），并返回一个异步函数，该函数接收 model_inputs（模型输入）并返回 model_outputs。

代码示例：实现简单的重试机制

from langchain.agents.middleware import wrap_model_call
import asyncio

[@wrap_model_call](/user/wrap_model_call)
async def retry_middleware(request, handler):
    """
    当模型调用失败时，自动重试 3 次。
    request: 包含模型输入信息的对象
    handler: 下一个中间件或实际的模型调用函数
    """
    max_retries = 3
    for attempt in range(max_retries):
        try:
            # 调用下一个处理步骤（即真正的模型调用）
            return await handler(request)
        except Exception as e:
            if attempt == max_retries - 1:
                raise e  # 最后一次尝试失败，抛出异常
            print(f"模型调用失败，正在重试 ({attempt + 1}/{max_retries}): {e}")
            await asyncio.sleep(2 ** attempt)  # 指数退避

2. [@wrap_tool_call](/user/wrap_tool_call)：包裹工具调用

作用位置：在 Agent 决定调用某个工具（Tool），实际执行该工具函数之前和之后。

主要用途：

• 权限控制 (Authorization)：检查当前用户是否有权限执行该工具（例如：禁止普通用户调用“删除数据库”工具）。
• 参数修正/验证 (Input Validation)：在工具执行前，清洗或修正模型生成的参数（例如：将相对时间“明天”转换为具体日期）。
• 模拟/Mock (Testing)：在测试环境中，拦截真实工具调用，返回固定的假数据。
• 超时控制 (Timeout)：防止某个工具执行时间过长卡死整个 Agent。
• 错误格式化：捕获工具抛出的异常，将其转换为 LLM 能理解的友好错误消息，而不是让 Agent 崩溃。

函数签名逻辑： 它接收 request（包含工具名称、参数等信息）和 handler（实际的工具执行函数）。

代码示例：实现工具超时和错误处理

from langchain.agents.middleware import wrap_tool_call
import asyncio

[@wrap_tool_call](/user/wrap_tool_call)
async def timeout_and_error_middleware(request, handler):
    """
    为工具调用设置 10 秒超时，并捕获所有异常返回友好提示。
    request: 包含 tool_name, tool_args 等信息
    handler: 实际的工具执行函数
    """
    try:
        # 使用 asyncio.wait_for 设置超时
        result = await asyncio.wait_for(handler(request), timeout=10.0)
        return result
    except asyncio.TimeoutError:
        return f"错误：工具 '{request.tool_name}' 执行超时（超过10秒）。"
    except Exception as e:
        # 捕获异常，返回字符串而不是抛出，让 LLM 知道出错了并尝试其他方案
        return f"工具执行出错：{str(e)}。请检查参数或尝试其他方法。"

3. 如何在 create_agent 中使用

这两个装饰器定义的函数需要作为 中间件列表 传递给 create_agent。

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

# 1. 定义模型
model = ChatOpenAI(model="gpt-4o")

# 2. 定义工具
tools = [get_weather] # 假设已定义

# 3. 组合中间件
# 顺序很重要：请求会按顺序经过这些中间件
my_middleware = [
    retry_middleware,          # 先处理模型重试
    timeout_and_error_middleware, # 再处理工具超时
    # 还可以添加更多...
]

# 4. 创建 Agent
agent = create_agent(
    model=model,
    tools=tools,
    middleware=my_middleware  # 传入中间件列表
)

# 现在 agent 在执行时会自动应用上述逻辑
# response = await agent.ainvoke({"messages": [{"role": "user", "content": "北京天气？"}]})

总结对比

通过合理使用这两个装饰器，你可以构建出高可用、安全且可观测的企业级 Agent，而无需侵入性地修改 Agent 的核心源码。这是 LangChain V1.0 推荐的最佳实践模式。