Agno Agent Hooks 实战：pre_hooks、post_hooks 与 tool_hooks 拦截扩展机制详解

本文以 agno 仓库的 cookbook/02_agents/09_hooks 示例集为主体，系统讲解 Agno Agent 的三类钩子：pre_hooks（输入前拦截）、post_hooks(输出后校验) 和 tool_hooks（工具调用中间件）。文中完整保留了示例脚本的核心代码与运行方式，并结合 agent/_hooks.py、tools/function.py 等源码剖析钩子的参数注入、错误传播与中间件链式调用原理，帮助你在生产 Agent 中落地输入/输出安全校验、会话状态追踪与工具审计。

示例集总览：五类钩子场景

官方示例目录 cookbook/02_agents/09_hooks 覆盖以下场景：

示例文件	钩子类型	演示目的
post_hook_output.py	post_hooks	Agent 响应后的输出质量/安全校验
pre_hook_input.py	pre_hooks	Agent 处理输入前的综合输入校验
session_state_hooks.py	pre_hooks	在钩子中读写会话级状态（session state）
stream_hook.py	post_hooks	流式（streaming）运行生命周期中的通知钩子
tool_hooks.py	tool_hooks	包裹每次工具调用的中间件（计时、日志）
message_history_hooks.py	工具级 pre_hook / post_hook	通过 run_context.messages 访问当轮消息历史

注：最后一个是 README 的 Files 清单之外实际存在的补充示例，演示在单个工具的 @tool 装饰器上挂载钩子。所有示例在 TEST_LOG.md 中均有 2026-02-13 的 PASS 记录。

运行环境准备

按 README 的 Prerequisites 部分，运行这些示例需要：

1. 使用 direnv allow 加载环境变量（必须包含 OPENAI_API_KEY，示例统一使用 OpenAI 模型）。
2. 通过 ./scripts/demo_setup.sh 创建演示环境，之后用 .venvs/demo/bin/python 运行 cookbook 脚本。
3. 部分示例依赖可选的本地服务（如 pgvector）或特定供应商 API key；本目录五个示例主要依赖 OpenAI key。

运行方式统一为：

.venvs/demo/bin/python cookbook/02_agents/09_hooks/<file>.py

钩子在 Agent 上的挂载点：三个字段

在 Agent 模型定义中（agent/agent.py）可以看到钩子的三个挂载字段：

# A function that acts as middleware and is called around tool calls.
tool_hooks: Optional[List[Callable]] = None

# --- Agent Hooks ---
# Functions called right after agent-session is loaded, before processing starts
pre_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail, BaseEval]]] = None
# Functions called after output is generated but before the response is returned
post_hooks: Optional[List[Union[Callable[..., Any], BaseGuardrail, BaseEval]]] = None

从源码结构看有两点值得注意：

• pre_hooks / post_hooks 不仅接受普通可调用对象，还接受 BaseGuardrail、BaseEval 类型的守卫/评估对象，说明钩子机制与 agno 的 guardrail 体系是同一套拦截协议；
• 存在一个 _run_hooks_in_background 字段，由 AgentOS 设置，用于把非阻断性钩子放入后台任务执行（后文详述）。

钩子函数能收到哪些参数

钩子执行器（agent/_hooks.py 中的 execute_pre_hooks / aexecute_pre_hooks）会为每个钩子构造统一参数包：

all_args = {
    "run_input": run_input,        # RunInput：本次运行的输入
    "run_context": run_context,    # RunContext：会话状态、metadata、消息等
    "agent": agent,               # 当前 Agent 实例
    "session": session,           # AgentSession 会话对象
    "user_id": user_id,
    "debug_mode": debug_mode,
    "metadata": run_context.metadata,
}

随后通过 filter_hook_args（utils/hooks.py）按钩子函数的签名过滤参数——钩子函数只需要声明它关心的参数即可，这就是为什么示例中的钩子签名各不相同：有的只收 run_input，有的收 (run_output, run_context) 两个。

pre_hooks：输入前的综合校验

pre_hook_input.py 演示了用 pre-hook 对输入做「相关性 / 细节充分性 / 安全性」三维校验。其核心是一个「校验 Agent + 钩子函数」的组合模式：

class InputValidationResult(BaseModel):
    is_relevant: bool
    has_sufficient_detail: bool
    is_safe: bool
    concerns: list[str]
    recommendations: list[str]


