本节系统讲解 LangChain 1.0 Middleware如何在不污染 Agent 业务逻辑的前提下统一加入日志、重试、缓存、动态模型、工具筛选、自动摘要、调用限额与人工审核。1. 为什么需要 Middleware一个最小 Agent 只需要模型和工具但生产应用还会遇到很多横切需求每次模型和工具调用都要记录耗时网络抖动需要重试主模型故障需要降级长对话接近上下文上限时需要自动摘要不同用户只能看到自己有权限的工具删除、支付、发信等操作需要人工批准必须限制调用次数防止循环失控和成本异常。如果把这些逻辑写进每个工具或 Prompt代码会高度耦合。Middleware 将它们抽离为可组合组件并运行在create_agent()返回的 LangGraph 内部。2. Agent 生命周期与钩子2.1 Node-style HookHook执行时机适合场景before_agent整个 Agent 开始前仅一次初始化、输入校验before_model每次调用模型前裁剪消息、计数、动态上下文after_model每次模型返回后输出校验、用量统计after_agent整个 Agent 结束后仅一次审计、汇总、清理Node-style Hook 返回一个状态更新字典返回None表示不修改 State。2.2 Wrap-style HookHook包裹对象能力wrap_model_call每一次模型调用重试、缓存、路由、降级wrap_tool_call每一次工具调用鉴权、审计、错误转换Wrap Hook 可以决定不调用、调用一次或多次handler因此比前后置 Hook 更适合重试与短路缓存。3. 装饰器式 Middleware3.1 模型调用前记录状态fromtypingimportAnyfromlangchain.agents.middlewareimportbefore_model,AgentStatefromlanggraph.runtimeimportRuntimebefore_modeldeflog_before_model(state:AgentState,runtime:Runtime,)-dict[str,Any]|None:print(f准备调用模型当前消息数{len(state[messages])})returnNone3.2 包裹模型调用实现重试fromcollections.abcimportCallablefromlangchain.agents.middlewareimport(wrap_model_call,ModelRequest,ModelResponse,)wrap_model_calldefretry_model(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelResponse:forattemptinrange(3):try:returnhandler(request)exceptExceptionasexc:ifattempt2:raiseprint(f模型调用失败第{attempt1}次重试{type(exc).__name__})演示代码说明了控制点生产环境更推荐使用内置重试中间件因为它提供指数退避、抖动、异常筛选和最终失败策略。4. 类式 Middleware当一个中间件包含多个 Hook、需要配置参数或要同时实现同步/异步版本时使用类式写法更清晰。importtimefromtypingimportAnyfromlangchain.agents.middlewareimportAgentMiddleware,AgentStatefromlanggraph.runtimeimportRuntimeclassTimingMiddleware(AgentMiddleware):defbefore_agent(self,state:AgentState,runtime:Runtime,)-dict[str,Any]|None:runtime.context.started_attime.perf_counter()returnNonedefafter_agent(self,state:AgentState,runtime:Runtime,)-dict[str,Any]|None:elapsedtime.perf_counter()-runtime.context.started_atprint(fAgent 总耗时{elapsed:.3f}s)returnNone如果运行时上下文对象不可变可以把计时数据写入自定义 State或交给结构化日志/Tracing 系统管理。5. 内置重试与错误处理fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimport(ModelRetryMiddleware,ToolRetryMiddleware,)agentcreate_agent(modelmodel,tools[search_tool,database_tool],middleware[ModelRetryMiddleware(max_retries3,initial_delay1.0,backoff_factor2.0,jitterTrue,on_failureerror,),ToolRetryMiddleware(max_retries2,retry_on(ConnectionError,TimeoutError),on_failurecontinue,),],)模型的 429、超时、临时 5xx 适合重试参数错误、权限错误通常不应盲目重试on_failurecontinue会把失败转换为消息让模型决定降级回答on_failureerror会抛出异常由服务层统一处理。重试必须设置上限否则会放大上游故障和费用。6. 长对话自动摘要fromlangchain.agents.middlewareimportSummarizationMiddleware summary_middlewareSummarizationMiddleware(modelsummary_model,trigger(tokens,4000),keep(messages,20),)当上下文达到阈值时它会压缩较旧的消息并保留最近消息。摘要能降低 Token 成本但也可能损失细节因此重要业务事实应写入结构化长期记忆而不是只依赖摘要。7. 限制模型与工具调用次数fromlangchain.agents.middlewareimport(ModelCallLimitMiddleware,ToolCallLimitMiddleware,)limits[ModelCallLimitMiddleware(run_limit8,thread_limit40,exit_behaviorend,),ToolCallLimitMiddleware(run_limit10,thread_limit50,exit_behaviorcontinue,),]调用上限用于控制成本并防止 Agent 反复调用同一工具。具体参数可能随 LangChain 小版本变化项目中应锁定依赖版本并以对应版本文档为准。8. 动态模型路由简单问题使用低成本模型复杂上下文切换到能力更强的模型fromcollections.abcimportCallablefromlangchain.agents.middlewareimport(wrap_model_call,ModelRequest,ModelResponse,)wrap_model_calldefroute_model(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelResponse:selected_model(strong_modeliflen(request.messages)10elsefast_model)returnhandler(request.override(modelselected_model))路由条件除了消息数还可以考虑用户等级、任务类型、延迟预算、内容敏感度和主模型健康状态。9. 动态工具筛选当工具数量很多时把全部 Schema 都交给模型会增加 Token、降低选择准确率也可能造成越权。wrap_model_calldeffilter_tools(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)-ModelResponse:user_rolerequest.runtime.context.user_role allowed_tools[toolfortoolinrequest.toolsifuser_roleinTOOL_PERMISSIONS[tool.name]]returnhandler(request.override(toolsallowed_tools))工具必须先注册到 Agent再在运行时筛选。Middleware 中的筛选是模型可见性控制真实工具接口仍必须做后端鉴权。10. 高风险工具的人工审核fromlangchain.agents.middlewareimportHumanInTheLoopMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver agentcreate_agent(modelmodel,tools[search_tool,send_email,delete_record],middleware[HumanInTheLoopMiddleware(interrupt_on{send_email:True,delete_record:True,search_tool:False,})],checkpointerInMemorySaver(),)人工审核依赖 Checkpointer 保存暂停时的状态并使用固定thread_id恢复。审核界面应展示工具名、完整参数、影响范围与请求来源而不是只显示“是否允许”。11. 组合一个生产型 Middleware 栈fromlangchain.agents.middlewareimport(ModelRetryMiddleware,ToolRetryMiddleware,SummarizationMiddleware,ModelCallLimitMiddleware,HumanInTheLoopMiddleware,)middleware[log_before_model,ModelCallLimitMiddleware(run_limit8,exit_behaviorend),SummarizationMiddleware(modelsummary_model,trigger(tokens,6000),keep(messages,16),),ModelRetryMiddleware(max_retries2,on_failureerror),ToolRetryMiddleware(max_retries2,on_failurecontinue),HumanInTheLoopMiddleware(interrupt_on{send_email:True,delete_record:True}),]agentcreate_agent(modelmodel,toolsall_tools,middlewaremiddleware,checkpointercheckpointer,)12. Middleware 的执行顺序假设列表为[A, B, C]before_*A → B → C wrap_* A(B(C(handler))) after_* C → B → A因此顺序会影响行为。例如日志放在重试层外侧会记录一次整体调用放在内侧会记录每一次重试缓存放在限流前后也会影响缓存命中是否计数。13. 测试与排错对 Middleware 至少覆盖以下场景正常调用时是否只执行一次可重试异常是否达到预期次数非重试异常是否立即退出缓存命中后是否跳过模型调用达到上限是否结束未授权工具是否从模型视野中消失且后端拒绝人工拒绝后是否停止副作用日志中是否泄漏 Prompt、Token 或敏感参数。14. 本节小结Middleware 是 LangChain 1.0 从“能跑”走向“可控、可观测、可治理”的关键层。设计时要记住业务 Tool 只处理业务中间件处理横切能力Hook 顺序决定执行语义鉴权和高风险控制不能只依赖模型。