wrap_model_call 是中间件Middleware中最强大的钩子。wrap_model_call 不像 before/after 那样只是观察而是可以完全控制模型的执行过程——重试、降级、缓存、甚至跳过模型直接用预设回复。理解 handler 回调wrap_model_call 的核心是一个 handler 回调函数。调用 handler(request) 才会真正执行模型不调用则跳过模型。实例# wrap_model_call 的基本结构# request: 包含模型、消息、工具等全部信息# handler: 一个可调用对象执行它才会真正调用模型wrap_model_calldef my_middleware(request, handler):# 在模型调用前可以做任何事print(模型即将被调用...)# 调用 handler(request) 才真正执行模型response handler(request)# 在模型调用后可以做任何事print(模型调用完成)return response场景 1重试机制这是最常见的场景——模型调用可能因网络问题失败自动重试可以提高可靠性实例from dotenv import load_dotenvload_dotenv()from langchain.agents import create_agentfrom langchain.agents.middleware import wrap_model_callfrom langchain.chat_models import init_chat_modelfrom langchain.messages import HumanMessagewrap_model_calldef retry_on_error(request, handler):模型调用失败时自动重试最多 3 次max_retries 3last_error Nonefor attempt in range(max_retries):try:result handler(request)if attempt 0:print(f [重试成功] 第 {attempt 1} 次尝试)return resultexcept Exception as e:last_error eif attempt max_retries - 1:wait_time (attempt 1) * 2 # 递增等待2s, 4sprint(f [重试] 第 {attempt 1} 次失败{wait_time}秒后重试...)import timetime.sleep(wait_time)# 所有重试都失败了raise last_errormodel init_chat_model(deepseek:deepseek-v4-flash, temperature0)agent create_agent(modelmodel,middleware[retry_on_error],system_prompt你是菜鸟教程 RUNOOB 的助手。,)result agent.invoke({messages: [HumanMessage(content介绍一下菜鸟教程)]})print(f\n回复: {result[messages][-1].content[:100]}...)场景 2模型降级/故障转移当主模型不可用时自动切换到备用模型实例from langchain.agents.middleware import wrap_model_callfrom langchain.chat_models import init_chat_modelfrom langchain.messages import AIMessage# 主模型和备用模型primary_model init_chat_model(deepseek:deepseek-v4-flash, temperature0)fallback_model init_chat_model(deepseek:deepseek-v4-flash, temperature0,max_tokens100) # 降级模型限制 Tokenwrap_model_calldef fallback_on_error(request, handler):主模型失败时自动切换到备用模型try:# 尝试用主模型return handler(request)except Exception as e:print(f[降级] 主模型失败: {e}切换到备用模型...)# 覆盖 request 中的 model切换到备用模型request request.override(modelfallback_model)try:return handler(request)except Exception as e2:print(f[降级] 备用模型也失败了: {e2})# 两个模型都失败返回友好提示return AIMessage(content抱歉服务暂时不可用请稍后重试。)request.override() 是一个不可变方法——它返回一个新的 request 副本不会修改原始对象。这确保了每次调用都是独立和安全的。场景 3缓存模型响应对于重复的查询可以缓存模型响应以减少 API 调用成本实例from langchain.agents.middleware import wrap_model_callfrom langchain.messages import AIMessage# 简单的内存缓存cache {}wrap_model_calldef cache_responses(request, handler):缓存模型响应相同问题不重复调用# 使用最后一条用户消息的内容作为缓存键messages request.messagesif not messages:return handler(request)# 生成缓存键last_content str(messages[-1].content) if hasattr(messages[-1], content) else cache_key last_content[:200] # 截断超长内容# 检查缓存if cache_key in cache:print(f[缓存命中] 直接返回缓存结果)cached cache[cache_key]return AIMessage(contentf{cached}\n\n*(来自缓存)*)# 缓存未命中调用模型result handler(request)# AIMessage 的 content 可能混在列表中直接取第一个if hasattr(result, content):cache[cache_key] result.contentprint(f[缓存未命中] 已存入缓存当前 {len(cache)} 条)elif hasattr(result, model_response):# ExtendedModelResponse 的情况passreturn resultmodel init_chat_model(deepseek:deepseek-v4-flash, temperature0)agent create_agent(modelmodel,middleware[cache_responses],system_prompt你是菜鸟教程 RUNOOB 的助手回答要简洁。,)# 第一次查询缓存未命中result agent.invoke({messages: [{role: user, content: Python 是什么}]})print(f第一次: {result[messages][-1].content[:80]}...\n)# 第二次查询相同问题缓存命中result agent.invoke({messages: [{role: user, content: Python 是什么}]})print(f第二次: {result[messages][-1].content[:80]}...)运行结果[缓存未命中] 已存入缓存当前 1 条 第一次: Python 是一种高级编程语言以简洁易读的语法著称... [缓存命中] 直接返回缓存结果 第二次: Python 是一种高级编程语言以简洁易读的语法著称... *(来自缓存)*场景 4修改 request——动态注入系统消息实例from datetime import datetimefrom langchain.agents.middleware import wrap_model_callfrom langchain.messages import SystemMessagewrap_model_calldef inject_time_context(request, handler):在每次模型调用前注入当前时间信息now datetime.now()time_context (f当前时间{now.strftime(%Y年%m月%d日 %H:%M)}。f今天是{[周一,周二,周三,周四,周五,周六,周日][now.weekday()]}。)# 在原有 system_message 基础上追加时间信息if request.system_message:new_content f{request.system_message.content}\n\n{time_context}else:new_content time_context# 用 override 创建新的 requestnew_request request.override(system_messageSystemMessage(contentnew_content))return handler(new_request)场景 5多个 wrap_model_call 的组合多个 wrap_model_call 中间件会自动按顺序组合——第一个在最外层实例from langchain.agents.middleware import wrap_model_callwrap_model_calldef outer_middleware(request, handler):最外层中间件print([外层] 开始)result handler(request) # 这里会进入 inner_middlewareprint([外层] 结束)return resultwrap_model_calldef inner_middleware(request, handler):内层中间件print( [内层] 开始)result handler(request) # 这里才真正调用模型print( [内层] 结束)return result# 执行顺序# [外层] 开始# [内层] 开始# → 真正调用模型# [内层] 结束# [外层] 结束多个 wrap_model_call 就像洋葱一样层层包裹。最外层最先执行、最后返回。这让你可以组合多个独立的功能——比如外层做缓存检查内层做重试互不干扰。