🎯 课程主题
使用装饰器（@decorator）方式定义 Node-style 钩子函数，并将其挂载到 Agent 上实现自定义中间件。
📝 核心知识点
1. 两种自定义中间件的方式
- 概念说明：
- 装饰器方式（函数式挂载）：用
@before_model等装饰器修饰普通函数，直接传入middleware列表。 - 类方式：继承
AgentMiddleware基类，在类中定义钩子方法。
- 装饰器方式（函数式挂载）：用
- 关键细节：本节课聚焦装饰器方式，这是最简洁的自定义中间件实现方式。
2. 四个 Node-style 装饰器
- 概念说明：LangChain 提供四个装饰器用于定义节点风格的钩子函数。
| 装饰器 | 触发时机 |
|---|---|
@before_agent | Agent 整体执行前 |
@before_model | 每次模型调用前 |
@after_model | 每次模型调用后 |
@after_agent | Agent 整体执行结束后 |
3. 钩子函数的签名规范
- 概念说明：钩子函数必须遵循固定的参数和返回值规范。
- 关键细节：
- 参数：
state: AgentState（维护 Agent 运行状态，包含消息列表）、runtime: Runtime（维护上下文环境，包括长期记忆）。 - 返回值：返回
None表示不干预主流程，不返回任何修改结果。 - 如果用类方式定义，第一个参数是
self；装饰器方式定义的普通函数不需要self。
- 参数：
4. state 与 runtime 的作用
- 概念说明：
state（AgentState 实例）：维护 Agent 运行过程中的状态，包含完整的消息列表，可在钩子中读取/修改消息。runtime（Runtime 实例）：维护 Agent 运行过程中的上下文环境，包括长期记忆等信息。
- 关键细节：演示中通过
state获取消息列表，在最后一条消息内容后追加标记文本（如 "before_model"），以验证钩子确实在预期时机被调用。
5. 执行顺序验证
- 概念说明：通过在不同钩子中写入不同标记，运行后观察输出顺序。
- 关键细节：实际输出顺序为：
before_agent→before_model→ [模型调用] →after_model→after_agent。
🏗️ 架构与工作流
装饰器方式定义钩子 → 放入 middleware 列表 → create_agent → invoke
┌───────────────────────────────────────────────┐
│ Agent 执行流程（含钩子） │
│ │
│ [@before_agent] 钩子执行 │
│ │ │
│ ▼ │
│ [@before_model] 钩子执行 │
│ │ │
│ ▼ │
│ ★ LLM 模型调用 ★ │
│ │ │
│ ▼ │
│ [@after_model] 钩子执行 │
│ │ │
│ ▼ │
│ [@after_agent] 钩子执行 │
└───────────────────────────────────────────────┘
💻 代码实战
from langchain.agents import create_agent
from langchain.agents.middleware import (
before_model,
after_model,
before_agent,
after_agent,
)
from langchain.agents.middleware.types import AgentState, Runtime
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from typing import Any
model = ChatOpenAI(model="gpt-4o")
# ── 基于装饰器定义 Node-style 钩子函数 ──
@before_model
def before_model_hook(state: AgentState, runtime: Runtime) -> None:
"""模型调用前执行"""
messages = state.get("messages", [])
if messages:
last_msg = messages[-1]
# 在最后一条消息内容后追加标记，验证钩子执行
if hasattr(last_msg, "content") and last_msg.content:
last_msg.content += "\n[before_model 标记]"
return None
@after_model
def after_model_hook(state: AgentState, runtime: Runtime) -> None:
"""模型调用后执行"""
messages = state.get("messages", [])
if messages:
last_msg = messages[-1]
if hasattr(last_msg, "content") and last_msg.content:
last_msg.content += "\n[after_model 标记]"
return None
@before_agent
def before_agent_hook(state: AgentState, runtime: Runtime) -> None:
"""Agent 执行前"""
messages = state.get("messages", [])
if messages:
last_msg = messages[-1]
if hasattr(last_msg, "content") and last_msg.content:
last_msg.content += "\n[before_agent 标记]"
return None
@after_agent
def after_agent_hook(state: AgentState, runtime: Runtime) -> None:
"""Agent 执行后"""
messages = state.get("messages", [])
if messages:
last_msg = messages[-1]
if hasattr(last_msg, "content") and last_msg.content:
last_msg.content += "\n[after_agent 标记]"
return None
# ── 创建 Agent，挂载钩子 ──
agent = create_agent(
model=model,
tools=[],
middleware=[
before_model_hook,
after_model_hook,
before_agent_hook,
after_agent_hook,
],
)
# ── 调用 ──
response = agent.invoke({
"messages": [HumanMessage(content="你好")]
})
# 遍历输出消息，观察钩子添加的标记
for msg in response.get("messages", []):
print(msg.pretty_print() if hasattr(msg, "pretty_print") else msg)
# ── 预期输出顺序 ──
# [before_agent 标记] ← 最先执行
# [before_model 标记] ← 模型调用前
# (模型实际响应内容)
# [after_model 标记] ← 模型调用后
# [after_agent 标记] ← 最后执行
⚠️ 常见问题与避坑指南
- 钩子函数必须返回
None，返回其他值会干扰 Agent 主流程，可能导致异常。 - 装饰器方式定义的钩子函数不含
self参数，如果从类方法中复制签名记得删除self。 - 需要从
langchain.agents.middleware导入装饰器和AgentState、Runtime类型。 - 钩子中修改
state（如追加消息内容）是原地修改，会影响后续流程中的状态，使用时需谨慎。
💡 个人总结与延伸
装饰器方式是定义 Node-style 钩子最简洁的方式——不需要写类、不需要继承基类，只需用 @before_model 等装饰器标记普通函数并返回 None。四个 Node-style 钩子覆盖了 Agent 生命周期的关键节点，在实际开发中可根据需要选择挂载哪些。函数内部的 state 参数可读取和修改消息列表，是实现日志记录、内容过滤、Token 统计等功能的切入点。