Agent 中间件生命周期：五层钩子与装饰器模式

Agent 中间件生命周期：五层钩子与装饰器模式

学习目标

本章要解决什么问题：在 Agent 执行过程中，如何在不修改核心逻辑的情况下插入重试、缓存、动态提示词、人类审批等横切关注点？

读者将学到：

• AgentMiddleware 五层钩子的职责和适用场景
• 装饰器模式如何简化中间件创建
• can_jump_to 条件跳转机制
• 中间件的组合顺序和执行流程

前置知识

本章涉及中间件链的通用原理，建议先阅读：

• 工具调用中间件链

下文假设你已理解上述概念，直接聚焦 LangChain 的具体实现。

项目实践

五层钩子全景

LangChain v1 的 AgentMiddleware 基类定义了六个生命周期钩子，覆盖 Agent 从启动到结束的全过程：

每个钩子都有同步和异步两个版本（如 before_agent / abefore_agent），框架根据 Agent 调用方式（invoke() vs ainvoke()）自动选择。

钩子职责详解

钩子	输入	返回值	典型场景
before_agent	状态 + Runtime	状态更新	初始化变量、发送启动事件
before_model	状态 + Runtime	状态更新	动态修改消息列表、注入上下文
wrap_model_call	ModelRequest + handler	ModelResponse	重试、模型回退、缓存、改写响应
after_model	状态 + Runtime	状态更新	验证输出、记录日志、流事件
wrap_tool_call	ToolCallRequest + handler	ToolMessage	工具重试、缓存、权限检查
after_agent	状态 + Runtime	状态更新	清理资源、发送完成事件

装饰器快捷模式

LangChain 提供装饰器，将普通函数直接转换为中间件，无需继承类：

# 伪代码：装饰器创建中间件
@before_model(can_jump_to=["end"])
def check_token_limit(state, runtime) -> dict | None:
    """消息过多时提前结束。"""
    if len(state["messages"]) > 50:
        return {"jump_to": "end"}
    return None

@dynamic_prompt
def context_aware_prompt(request) -> str:
    """根据对话长度动态调整系统提示词。"""
    msg_count = len(request.state["messages"])
    if msg_count > 10:
        return "对话较长，请简洁回答。"
    return "你是一个有用的助手。"

@wrap_model_call
def retry_on_error(request, handler):
    """模型调用失败时重试。"""
    for attempt in range(3):
        try:
            return handler(request)
        except Exception:
            if attempt == 2:
                raise

@wrap_tool_call
def cache_tool_calls(request, handler):
    """工具结果缓存。"""
    cache_key = hash(request.tool_call)
    if cached := get_cache(cache_key):
        return ToolMessage(content=cached, tool_call_id=request.tool_call["id"])
    result = handler(request)
    save_cache(cache_key, result.content)
    return result

装饰器内部机制：装饰器使用 type() 动态创建 AgentMiddleware 子类，将函数绑定到对应的钩子方法上。这避免了样板代码，同时保持了与类定义中间件的完全兼容性。

中间件组合与执行顺序

多个中间件按注册顺序串联，第一个注册的成为最外层：

# 伪代码：中间件注册
agent = create_agent(
    model="openai:gpt-5.5",
    middleware=[
        retry_middleware,       # 最外层
        cache_middleware,       # 中间层
        hitl_middleware,        # 最内层
    ]
)

执行流程：

retry_middleware.wrap_model_call
  └── cache_middleware.wrap_model_call
       └── hitl_middleware.wrap_model_call
            └── 实际模型调用

每个中间件通过 handler(request) 调用内层，可以：

• 前置处理：在 handler() 之前修改 request
• 后置处理：在 handler() 之后修改 response
• 短路：不调用 handler()，直接返回
• 多次调用：多次调用 handler() 实现重试

can_jump_to 条件跳转

before_model 和 after_model 钩子支持声明式跳转：

@hook_config(can_jump_to=["end", "model"])
def before_model(self, state, runtime) -> dict | None:
    if should_terminate(state):
        return {"jump_to": "end"}
    if needs_remodel(state):
        return {"jump_to": "model"}
    return None

框架根据 can_jump_to 列表在 StateGraph 中创建对应的条件边。未声明的跳转目标在编译时被拒绝，防止运行时出现无效的路由。

问题与规避

同步/异步不匹配

问题：只实现了 awrap_model_call（异步），但使用 invoke()（同步）调用 Agent。

对策：每个钩子都有清晰的错误提示，指导用户实现对应版本或使用匹配的调用方式。确保中间件同时实现同步和异步版本。

中间件名称冲突

问题：提供两个同名的中间件实例。

对策：create_agent 在编译时检查中间件名称唯一性，发现重复抛出 AssertionError。中间件默认使用类名作为名称，可通过 name 参数自定义。

中间件修改 tools 导致的 Unknown Tool 错误

问题：中间件在 wrap_model_call 中修改了 request.tools，添加了 Agent 未注册的工具，导致工具节点无法执行。

对策：当 Agent 存在 wrap_tool_call 中间件时，跳过工具名称校验（因为中间件可以通过 wrap_tool_call 处理动态工具）。否则，框架会抛出明确的错误信息并提供修复建议。

设计取舍

类继承 vs 装饰器

方案	优势	代价
类继承	支持复杂状态管理、多个钩子组合	样板代码多
装饰器	零样板代码、适合单一职责	每个装饰器函数只能实现一个钩子

LangChain 两者都支持：简单逻辑用装饰器，复杂中间件（需要多个钩子协作、维护内部状态）用类继承。

钩子返回值类型

钩子	返回类型	原因
before/after	`dict[str, Any]	None`
wrap_model_call	`ModelResponse	AIMessage
wrap_tool_call	`ToolMessage	Command`

返回值类型的差异反映了各钩子在 Agent 循环中的角色差异。

参考来源

• LangChain Agent Middleware 文档