🎯 课程主题

基于类（Class）继承方式实现 Node-style 钩子函数（Hook），替代装饰器（Decorator）方式完成中间件（Middleware）的自定义开发。

📝 核心知识点

1. 类方式的核心规则

规则说明
继承父类必须继承 AgentMiddleware
方法名固定重写父类的四个钩子方法：before_model、after_model、before_agent、after_agent
类名随意自定义，建议见名知意
底层原理LangGraph 通过判断是否继承于 AgentMiddleware、是否有这些固定方法来识别钩子

2. 四个固定的钩子方法

  • before_model(self, state, runtime) — 调用大模型之前执行
  • after_model(self, state, runtime) — 调用大模型之后执行
  • before_agent(self, state, runtime) — Agent 整体执行前执行
  • after_agent(self, state, runtime) — Agent 整体执行后执行

3. 返回值的影响

返回值效果
None不影响主流程行为
dict（如 {"count": 10}）更新 state 中的状态
包含 jump_to跳转到指定节点，影响流程控制

jump_to 可选目标："end"（结束 Agent）、"tools"（工具节点）、其他自定义节点。

4. 装饰器与类方式的统一

装饰器本质上也是底层构建一个 AgentMiddleware 的子类实例。比如 @after_model 装饰器：

  • 底层通过 wrap 内部装饰器构建
  • 生成的子类包含 state_schema 和 tools 两个属性
  • 一个装饰器修饰的钩子函数 → 一个中间件实例（仅含一个钩子方法）

5. 关键参数

  • state — 包含消息列表（messages）
  • runtime — 包含运行时上下文信息

🏗️ 架构与工作流

AgentMiddleware (父类)
    ├── before_agent()    ← 重写
    ├── before_model()    ← 重写
    ├── after_model()     ← 重写
    └── after_agent()     ← 重写

执行顺序：
  before_agent → before_model → [模型调用] → after_model → after_agent

LangGraph 底层流程：Agent 基于 StateGraph 构建，中间件作为节点（Node）插入到工作流中，通过检查继承关系和方法签名来路由执行。

💻 代码实战

基于类定义中间件

from langchain.agents.middleware import AgentMiddleware

class MyMiddleware(AgentMiddleware):
    """自定义中间件：重写四个钩子方法"""

    def before_agent(self, state, runtime):
        print("=== before_agent 执行 ===")
        return None

    def before_model(self, state, runtime):
        print("=== before_model 执行 ===")
        return None

    def after_model(self, state, runtime):
        print("=== after_model 执行 ===")
        return None

    def after_agent(self, state, runtime):
        print("=== after_agent 执行 ===")
        return None

创建 Agent 并传入中间件实例

from langchain.agents import create_agent

# 实例化中间件
my_middleware = MyMiddleware()

# 创建 Agent，传入中间件实例列表
agent = create_agent(
    model=model,
    tools=tools,
    middleware=[my_middleware]
)

# 调用
response = agent.invoke({"messages": [HumanMessage(content="你好")]})

for msg in response["messages"]:
    msg.pretty_print()

输出顺序（验证执行流程）

=== before_agent 执行 ===
=== before_model 执行 ===
[模型调用的输出...]
=== after_model 执行 ===
=== after_agent 执行 ===

返回值影响状态示例

def before_model(self, state, runtime):
    # 从状态中读取 count，修改后返回以更新状态
    return {"count": state.get("count", 0) + 1}

跳转控制示例

def before_model(self, state, runtime):
    if some_condition:
        return {"jump_to": "end"}       # 直接结束
        # return {"jump_to": "tools"}   # 跳到工具节点
        # return {"jump_to": "custom"}  # 跳到自定义节点
    return None

⚠️ 常见问题与避坑指南

问题说明
忘记加 self类方法必须带 self 参数，否则调用时报参数数量错误
方法名写错必须是 before_model / after_model / before_agent / after_agent，拼写错误将不会被识别
忘记继承不继承 AgentMiddleware 则 LangGraph 不会将其识别为中间件
装饰器残留从装饰器方式迁移到类方式时，记得删除 @before_model 等装饰器
返回值类型返回字典会更新 state；返回 None 不影响；返回带 jump_to 的字典会影响流程

💡 个人总结与延伸

  • 装饰器方式 = 便捷写法：每个装饰器修饰的函数底层生成一个只包含单一钩子的 AgentMiddleware 子类实例
  • 类方式 = 原生写法：可以同时在一个类中定义多个钩子方法，适合逻辑聚合
  • 类比 Node.js 中间件：before_* / after_* 风格与 Express/Koa 的中间件洋葱模型一致
  • before_model / after_model 典型应用场景：
    • 消息修剪（Token 截断）
    • 敏感信息脱敏（PII Masking）
    • 输入/输出验证
    • 条件路由
    • 状态更新与统计
最近更新:
Contributors: yangwenguang