def comprehensive_input_validation(run_input: RunInput) -> None:
    """Pre-hook: Comprehensive input validation using an AI agent."""
    validator_agent = Agent(
        name="Input Validator",
        model=OpenAIResponses(id="gpt-5-mini"),
        instructions=[
            "You are an input validation specialist. Analyze user requests for:",
            "1. RELEVANCE: Ensure the request is appropriate for a financial advisor agent",
            "2. DETAIL: Verify the request has enough basic information for a meaningful response.",
            "3. SAFETY: Ensure the request is not harmful or unsafe",
            "",
            "Be lenient with detail checks ...",
        ],
        output_schema=InputValidationResult,
    )

    validation_result = validator_agent.run(
        input=f"Validate this user request: '{run_input.input_content}'"
    )
    result = validation_result.content

    if not result.is_safe:
        raise InputCheckError(
            f"Input is harmful or unsafe. ...",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )
    if not result.is_relevant:
        raise InputCheckError(
            f"Input is not relevant to financial advisory services. ...",
            check_trigger=CheckTrigger.OFF_TOPIC,
        )
    if not result.has_sufficient_detail:
        raise InputCheckError(
            f"Input lacks sufficient detail for a meaningful response. ...",
            check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
        )

挂载到业务 Agent 上：

agent = Agent(
    name="Financial Advisor",
    model=OpenAIResponses(id="gpt-5-mini"),
    pre_hooks=[comprehensive_input_validation],
    instructions=["You are a knowledgeable financial advisor ..."],
)

示例给出三组典型拦截场景：模糊输入（"Help me invest"）、跑题请求（披萨做法）、潜在有害内容（操纵股价），均以 InputCheckError 抛出并附带 check_trigger 触发原因。

错误类型与触发枚举的源码依据

InputCheckError / OutputCheckError 定义于 exceptions.py，二者都携带 check_trigger（决定 error_id）、message 与可选的 additional_data。CheckTrigger 枚举提供了六类标准化触发原因，可作为拦截原因的结构化标识：

class CheckTrigger(Enum):
    OFF_TOPIC = "off_topic"
    INPUT_NOT_ALLOWED = "input_not_allowed"
    OUTPUT_NOT_ALLOWED = "output_not_allowed"
    VALIDATION_FAILED = "validation_failed"
    PROMPT_INJECTION = "prompt_injection"
    PII_DETECTED = "pii_detected"

从 agent/_hooks.py 的执行循环可以看到错误传播的关键设计：钩子抛出的 InputCheckError / OutputCheckError 会被 except (InputCheckError, OutputCheckError) as e: raise e 原样向上抛出（阻断运行），而其他异常仅记录日志（log_exception(f"Pre-hook #{i + 1} execution failed")）不中断流程——即「校验失败必须阻断，钩子自身 bug 不应拖垮 Agent」。另外同步 run() 下遇到协程函数钩子会打 warning 并跳过，异步钩子需配 arun() 使用。

post_hooks：响应后的输出校验

post_hook_output.py 演示两种粒度的输出校验，钩子接收 RunOutput（注意此时 run_output.content 已是模型最终响应）：

1）基于 LLM 的全面质量校验——用一个 gpt-5-mini 校验 Agent 输出结构化的判定结果：

class OutputValidationResult(BaseModel):
    is_complete: bool
    is_professional: bool
    is_safe: bool
    concerns: list[str]
    confidence_score: float


