2 LangChain 1.0 中间件（Middleware）- wrap_model_call、wrap_tool_call

一、问题理解与目标定义

核心问题：厘清 wrap_model_call 和 wrap_tool_call 两个中间件的职责边界、适用场景及实现差异。
分析目标：帮助开发者在实际项目中准确选择和使用这两个钩子，避免误用，并通过 Python 案例展示其典型应用。

二、中间件功能对比

💡 关键区别：

• wrap_model_call 关注 “Agent 如何思考”（模型推理层）
• wrap_tool_call 关注 “Agent 如何行动”（工具执行层）

3 中间件实现对比

3.1 wrap_model_call：动态模型路由 + 缓存

• 使用场景说明：
  • 当对话历史较短时，使用便宜模型节省成本；
  • 当上下文较长或任务复杂时，自动切换到高性能模型保证质量；
  • 对相同输入缓存结果，避免重复调用。

# -*- coding: utf-8 -*-# 运行的时候动态的选择模型from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver from langchain_core.tools import tool from langchain_community.callbacks import get_openai_callback from langchain.chat_models import init_chat_model from langchain.agents.middleware import AgentMiddleware, ModelRequest from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse from langchain.messages import AIMessage import hashlib from typing import Optional, Callable import asyncio import os from dotenv import load_dotenv load_dotenv(override=True) simple_model = init_chat_model(model="deepseek-v3.1",# model_provider='openai', api_key= os.getenv("api_key"), base_url= os.getenv("base_url"), temperature=0.3, max_retries=4,#max_tokens=10) complex_model = init_chat_model(model="gpt-4.1-mini",# model_provider='openai', api_key= os.getenv("api_key"), base_url= os.getenv("base_url"), temperature=0.3, max_retries=4,#max_tokens=10)# 全局缓存（生产环境建议用 Redis），对于已经问过的问题 可以直接返回 _MODEL_CACHE ={}# 1. 系统提示词 system_prompt ="""你是一位幽默的天气预报员。 根据天气给出穿衣建议，用轻松的方式表达。"""@wrap_model_calldefdynamic_model_select(request: ModelRequest, handler)->ModelResponse:""" 根据任务复杂度选择模型 ModelRequest：封装了输入数据（如 messages）、模型配置（如 model）、会话状态（state）等。 ModelResponse：标准化输出格式，确保中间件能统一处理。 wrap_model_call属于 Wrap-Style Hook，其调用时机如下： 触发条件：每次 Agent 尝试调用模型（如 handler(request)）时，该中间件会包裹整个模型调用过程。 执行顺序：在模型实际执行前（handler(request) 调用前）和返回结果后均可插入逻辑。 """# 获取状态 state = request.state messages = state.get("messages",[])# 判断任务复杂度if is_complex_task(messages):#request.model = complex_model request.override(model=complex_model)print("使用复杂的模型GPT-mini")else: request.override(model=simple_model)print("使用简单的模型deepseekV3")# 生成缓存键print("当前的消息：", messages[-1].content) cache_key = hashlib.md5(str(messages[-1].content).encode()).hexdigest()# 缓存命中if cache_key in _MODEL_CACHE:print(f"[wrap_model_call] 缓存命中: {cache_key[:8]}")return _MODEL_CACHE[cache_key]#return AIMessage(content=_MODEL_CACHE[cache_key]) response = handler(request) _MODEL_CACHE[cache_key]= response # 调用 handler(request) 会将修改后的请求传递给下一个中间件或最终模型，确保逻辑可组合。return response defis_complex_task(messages):"""判断任务是否复杂"""ifnot messages:returnFalse last_message = messages[-1].content ifhasattr(messages[-1],'content')elsestr(messages[-1])# 复杂任务判断逻辑 complex_keywords =["分析","比较","优化","设计","架构","多步骤"]# 长度超过 200 字或包含复杂关键词iflen(last_message)>200:returnTruefor keyword in complex_keywords:if keyword in last_message:returnTruereturnFalse# 2. 定义工具@tooldefget_weather(city:str)->str:"""获取指定城市的天气"""returnf"{city}：晴，25度，微风徐徐"@tooldefget_location()->str:"""获取用户位置"""return"北京"# 4. 添加记忆 checkpointer = InMemorySaver() agent = create_agent( model=simple_model,# 默认模型（会被中间件覆盖） tools=[get_weather,get_location], middleware=[dynamic_model_select],) config ={"configurable":{"thread_id":"user-001"}}for event in agent.stream({"messages":[{"role":"user","content":"我在哪里？天气如何？分析一下"}]}, config=config, stream_mode="values"): event['messages'][-1].pretty_print()print("*******************************************************") config ={"configurable":{"thread_id":"user-001"}}for event in agent.stream({"messages":[{"role":"user","content":"我在哪里？天气如何？分析一下"}]}, config=config, stream_mode="values"): event['messages'][-1].pretty_print()

3.2 wrap_tool_call：工具权限控制 + 参数脱敏

使用场景说明：

• 阻止非授权用户执行高危操作（如删除、转账）；
• 在工具调用前后记录审计日志，满足合规要求；
• 对敏感参数（如手机号、邮箱）进行日志脱敏，防止数据泄露。

