LangChain 应用开发（十三）：自定义中间件与 Hook

目录一、回顾 Middleware 与 Hook1. 什么是 Hook2. LangChain Hook 的两大分类二、Node-style Hooks1. 什么是 Node-style Hook2. 使用装饰器定义 Hook3. 使用类定义三、深入理解 Node-style Hook1. Node-style Hook 能做什么2. 改变执行路径与 can_jump_to3. 代码案例四、Wrap-style Hooks1. Node-style 与 Wrap-style2. wrap_model_call掌控大模型调用3. wrap_tool_call掌控工具执行五、装饰器与类实现的统一1. 装饰器写法的本质2. 为什么提供两种方式六、装饰器还是 Middleware 类七、Hook 的执行顺序1. 单个 Middleware 的生命周期2. 多个 Middleware 的拦截顺序总结一、回顾 Middleware 与 Hook在上篇中我们拆解了 HumanInTheLoopMiddleware、PIIMiddleware 以及 TodoListMiddleware 等内置中间件。在不需要修改 LLM 提示词或工具内部代码的前提下完成了数据脱敏、人工审批与任务规划但面对特定业务逻辑时内置中间件显然不够用。要实现任意精细度的拦截就需要掌握自定义 Middleware 的内核原理Hook钩子1. 什么是 HookHook钩子是 Agent 在执行生命周期中暴露给开发者的拦截接口一个完整的 Agent 推导循环包含多个固定阶段接收用户输入、构建模型请求、调用 LLM、解析工具指令、执行工具调用以及更新状态记录。Hook 就像是在这条流水线的关键位置预留的插槽允许我们在特定时机插入自定义代码以读取、修改乃至改变Agent 的执行路径引入自定义 Middleware 的根本原因在于实现业务逻辑与底层 Agent 推导架构的解耦。无需改动模型配置或工具代码即可像挂载插件一样随插随用2. LangChain Hook 的两大分类LangChain 将 Hook 分为两套核心机制Node-style Hooks与Wrap-style HooksMiddleware │ ┌────────────┴────────────┐ ↓ ↓ Node-style Hooks Wrap-style Hooks │ │ 在特定节点执行 包裹一次调用 │ │ before / after ... wrap_model_call wrap_tool_callNode-style 更像 执行到这里时顺便做点什么Wrap-style 更像 把原来的执行过程整个包起来Node-style Hooks基于生命周期节点的单点拦截。Agent 执行到特定的状态节点时如模型调用前 before_model触发运行适合做日志记录、状态校验或控制跳转Wrap-style Hooks基于闭包与控制权移交的环绕拦截。它不只是在两端做加法而是将整个模型或工具的调用过程直接包裹Wrap起来由你的代码控制原始操作何时执行、执行几次如重试机制以及如何捕获异常二、Node-style Hooks与传统后端框架中的生命周期钩子类似Node-style Hooks作用于 Agent 图流程中的固定节点。当 Agent 执行流动到特定阶段时对应注册的 Hook 函数就会被自动触发1. 什么是 Node-style HookNode-style Hook 的核心在于节点控制。Agent 在单次调用或推导循环中按固定轨迹运行Hook 则插入在这些关键节点的前后按 Agent 执行生命周期排列常用 Hook 分为两类任务级钩子单次运行触发一次before_agentAgent 接收到用户请求、启动推导图之前执行适合用于上下文初始化或全局权限校验after_agentAgent 完成所有推导与工具执行并准备输出最终结果时执行适合清理资源或进行最终审计循环级钩子每次推导循环触发before_modelAgent 在将消息发送给 LLM 之前触发常用于动态修改 Prompt、删减过长上下文或拦截校验after_model大模型返回响应后立即触发常用于校验输出格式、捕获敏感内容或记录耗时2. 使用装饰器定义 Hook使用装饰器是实现单点拦截最轻量的方式。以下是 before_model 的典型用法from langchain.agents.middleware import before_model from langgraph.pregel.types import StateT from langgraph.runtime import Runtime from typing import Any, Optional before_model def log_and_guard(state: dict[str, Any], runtime: Runtime) - Optional[dict[str, Any]]: # 打印当前消息历史数量 messages state.get(messages, []) print(f[Log] 准备调用模型当前消息数量: {len(messages)}) # 若消息数量异常可注入预警信息 if len(messages) 20: return {system_notice: 上下文过多请注意压缩} return None理解装饰器定义的 Hook只需看透三个核心要素before_model这个函数什么时候执行挂载此装饰器后每次 Agent 循环准备发起 LLM 请求前该函数都会被自动调用state / runtime能拿到什么state当前的 Agent 状态字典如包含对话历史 messages 或自定义状态属性runtime当前 Agent 的运行环境上下文如配置参数、Thread ID、数据存储资源等return能修改什么返回dict该字典会通过状态图的 Reducer 机制合并更新到 Agent State 中返回None不做任何状态修改仅执行只读操作如日志上报、埋点返回控制指令更改控制流直接跳过后续节点3. 使用类定义当逻辑变复杂或需要跨节点协同时应使用继承 AgentMiddleware 的类来实现class MyMiddleware(AgentMiddleware): def __init__(self): super().__init__() def before_model(self, state: AgentState, runtime: Runtime) - dict[str, Any] | None: print( - before_model - ) return None def after_model(self, state: AgentState, runtime: Runtime) - dict[str,Any] | None: print( - after_model - ) return None agent create_agent( modelmodel, middleware[MyMiddleware()], ) agent.invoke({ messages: [ {role: user, content: 你好请介绍一下你自己} ] })输出结果无论采用装饰器写法还是类定义写法继承 AgentMiddleware 并重写 before_model 方法两者的底层机制完全等价——它们本质上都是向 Agent 系统的生命周期节点注册拦截逻辑三、深入理解 Node-style Hook理解 Node-style Hook 的关键在于不要把它当成死板的 参数手册而是明白它能为 Agent 注入什么能力1. Node-style Hook 能做什么归纳来看Node-style Hook 赋予了开发者以下几个维度的控制力Node-style Hook │ ├── 读取当前 State ────────► 监控/埋点/状态审计 │ ├── 修改 State ───────────► 注入动态 Prompt/更新元数据 │ ├── 修改上下文 ───────────► 裁剪过长 Messages 历史 │ └── 改变后续执行路径 ─────► 条件跳转/拦截阻断 (can_jump_to)2. 改变执行路径与 can_jump_to在常规的推导生命周期中Agent 严格按照预设链路运行before_model ──► Model 调用 ──► after_model ──► Tool 调用但在真实业务场景中中间件经常需要打破这一默认路径。例如缓存命中在 before_model 中发现用户提问已在 Redis 中缓存无需调用 LLM想直接输出答案并结束安全阻断在 before_model 中检测到违规输入需要跳过 LLM 和 Tool 执行直接跳转至结束阶段并返回提示这就是can_jump_to解决的痛点LangChain Agent 底层由LangGraph图引擎驱动。can_jump_to 本质上决定了该 Middleware 是否有权限修改 Agent 图模型的控制流 Edge条件边当中间件配置了 can_jump_to且 Hook 返回了特定的控制跳转信号时 Agent 图引擎将立即终止原本顺位执行的下一个节点直接将控制权移交给指定的受控节点。这极大地提升了 Agent 在应对复杂异常和业务分支时的灵活性3. 代码案例can_jump_to 接收一个列表用于在编译 Agent 控制图时显式声明允许跳转的目标节点从而建立条件边。目前支持的三个标准合法值包括end直接跳转至结束。跳过后续的大模型推导与工具执行直接终止流程或进入 after_agent 阶段。适用于安全拦截、触发表单熔断、命中缓存直接返回结果等场景tools直接跳转至工具节点。绕过 LLM 的思考与路由阶段直接将执行流移交给 Tool 执行节点。适用于基于规则的指令匹配model强制跳转回模型节点。重新触发 LLM 的推导循环。适用于修正上下文后要求模型重新推理在 Hook 中触发跳转时需要在装饰器声明 can_jump_to 的同时在 Hook 返回值字典中指定 jump_to 目标装饰器写法敏感词阻断与早退# 1. 显式声明该 Hook 允许跳转至 end 节点 before_model(can_jump_to[end]) def sensitive_word_guard(state: dict[str, Any], runtime: Runtime) - Optional[dict[str, Any]]: messages state.get(messages, []) if not messages: return None last_user_msg messages[-1].content # 检测到违规输入拦截并提前结束 if 退款账号密码 in last_user_msg: print([Guard] 检测到高风险提问强行拦截) return { # 注入提示消息 messages: [AIMessage(content安全提示请勿在对话框中输入账户密码信息。)], # 控制流指令直接跳过 LLM 和 Tool跳转至 end jump_to: end } return None类写法如果通过类继承 AgentMiddleware需要借助 hook_config 装饰器为具体的 Hook 方法标记 can_jump_to 权限class TokenQuotaMiddleware(AgentMiddleware): def __init__(self, max_messages: int 30): self.max_messages max_messages # 使用 hook_config 标记 can_jump_to 声明条件跳转路径 hook_config(can_jump_to[end]) def before_model(self, state: dict[str, Any], runtime: Runtime) - Optional[dict[str, Any]]: messages state.get(messages, []) # 超过最大对话轮数中断 Agent 并直接结束 if len(messages) self.max_messages: return { messages: [AIMessage(content已达到单次会话消息上限请发起新对话。)], jump_to: end } return None四、Wrap-style HooksWrap-style Hooks赋予了开发者完全的控制下发权不仅能观测前后更能决定原始调用是否执行、执行几次以及如何应对异常1. Node-style 与 Wrap-style从思维方式来看两者的拦截视角有着本质区别在 Wrap-style 中中间件函数不会被系统自动按顺位推着走而是拿到一个指向原始操作的句柄——handlerhandler 是什么handler(request)就是 通知系统按照原本的计划去真正发起 LLM 调用或执行 Tool为什么要包裹因为有了 handler 的控制权你可以在它外层套上 try-except捕获异常、while 循环失败重试、耗时计算计时器甚至在调用前直接修改 request 里的参数2. wrap_model_call掌控大模型调用wrap_model_call 专门用来环绕包裹 Agent 对 LLM 的底层请求wrap_model_call def custom_model_call(request: Any, handler: Any) - Any: # 1. Calling LLM 前记录耗时、打印/修改请求 start_time time.time() print(f[Model Wrap] 准备调用 LLM请求模型: {getattr(request, model, 未知)}) # 2. 核心由我们决定何时交出控制权发起真正调用 try: response handler(request) except Exception as e: print(f[Model Wrap] LLM 调用抛出异常: {e}) raise e # 3. Calling LLM 后审计返回结果、计算耗时 cost_time time.time() - start_time print(f[Model Wrap] LLM 调用完成耗时: {cost_time:.2f}s) return response agent create_agent( modelmodel, middleware[custom_model_call], ) agent.invoke({ messages: [ {role: user, content: 你好请介绍一下你自己} ] })输出结果片段Wrap-style 能在模型调用层实现的能力修改请求在调用 handler(request) 之前可以修改 request 中的 Prompt、Temperature 或调整传入的 tools 工具列表错误重试与降级当 handler(request) 抛出网络超时或限流异常时可以使用 for 循环重新调用或者在重试多次依然失败后切换备用模型请求响应改写与过滤在 handler(request) 返回 response 后可以在将其交还给 Agent 前进行格式修饰或敏感词二次过滤3. wrap_tool_call掌控工具执行与 wrap_model_call 对应wrap_tool_call 专门用于拦截和环绕 Agent 对 Tool 工具 的实际调用代码示例带重试与异常保护的 Tool 拦截器wrap_tool_call def safe_tool_executor(request: Any, handler: Any) - Any: tool_name getattr(request, name, unknown_tool) print(f[Tool Wrap] 准备执行工具: {tool_name}) # 设置最多重试 2 次 max_retries 2 for attempt in range(max_retries 1): try: # 移交控制权真正执行业务 Tool result handler(request) print(f[Tool Wrap] 工具 {tool_name} 执行成功) return result except Exception as e: print(f[Tool Wrap] 工具 {tool_name} 第 {attempt 1} 次执行报错: {e}) if attempt max_retries: # 达到重试上限后兜底避免系统直接 Crash return f工具 {tool_name} 执行失败错误信息: {str(e)} agent create_agent( modelmodel, middleware[safe_tool_executor], ) agent.invoke({ messages: [ {role: user, content: 你好请介绍一下你自己} ] })上一篇博客中提到的 ToolRetryMiddleware工具重试中间件与 ToolErrorMiddleware工具异常捕捉中间件它们在底层的实现原理正是基于 wrap_tool_call而 ModelFallbackMiddleware模型降级中间件则是基于 wrap_model_call五、装饰器与类实现的统一在前面的章节中我们分别展示了装饰器与类继承两种编写方式。它们在底层完全统一。装饰器并不是另一种机制而是 LangChain 提供的语法糖——它在运行时动态地将你的函数封装为一个 AgentMiddleware 实例1. 装饰器写法的本质当你写以下装饰器代码时before_model(can_jump_to[end]) def my_hook(state, runtime): ...LangChain 会在后台自动创建一个 AgentMiddleware 的匿名子类把 my_hook 函数绑定为该子类的 before_model 方法并根据装饰器参数生成对应的配置项。最终传入 create_agent(middleware[...]) 的依然是一个标准的 AgentMiddleware 对象类写法则是直接将这种结构显式表达出来class MyMiddleware(AgentMiddleware): hook_config(can_jump_to[end]) def before_model(self, state, runtime): ...无论选择哪种方式Agent 引擎在编译执行图时处理 Hook 的逻辑与执行顺序是完全一致的2. 为什么提供两种方式这种 单函数装饰器 完整类 的设计是为了平衡开发敏捷与完备性装饰器模式降低门槛适合轻量化、单一功能的 Hook例如临时打日志、单步参数改写。无需手写类模板代码几行函数加个装饰器就能即插即用类继承模式工程首选适合复杂场景。当 Middleware 需要接收初始化参数如重试次数、阈值、维持内部状态、同时定义同步与异步实现或者需要组合多个 Hook 方法时类结构能提供极佳的封装性与重用性六、装饰器还是 Middleware 类理解了底层机制后最直接的问题就是面对具体的业务需求到底该选装饰器还是 Middleware 类这取决于功能的复杂程度与是否需要跨 Hook 共享状态/配置选型逻辑优先选择装饰器当拦截逻辑足够单一、无状态时装饰器是最高效的选择。例如单步日志上报、单次 Prompt 追加、简单的输入参数合法性校验或快速代码实验。它无需编写类继承结构随手写个函数加上装饰器即可生效优先选择 Middleware 类当中间件涉及状态维护、配置注入或多钩子协同时继承 AgentMiddleware 是唯一优雅的方式。例如需要在 before_model 记录开始时间并在 after_model 计算耗时跨 Hook 传递变量、需要通过 __init__ 接收用户配置参数如重试次数、过滤规则、或者需要将拦截能力封装为独立的 Python 包供其他项目复用七、Hook 的执行顺序当我们在 Agent 中配置了多个 Middleware 时拦截逻辑并不是乱序执行的而是遵循着严格的生命周期顺序与嵌套结构1. 单个 Middleware 的生命周期在一个标准的 Agent 推导循环中各个 Hook 触发的先后顺序如下before_agent (全局初始化仅一次) │ ▼ ◄─────────────────────────────────────────────┐ before_model │ │ │ wrap_model_call (进入包裹) │ │ │ ├──► [ Model 实际调用 ] │ ( Agent 循环) │ │ after_model │ │ │ wrap_tool_call (进入包裹) │ │ │ └──► [ Tool 实际执行 ] ─────────────────────────┘ │ ▼ after_agent (全局收尾仅一次)2. 多个 Middleware 的拦截顺序当传入 create_agent(middleware[M1, M2, M3]) 注册多个中间件时执行顺序遵循洋葱模型请求进入 ──► [ M1 ] ──► [ M2 ] ──► [ M3 ] ──► [ 核心 Model / Tool ] │ 响应返回 ◄── [ M1 ] ◄── [ M2 ] ◄── [ M3 ] ◄────────────┘各类型的具体排序规则如下before_* 钩子正序按照列表传入顺序从头到尾依次执行M1.before_model - M2.before_model - M3.before_modelwrap_*​​​​​​​ 钩子嵌套最前端的中间件处于最外层包裹。M1 的 handler 指向 M2M2 指向 M3最后才到达真正的底座机制after_*​​​​​​​ 钩子逆序按照倒序先进后出触发M3.after_model - M2.after_model - M1.after_model总结回顾自定义 Middleware 的全部内核原理我们可以将所有知识归纳为三个维度的设计抉择1. 拦截时机Agent 级before_agent / after_agent全局一次性逻辑初始化、全流程审计Model 级before_model / after_model / wrap_model_call大模型推导前后或请求拦截Prompt 注入、限流、重试Tool 级wrap_tool_call工具调用环绕拦截入参校验、异常兜底2. 拦截方式Node-style基于状态节点接收 state / runtime返回更新字典或跳转指令Wrap-style基于闭包句柄接收 request / handler控制原始调用的执行次数与异常处理3. 实现形式轻量装饰器如 before_model单函数无状态拦截代码极简继承类AgentMiddleware维护内部配置与状态支持跨 Hook 协同适合工程化复用至此我们已经能够根据业务需求主动介入 Agent 的执行过程。下一篇将进入上下文与记忆学习 Agent 如何保存、管理和利用运行过程中的信息让智能体不仅能够执行任务还能够在多轮交互中持续利用上下文