def validate_response_quality(run_output: RunOutput) -> None:
    if not run_output.content or len(run_output.content.strip()) < 10:
        raise OutputCheckError(
            "Response is too short or empty",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

    validator_agent = Agent(
        name="Output Validator",
        model=OpenAIResponses(id="gpt-5-mini"),
        instructions=[...],          # 校验 completeness / professionalism / safety
        output_schema=OutputValidationResult,
    )
    result = validator_agent.run(
        input=f"Validate this response: '{run_output.content}'"
    ).content

    if not result.is_complete or not result.is_professional or not result.is_safe:
        raise OutputCheckError(..., check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED)
    if result.confidence_score < 0.6:
        raise OutputCheckError(
            f"Response quality score too low ({result.confidence_score:.2f}). ...",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

2）轻量级长度校验——纯规则、零额外模型调用：

def simple_length_validation(run_output: RunOutput) -> None:
    content = run_output.content.strip()
    if len(content) < 20:
        raise OutputCheckError(
            "Response is too brief to be helpful",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )
    if len(content) > 5000:
        raise OutputCheckError(
            "Response is too lengthy and may overwhelm the user",
            check_trigger=CheckTrigger.OUTPUT_NOT_ALLOWED,
        )

示例主流程用三组测试验证：正常客服问答应通过校验；把指令改为 "Answer in 1-2 words only." 的 Brief Agent 会触发长度不足拦截；正常长度回答通过简单校验。捕获侧统一写成 except OutputCheckError as e: 并读取 e.check_trigger，方便上层按触发原因做降级或提示。

从 agent.py 字段注释看，post_hooks 的触发时机是 after output is generated but before the response is returned（输出已生成、响应返回前），因此被拦截的响应不会回传给用户，但本次运行已经消耗了模型调用——这是设计输出校验时需要考虑的成本点。

在 pre_hook 中读写会话状态

session_state_hooks.py 展示了钩子与持久化会话状态（session state）的配合：pre-hook 每次运行前调用一个「话题分析 Agent」，把用户消息中的主题累积进 run_context.session_state：

def track_conversation_topics(run_context: RunContext, run_input: RunInput) -> None:
    if run_context.session_state is None:
        run_context.session_state = {"topics": []}
    elif run_context.session_state.get("topics") is None:
        run_context.session_state["topics"] = []

    topics_analyzer_agent = Agent(
        name="Topics Analyzer",
        model=OpenAIResponses(id="gpt-5-mini"),
        output_schema=ConversationTopics,   # {"topics": list[str]}
    )
    response = topics_analyzer_agent.run(
        input=f"Extract the topics present in the following user message: {run_input.input_content}"
    )
    run_context.session_state["topics"].extend(response.content.topics)

agent = Agent(
    name="Simple Agent",
    model=OpenAIResponses(id="gpt-5-mini"),
    pre_hooks=[track_conversation_topics],
    db=SqliteDb(db_file="test.db"),   # 会话状态需要 db 才能跨运行持久化
)

agent.print_response(
    input="I want to know more about AI Agents.",
    session_id="topics_analyzer_session",
)
print(agent.get_session_state(session_id="topics_analyzer_session"))

要点：

• RunContext.session_state 是普通 Dict，定义见 run/base.py（同文件还定义了 metadata 与 messages 字段）；
• 钩子对 session_state 的修改会随会话持久化（示例配了 SqliteDb），下次同 session_id 运行时可直接读取累积结果；
• 钩子中嵌套运行另一个 Agent（topics analyzer）是合法的，_hooks.py 在 finally 块中专门 set_debug(agent, ...) 复位日志模式，注释说明就是为了应对「pre-hook 内部的 agent 改变了全局 debug 状态」这种情况。

流式运行下的钩子：stream 场景通知

stream_hook.py 演示在 stream=True 的异步运行中使用 post-hook：响应生成后根据运行 metadata 里的邮箱地址发送（模拟的）邮件通知。

def send_notification(run_output: RunOutput, run_context: RunContext) -> None:
    """Post-hook: Send a notification to the user."""
    if run_context.metadata is None:
        return
    email = run_context.metadata.get("email")
    if email:
        send_email(email, run_output.content)


agent = Agent(
    name="Financial Report Agent",
    model=OpenAIResponses(id="gpt-5-mini"),
    post_hooks=[send_notification],
    tools=[YFinanceTools()],
    instructions=[...],
)

await agent.aprint_response(
    "Generate a financial report for Apple (AAPL).",
    user_id="user_123",
    metadata={"email": "[e-mail]"},   # metadata 随 run() 传入，钩子中可读
    stream=True,
)

这个示例揭示了两个实用点：其一，run() 调用时传入的 metadata 字典会挂到 RunContext.metadata 上，钩子可据此获取调用方上下文（用户邮箱、渠道、租户等），实现「响应完成 → 触发外部通知」这类副作用；其二，post-hook 在流式与非流式两种模式下都会执行，钩子拿到的是完整的 RunOutput.content，而不是流式增量。

tool_hooks：包裹每次工具调用的中间件

tool_hooks.py 的模块 docstring 直接给出了中间件契约：

Tool hooks act as middleware: each hook receives the tool name, arguments, and a next_func callback. The hook must call next_func(**args) to continue the chain, and can inspect or modify args before and the result after.

def timing_hook(function_name: str, func: callable, args: dict):
    """Measure and print the execution time of each tool call."""
    start = time.time()
    result = func(**args)
    elapsed = time.time() - start
    print(f"[timing_hook] {function_name} took {elapsed:.3f}s")
    return result


def logging_hook(function_name: str, func: callable, args: dict):
    """Log the tool name and arguments before execution."""
    print(f"[logging_hook] Calling {function_name} with args: {list(args.keys())}")
    return func(**args)


agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[WebSearchTools()],
    tool_hooks=[logging_hook, timing_hook],   # 按 middleware 顺序应用于每次工具调用
    markdown=True,
)

中间件链是怎么构建的：源码级原理

tool_hooks 的落地实现在 tools/function.py。执行器把每个 hook 包成一个 wrapper，再用 functools.reduce 从内向外卷成链：

def create_hook_wrapper(inner_func, hook):
    """Create a nested wrapper for the hook."""

    def wrapper(name, func, args):
        # Pass the inner function as next_func to the hook
        # The hook will call next_func to continue the chain
        def next_func(**kwargs):
            return inner_func(name, func, kwargs)

        hook_args = self._build_hook_args(hook, name, next_func, args)
        return self._safe_hook_call(hook, hook_args)

    return wrapper

# Build the chain from inside out - reverse the hooks to start from the innermost
hooks = list(reversed(final_hooks))
chain = reduce(create_hook_wrapper, hooks, execute_entrypoint)

由此可以确认几个行为细节：

• 链式语义：tool_hooks=[A, B] 意味着 A 在外层、B 在内层，调用顺序为 A → B → 工具本体 → B → A，与典型 middleware 栈一致；
• 必须调用 next_func：钩子若不把 func(**args) 的返回值接住或干脆不调用，链即被截断；钩子可以在调用前改写 args、在返回后改写结果，这是做参数脱敏、结果审计、缓存、重试的基础；
• Agent 级 hook 自动下发到工具：agent/_tools.py 中可见 if agent.tool_hooks is not None: _func.tool_hooks = agent.tool_hooks 的赋值逻辑，Agent 上声明的 tool_hooks 会被注入到其挂载的每个函数工具上；
• 缓存命中也走钩子：源码注释明确 "A cache hit still runs tool_hooks and post_hook, so audit hooks see" 到缓存路径，保证审计类钩子的可见性；
• 异步钩子限制：同步执行路径下协程函数钩子会被跳过并打 warning（async 路径有对应的 aexecute 链），与 pre/post hooks 的行为一致。

工具级钩子与消息历史访问

除 Agent 级 tool_hooks 外，@tool 装饰器本身也支持钩子参数（tools/decorator.py：pre_hook / post_hook / tool_hooks）。message_history_hooks.py 演示了给单个工具挂 pre/post 钩子，并通过 run_context.messages 查看当轮消息历史：

def pre_hook(run_context: RunContext, fc: FunctionCall):
    msgs = run_context.messages
    count = len(msgs) if msgs else 0
    print(f"[pre-hook] {fc.function.name} - {count} messages in run")


def post_hook(run_context: RunContext, fc: FunctionCall):
    print(f"[post-hook] {fc.function.name} returned '{fc.result}' - {count} messages in run")


@tool(pre_hook=pre_hook, post_hook=post_hook)
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 72F in {city}"

与 Agent 级 pre/post hook 的区别在于：工具级钩子的入参是 RunContext 与 FunctionCall（可读到 fc.function.name、fc.result），作用域仅限被装饰的那个工具；从 run/base.py 的注释看，钩子收到的 messages 是浅拷贝，避免钩子直接改动运行中的消息列表。

进阶行为：守卫钩子与后台执行

阅读 agent/_hooks.py 与 utils/hooks.py 可以看到钩子系统还支持两组超出示例范围的能力，理解它们有助于正确选择钩子写法：

1. 全局后台模式：当 _run_hooks_in_background 为 True（由 AgentOS 设置）且存在后台任务队列时，guardrail 类型钩子（is_guardrail_hook 判定）仍同步执行——注释解释为 "Guardrails MUST block so InputCheckError/OutputCheckError can propagate"；其余非阻断钩子（日志、webhook 类）被 background_tasks.add_task 入队，并且只在所有 guardrail 通过后才入队，防止输入被拒绝时副作用（日志/通知）已经发出。
2. 单钩子后台标记：should_run_hook_in_background 检查钩子上由 @hook(run_in_background=True) 装饰器设置的 _agno_run_in_background 属性，配合 copy_args_for_background（对 run_input/session_state/metadata 做 deepcopy）避免后台任务与主流程的竞态。
3. 流式事件：开启 stream_events 时，每个 pre/post hook 会前后各发出 PreHookStartedEvent / PreHookCompletedEvent（create_pre_hook_started_event 等），前端可据此渲染「正在执行校验」等状态。
4. debug 模式复位：每个钩子执行完在 finally 中调用 set_debug，防止钩子内部嵌套 Agent 改变了主 Agent 的日志级别。

实践建议与文件索引

结合上述源码与示例，落地 agno hooks 时的几条经验：

• 安全校验用钩子 + 标准异常：抛出 InputCheckError / OutputCheckError 并指定 CheckTrigger（如 PROMPT_INJECTION、PII_DETECTED），上层即可按结构化 error_id 分流处理；
• 注意钩子内嵌套 Agent 的成本：示例中的「校验 Agent 模式」每次运行都多消耗一次 LLM 调用，可用「规则钩子 + 低成本低模型校验钩子」组合分层；
• 需要持久化数据进钩子就用 run_context.session_state + db，纯运行期数据放 metadata；
• 工具横切关注点（计时、审计、限流、脱敏）优先用 tool_hooks 中间件而非改每个工具实现，缓存路径同样会被覆盖。

延伸阅读路径（均为仓库相对路径）：

• 示例与测试记录：cookbook/02_agents/09_hooks/README.md、TEST_LOG.md
• 核心实现：agent/_hooks.py、agent/agent.py、exceptions.py、tools/function.py、tools/decorator.py、utils/hooks.py、run/base.py