# -*- coding: utf-8 -*-""" agent_with_memory Author: user Date: 2026/3/16 Description: """from langchain.agents import create_agent from langgraph.checkpoint.memory import InMemorySaver from langchain_core.tools import tool from langchain_community.callbacks import get_openai_callback from langchain.chat_models import init_chat_model from langchain.agents.middleware import wrap_tool_call import re import os from dotenv import load_dotenv load_dotenv(override=True) model = init_chat_model(model="deepseek-v3.1",# model_provider='openai', api_key= os.getenv("api_key"), base_url= os.getenv("base_url"), temperature=0.3, max_retries=4,#max_tokens=10)# 1. 系统提示词 system_prompt ="""你是一位幽默的天气预报员。 根据天气给出穿衣建议，用轻松的方式表达。"""# 2. 定义工具@tooldefget_weather(city:str)->str:"""获取指定城市的天气"""returnf"{city}：晴，25度，微风徐徐"@tooldefget_location()->str:"""获取用户位置"""return"北京"@tooldefsend_email(email:str,user:str,content:str):""" 给用户发送邮件 :param email: 用户email :param user: 用户名字 :param content: 内容 :return: """returnf"已经给用户{user}{email},发功成功!内容是：{content}"@wrap_tool_calldefsecure_tool_execution(request,call):""" 对敏感工具调用进行权限校验和参数脱敏 """print(request) tool_name = request.tool.name args = request.tool_call.get('args')# 工具调用参数字典ifhasattr(request.runtime,'context'): user_role = request.runtime.context.get("user_role")else: user_role='guest'# 权限控制：仅管理员可获取用户地址if tool_name =="get_location"and user_role !="admin":raise PermissionError("无权执行删除操作")# 参数脱敏：对 send_email 工具中的邮箱脱敏（审计用）if tool_name =="send_email"and"email"in args: original_email = args["email"] masked_email = re.sub(r'(.{2}).*(@.*)',r'\1***\2', original_email)print(f"[wrap_tool_call] 邮箱脱敏: {original_email} → {masked_email}")# 注意：此处仅用于日志，实际调用仍用原始参数# 若需真正脱敏，应修改 args（但可能破坏功能）# 记录审计日志print(f"[AUDIT] 用户({user_role}) 调用工具: {tool_name}, 参数: {list(args.keys())}")# 执行原始工具调用 result = call(request)return result # 4. 添加记忆 checkpointer = InMemorySaver()# 5. 创建 Agent agent = create_agent( model=model, tools=[get_weather, get_location,send_email], system_prompt=system_prompt, checkpointer=checkpointer, middleware=[secure_tool_execution])# 6. 运行对话 config ={"configurable":{"thread_id":"user-001"}}with get_openai_callback()as cb:# 使用 stream 方法for event in agent.stream({"messages":[{"role":"user","content":"今天穿什么好？"}]}, config=config, context={"user_role":"admin","session_id":"sess_123"}, stream_mode="values"):#print(event) event['messages'][-1].pretty_print()print("\n--- Token Usage ---")print(f"Total Tokens: {cb.total_tokens}")print(f"Prompt Tokens: {cb.prompt_tokens}")print(f"Completion Tokens: {cb.completion_tokens}")print(f"Total Cost (USD): ${cb.total_cost:.6f}")print(f"Successful Requests: {cb.successful_requests}")print()#print("助手:", response1["messages"][-1].content)# 第二轮对话 - Agent 记住了上下文# response2 = agent.invoke(# {"messages": [{"role": "user", "content": "那我需要带伞吗？"}]},# config=config# )# print("助手:", response2["messages"][-1].content)for event in agent.stream({"messages":[{"role":"user","content":"给小明发送一个邮件（[email protected]）问好！"}]}, config=config, context={'user_role':'guest'}, stream_mode="values"):#print(event) event['messages'][-1].pretty_print()

上面只是一个思想，无法达到真正的脱密。使用 LangChain 内置 PIIMiddleware（最简单）可实现真正的脱敏。

from langchain.agents.middleware import wrap_tool_call,PIIMiddleware agent = create_agent( model=model, tools=[get_weather, get_location,send_email], system_prompt=system_prompt, checkpointer=checkpointer, middleware=[secure_tool_execution, PIIMiddleware("email", strategy="redact", apply_to_input=True)])

四 对比

• wrap_model_call 不适用于工具逻辑
  它只拦截 LLM 调用，无法感知 Agent 是否打算调用工具。
• wrap_tool_call 无法干预模型决策
  它在模型已决定调用某工具后才触发，不能改变“是否调用”的决策。
• 两者可协同工作
  典型流程：
  模型思考（wrap_model_call） → 决定调用工具 → 执行工具（wrap_tool_call）
• 异常处理
  • 在 wrap_model_call 中抛出异常会中断整个 Agent 流程；
  • 在 wrap_tool_call 中抛出异常仅中断当前工具调用，Agent 可继续尝试其他工具。

五、最佳实践建议

✅ 安全控制优先用 wrap_tool_call：因为高危操作都发生在工具层。
✅ 成本优化优先用 wrap_model_call：模型调用是主要费用来源。
✅ 不要在 wrap_tool_call 中修改业务关键参数（除非你确定后果），避免破坏工具契约。
✅ 结合 before_model / after_model 实现更精细的上下文治理（如摘要、注入）。