writing-custom-middleware

能力目标

写出一件继承基类的自定义中间件、挂进智能体、看到自己的钩子被真实触发，并能按需求选对钩子：读写状态用一族、改模型或工具调用的入参出参用另一族，且能拿出「输出真的变了」的效果证据而不只是「钩子被触发了」。

前置

• 已能构造智能体（见 bootstrapping-deepagents-env）。
• 唯一的语言门槛是类继承与方法覆盖。
• 基类来自 langchain.agents.middleware，随装具框架的依赖链一并装上。
• 一件事先说清：middleware= 参数是追加到框架预装的那一摞上，不是替换。挂了自定义中间件后，预装的那些节点仍在图里。

实操流程

1. 写最小可观测的一件——统计模型被调用了几次：

   from langchain.agents.middleware import AgentMiddleware
   
   class CallCounterMiddleware(AgentMiddleware):
       """统计模型调用次数。"""
   
       def __init__(self):
           self.calls = 0
   
       def wrap_model_call(self, request, handler):
           self.calls += 1
           result = handler(request)   # ← 必须调，不调就断链
           return result
   

2. 挂载并跑一次：

   mw = CallCounterMiddleware()
   agent = create_deep_agent(model=model, tools=[], middleware=[mw])
   agent.invoke({"messages": [{"role": "user", "content": "Say hello in one word"}]})
   print(mw.calls)   # >0 即钩子真触发
   

3. 按需求选钩子。六个钩子分两族，选择口诀：只需要读写状态或在生命周期节点插一段逻辑，用观察族（签名带状态与运行时、返回状态增量或空、落为独立图节点、不必转发）；需要改模型或工具调用的入参出参、或控制调用次数，用包裹族（签名带请求与内层入口、必须转发）：

   钩子	用来做什么	触发时机
   before_agent	启动前初始化、校验	每次调用一次
   before_model	每次模型调用前读写状态、可写提前终止字段	每次模型调用前
   after_model	每次模型调用后读写状态	每次模型调用后
   after_agent	结束后清理、统计、持久化	每次调用一次
   wrap_model_call	注入提示、裁剪工具、短路、重试	包裹每次模型调用
   wrap_tool_call	工具调用监控、修改、重试	包裹每次工具调用

4. 用包裹族改发给模型的请求。官方接口是请求对象的覆写方法，它返回一个新请求实例，必须把新对象传给内层：

   class SystemPromptInjectorMW(AgentMiddleware):
       def __init__(self, injection):
           self.injection = injection
   
       def wrap_model_call(self, request, handler):
           original = getattr(request, "system_message", None)
           new_system = f"{original}\n\n{self.injection}" if original else self.injection
           return handler(request.override(system_message=new_system))
   

   同一套写法换个字段就是按需裁剪工具：

   class ToolFilterMW(AgentMiddleware):
       def __init__(self, filter_out):
           self.filter_out = filter_out
   
       def wrap_model_call(self, request, handler):
           filtered = [t for t in (request.tools or []) if getattr(t, "name", "") not in self.filter_out]
           return handler(request.override(tools=filtered))
   

5. 用内层入口的调用次数当闸门，控制内层行为：

   • 调 0 次 = 短路。不转发、直接返回一条消息，内层所有逻辑与模型调用全被跳过。这是关键词拦截、限速、缓存命中直接返回的实现方式：

     from langchain_core.messages import AIMessage
     
     class ConditionalShortCircuitMW(AgentMiddleware):
         def wrap_model_call(self, request, handler):
             if self.trigger_keyword in user_text.lower():
                 return AIMessage(content=self.blocked_reply)   # 不调内层
             return handler(request)
     

   • 调 1 次 = 正常流程。

   • 调 N 次 = 重试或冗余采样，内层模型真跑 N 遍、外层只看到一次包裹：

     result1 = handler(request)
     result2 = handler(request)
     return result2   # 以最后一次为准
     

6. 用观察族叫停整个循环。声明可跳转目标，运行时往状态写跳转字段即可：

   from langchain.agents.middleware.types import hook_config
   
   class EarlyStopMiddleware(AgentMiddleware):
       def __init__(self):
           self.call_count = 0
   
       @hook_config(can_jump_to=["end"])      # 必须声明，否则不生成条件边
       def before_model(self, state, runtime):
           self.call_count += 1
           if self.call_count >= 2:
               return {"jump_to": "end"}
           return None
   

7. 需要在状态里挂自己的字段时，静态声明状态结构，它会在构建期被合并进智能体状态：

   class MyCustomState(TypedDict):
       messages: Annotated[list, add_messages]
       visit_count: int
   
   class StateInspectorMW(AgentMiddleware):
       state_schema = MyCustomState
   
       def before_agent(self, state, runtime):
           return {"visit_count": 0}
   
       def before_model(self, state, runtime):
           return {"visit_count": state.get("visit_count", 0) + 1}
   

校验回路

• 钩子触发：计数器在一次任务后大于 0。
• 落图形态对照：覆盖观察族钩子的中间件会以「类名.钩子名」出现在执行图节点列表里；只覆盖包裹族钩子的不会出现在节点列表里（它是对已有节点的闭包包裹），但运行时打印照样出现。两者都生效，只是挂载机制不同。
• 触发顺序：一次简单任务的打印序列应为启动钩子、模型前钩子、包裹进站、包裹出站、模型后钩子、结束钩子——两层书挡，外层裹住整次运行、内层裹住每次模型调用。
• 多件包裹族中间件的洋葱顺序：列表第 0 个是最外层，序列为外层进站、内层进站、内层出站、外层出站。内层入口不是「直接打模型的固定函数」，而是下一层的入口。
• 效果证据要看最终输出的差异，不只是钩子被调用：挂注入中间件的一组输出风格明显不同；裁剪掉某工具后，需要该工具的问题拿不到答案，而不裁剪的一组能正常给出。
• 自定义状态字段在多轮后仍能累加取到，说明状态结构合并生效。

常见陷阱

• 包裹族钩子忘了转发：智能体会卡住或报错。这不是框架的规范要求，是闭包链的物理现实——你不往下传，下面就不会执行。观察族没有这条义务，它只读写状态。
• 用直接属性赋值改请求：该写法已废弃、会触发警告。一律用请求对象的覆写方法，并把返回的新对象传给内层。
• 写了跳转字段却没生效：缺了可跳转目标的声明，图在编译时不会生成对应的条件边，写进状态也会被忽略。另外这个字段走的是用完即焚的通道，不污染后续轮。
• 测跳转或规划类钩子时任务没给触发机会：跳转需要至少两次模型调用才有第二次机会。一次模型调用只返回一条消息，不挂工具的纯自然语言多步任务会被模型一口气答完、只触发一次。可靠做法是定义一个返回数据的工具、要求先调工具再总结，工具返回后还需再调一次模型消化结果，第二次机会才会到来。这类问题不是中间件写错了，是任务设计没给它触发条件。
• 记成有工具调用后钩子：基类没有这个钩子，包裹工具调用的正确名字是包裹族的那个。
• 数不清工具从哪来：状态字段与工具集都是构建期由所有中间件的静态声明合并汇总的结果。传 1 个自己的工具时运行期工具集是 9 个，内置 8 个的来源是规划中间件 1 个、子智能体中间件 1 个、文件系统中间件 6 个。执行命令的那个工具默认不在其中，它是条件性注入的、只有配了具备执行能力的后端才出现。