企业官网建设流程全解析

本节系统讲解 LangChain 1.0 Middleware：如何在不污染 Agent 业务逻辑的前提下统一加入日志、重试、缓存、动态模型、工具筛选、自动摘要、调用限额与人工审核。

1. 为什么需要 Middleware

一个最小 Agent 只需要模型和工具，但生产应用还会遇到很多横切需求：

• 每次模型和工具调用都要记录耗时；
• 网络抖动需要重试，主模型故障需要降级；
• 长对话接近上下文上限时需要自动摘要；
• 不同用户只能看到自己有权限的工具；
• 删除、支付、发信等操作需要人工批准；
• 必须限制调用次数，防止循环失控和成本异常。

如果把这些逻辑写进每个工具或 Prompt，代码会高度耦合。Middleware 将它们抽离为可组合组件，并运行在create_agent()返回的 LangGraph 内部。

2. Agent 生命周期与钩子

2.1 Node-style Hook

Node-style Hook 返回一个状态更新字典；返回None表示不修改 State。

2.2 Wrap-style Hook

Wrap Hook 可以决定不调用、调用一次或多次handler，因此比前后置 Hook 更适合重试与短路缓存。

3. 装饰器式 Middleware

3.1 模型调用前记录状态

fromtypingimportAnyfromlangchain.agents.middlewareimportbefore_model,AgentStatefromlanggraph.runtimeimportRuntime@before_modeldeflog_before_model(state:AgentState,runtime:Runtime,)->dict[str,Any]|None:print(f"准备调用模型，当前消息数：{len(state['messages'])}")returnNone

3.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:ifattempt==2:raiseprint(f"模型调用失败，第{attempt+1}次重试：{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_at=time.perf_counter()returnNonedefafter_agent(self,state:AgentState,runtime:Runtime,)->dict[str,Any]|None:elapsed=time.perf_counter()-runtime.context.started_atprint(f"Agent 总耗时：{elapsed:.3f}s")returnNone

如果运行时上下文对象不可变，可以把计时数据写入自定义 State，或交给结构化日志/Tracing 系统管理。

5. 内置重试与错误处理

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimport(ModelRetryMiddleware,ToolRetryMiddleware,)agent=create_agent(model=model,tools=[search_tool,database_tool],middleware=[ModelRetryMiddleware(max_retries=3,initial_delay=1.0,backoff_factor=2.0,jitter=True,on_failure="error",),ToolRetryMiddleware(max_retries=2,retry_on=(ConnectionError,TimeoutError),on_failure="continue",),],)

• 模型的 429、超时、临时 5xx 适合重试；
• 参数错误、权限错误通常不应盲目重试；
• on_failure="continue"会把失败转换为消息，让模型决定降级回答；
• on_failure="error"会抛出异常，由服务层统一处理。

重试必须设置上限，否则会放大上游故障和费用。

6. 长对话自动摘要

fromlangchain.agents.middlewareimportSummarizationMiddleware summary_middleware=SummarizationMiddleware(model=summary_model,trigger=("tokens",4000),keep=("messages",20),)

当上下文达到阈值时，它会压缩较旧的消息并保留最近消息。摘要能降低 Token 成本，但也可能损失细节，因此重要业务事实应写入结构化长期记忆，而不是只依赖摘要。

7. 限制模型与工具调用次数

fromlangchain.agents.middlewareimport(ModelCallLimitMiddleware,ToolCallLimitMiddleware,)limits=[ModelCallLimitMiddleware(run_limit=8,thread_limit=40,exit_behavior="end",),ToolCallLimitMiddleware(run_limit=10,thread_limit=50,exit_behavior="continue",),]

调用上限用于控制成本并防止 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(model=selected_model))

路由条件除了消息数，还可以考虑用户等级、任务类型、延迟预算、内容敏感度和主模型健康状态。

9. 动态工具筛选

当工具数量很多时，把全部 Schema 都交给模型会增加 Token、降低选择准确率，也可能造成越权。

@wrap_model_calldeffilter_tools(request:ModelRequest,handler:Callable[[ModelRequest],ModelResponse],)->ModelResponse:user_role=request.runtime.context.user_role allowed_tools=[toolfortoolinrequest.toolsifuser_roleinTOOL_PERMISSIONS[tool.name]]returnhandler(request.override(tools=allowed_tools))

工具必须先注册到 Agent，再在运行时筛选。Middleware 中的筛选是模型可见性控制，真实工具接口仍必须做后端鉴权。

10. 高风险工具的人工审核

fromlangchain.agents.middlewareimportHumanInTheLoopMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent(model=model,tools=[search_tool,send_email,delete_record],middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email":True,"delete_record":True,"search_tool":False,})],checkpointer=InMemorySaver(),)

人工审核依赖 Checkpointer 保存暂停时的状态，并使用固定thread_id恢复。审核界面应展示工具名、完整参数、影响范围与请求来源，而不是只显示“是否允许”。

11. 组合一个生产型 Middleware 栈

fromlangchain.agents.middlewareimport(ModelRetryMiddleware,ToolRetryMiddleware,SummarizationMiddleware,ModelCallLimitMiddleware,HumanInTheLoopMiddleware,)middleware=[log_before_model,ModelCallLimitMiddleware(run_limit=8,exit_behavior="end"),SummarizationMiddleware(model=summary_model,trigger=("tokens",6000),keep=("messages",16),),ModelRetryMiddleware(max_retries=2,on_failure="error"),ToolRetryMiddleware(max_retries=2,on_failure="continue"),HumanInTheLoopMiddleware(interrupt_on={"send_email":True,"delete_record":True}),]agent=create_agent(model=model,tools=all_tools,middleware=middleware,checkpointer=checkpointer,)

12. Middleware 的执行顺序

假设列表为[A, B, C]：

before_*：A → B → C wrap_*： A(B(C(handler))) after_*： C → B → A

因此顺序会影响行为。例如日志放在重试层外侧会记录一次整体调用，放在内侧会记录每一次重试；缓存放在限流前后也会影响缓存命中是否计数。

13. 测试与排错

对 Middleware 至少覆盖以下场景：

1. 正常调用时是否只执行一次；
2. 可重试异常是否达到预期次数；
3. 非重试异常是否立即退出；
4. 缓存命中后是否跳过模型；
5. 调用达到上限是否结束；
6. 未授权工具是否从模型视野中消失且后端拒绝；
7. 人工拒绝后是否停止副作用；
8. 日志中是否泄漏 Prompt、Token 或敏感参数。

14. 本节小结

Middleware 是 LangChain 1.0 从“能跑”走向“可控、可观测、可治理”的关键层。设计时要记住：业务 Tool 只处理业务，中间件处理横切能力；Hook 顺序决定执行语义；鉴权和高风险控制不能只依赖模型。