🎯 课程主题

深入理解 Wrap-style（包裹风格）钩子函数中第一个核心：wrap_model_call，掌握其对模型调用的"包裹"机制，包括拦截、修改请求、处理响应和缓存等实际应用场景。

📝 核心知识点

1. Wrap-style 钩子概述

与 Node-style（before_* / after_*）不同，Wrap-style 钩子采用"包裹"模式：

钩子作用
wrap_model_call包裹模型的调用过程
wrap_tool_call包裹工具的调用过程

2. wrap_model_call 核心参数

@wrap_model_call
def my_middleware(request, handler):
    ...
参数类型说明
requestRequest发送给模型的请求数据，包含 messages、tools、state、temperature 等
handlercallable模型的调用句柄（处理器），handler(request) 即执行模型调用
返回值Response必须返回 handler 的响应结果

3. 包裹模型的核心模式

修饰前（handler 处理）：
  request → handler → response

修饰后（wrap_model_call 包裹）：
  request → [请求前处理] → handler → [响应后处理] → response

代码结构：

@wrap_model_call
def my_wrapper(request, handler):
    # 1. 调用前：修改 request（如追加消息、修改提示词）
    # 2. 调用中：handler(request) → response
    response = handler(request)
    # 3. 调用后：修改 response（如过滤、格式化、缓存）
    return response

4. 使用场景

场景说明
重试逻辑调用失败或结果不满意时，循环 handler 重试
响应缓存调用前查缓存（命中则直接返回），调用后写缓存
修改系统提示词在 request 中注入额外信息（时间、位置、语言偏好等）
敏感词过滤在 response 返回前过滤/替换敏感内容
输出格式化强制对模型输出进行格式化处理

🏗️ 架构与工作流

                     wrap_model_call 包裹层
         ┌──────────────────────────────────────┐
         │                                      │
         │  ① 请求前处理                        │
         │  (修改 messages/tools/prompts)        │
         │         │                            │
         │         ▼                            │
request ─┤  ② handler(request) ──→ response     │
         │     (真正的模型调用)                   │
         │         │                            │
         │         ▼                            │
         │  ③ 响应后处理                        │
         │  (过滤/格式化/缓存)                   │
         │                                      │
         └──────────────────────────────────────┘

💻 代码实战

1. 基于装饰器的 wrap_model_call

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call
from langchain_core.messages import HumanMessage

# ---- 定义 wrap_model_call 钩子 ----
@wrap_model_call
def my_model_wrapper(request, handler):
    """
    包裹模型调用：调用前在消息内容后追加标记，调用后在响应后追加标记
    """
    # ① 调用前：修改请求中的消息
    messages = request.messages
    last_msg = messages[-1]
    if hasattr(last_msg, "content"):
        last_msg.content += " [wrap_model_call before]"

    # ② 执行模型调用
    response = handler(request)

    # ③ 调用后：修改响应内容
    result_content = response["messages"][0].content
    response["messages"][0].content = result_content + " [wrap_model_call after]"

    return response

# ---- 创建 Agent ----
agent = create_agent(
    model=model,
    middleware=[my_model_wrapper]
)

# ---- 调用测试 ----
response = agent.invoke({"messages": [HumanMessage(content="你好")]})
for msg in response["messages"]:
    msg.pretty_print()

2. 基于类的实现

from langchain.agents.middleware import AgentMiddleware

class ModelCallMiddleware(AgentMiddleware):
    """基于类实现 wrap_model_call"""

    def wrap_model_call(self, state, request, handler):
        # ① 请求前处理
        messages = request.messages
        last_msg = messages[-1]
        if hasattr(last_msg, "content"):
            last_msg.content += " [wrap_model_call before]"

        # ② 执行模型调用
        response = handler(request)

        # ③ 响应后处理
        result_content = response["messages"][0].content
        response["messages"][0].content = result_content + " [wrap_model_call after]"

        return response

# 使用
agent = create_agent(
    model=model,
    middleware=[ModelCallMiddleware()]
)

3. 典型应用场景：响应缓存

from functools import lru_cache

# 简易内存缓存
cache = {}

@wrap_model_call
def caching_wrapper(request, handler):
    """命中缓存则直接返回，否则调用模型后缓存结果"""
    # 用消息内容生成缓存键
    user_content = request.messages[-1].content
    cache_key = hash(user_content)

    # ① 查缓存
    if cache_key in cache:
        print(f"✓ 命中缓存: {user_content[:30]}...")
        return cache[cache_key]

    # ② 未命中，调用模型
    print(f"✗ 未命中缓存，调用模型...")
    response = handler(request)

    # ③ 写入缓存
    cache[cache_key] = response
    return response

4. 典型应用场景：修改系统提示词

from datetime import datetime

@wrap_model_call
def enrich_system_prompt(request, handler):
    """在请求中注入当前时间和用户偏好"""
    # 构造新的 messages（追加系统提示）
    enriched = list(request.messages)
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    enriched.insert(0, {
        "role": "system",
        "content": f"当前时间: {now}。请使用中文回答。"
    })
    # 创建新的 request 对象
    new_request = request.copy(update={"messages": enriched})
    return handler(new_request)

5. 典型应用场景：重试逻辑

@wrap_model_call
def retry_wrapper(request, handler):
    """失败时最多重试3次"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            response = handler(request)
            return response
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"第{attempt + 1}次调用失败: {e}，重试中...")
    return None

⚠️ 常见问题与避坑指南

问题说明
忘记调用 handler如果不调用 handler(request)，模型根本不会执行，响应为空
忘记 return response必须返回 handler 的结果；不返回会导致 Agent 收不到模型响应
修改 request 方式直接修改 request.messages 可能不安全，推荐创建新的 request 对象
类方式方法名必须用 wrap_model_call，与装饰器名称一致；不能写成 wrap_model_call_middleware
缓存键设计只按消息内容做缓存键可能不够，需考虑 tools、system_prompt 等上下文

💡 个人总结与延伸

  • wrap_model_call 是比 before_model / after_model 更强大的钩子机制——它直接拿到 request 和 handler，能做拦截、缓存、重试等高级操作
  • 与 Python 装饰器模式（Decorator Pattern）本质一致：wrapper(func) → 增强版 func
  • 相比 Node-style 的"观察"模式，Wrap-style 是"代理"模式——即 handler 是否被调用、何时被调用、调用几次，完全由你控制
  • DeepSeek 等厂商对"命中缓存"的定价极低（输入 $0.014/百万 Token vs 缓存命中 $0.014/百万 Token），通过 wrap_model_call 实现应用层缓存可进一步降低成本
最近更新:
Contributors: yangwenguang