09 - 钩子与生命周期：精细控制 Agent 行为

一句话总结：用 pre_hooks 在 Agent 处理前拦截输入，用 post_hooks 在响应后做验证和记录，用 tool_hooks 给工具调用加中间件 – 像 Web 框架的 middleware 一样控制 Agent 的每个阶段。

什么是 Hooks

你写过 Express 或 FastAPI 的中间件吗？Hooks 的思路一模一样。

Agent 执行一次请求，大致经过这几个阶段：

用户输入 --> [pre_hooks] --> 模型处理 --> 工具调用 --> [post_hooks] --> 返回响应
                                            |
                                      [tool_hooks]

你可以在每个节点插入自己的逻辑：记日志、做校验、收集指标、写审计记录。Hooks 不会改变 Agent 的核心行为，但能让你对整个执行过程了如指掌。

三种 Hook 各司其职：

Hook 类型	触发时机	典型用途
`pre_hooks`	模型处理之前	输入日志、输入校验、审计
`post_hooks`	响应生成之后	输出校验、指标收集、通知
`tool_hooks`	每次工具调用前后	计时、限流、工具调用日志

pre_hooks：在处理前拦截

最简单的用法 – 记录每次用户输入：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.run.agent import RunInput
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def log_input(run_input: RunInput) -> None:
    """记录每次输入 - 用于日志和监控"""
    logger.info(f"用户输入: {run_input.input_content_string()[:100]}")


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    pre_hooks=[log_input],
    instructions="你是一个友好的中文助手",
)
agent.print_response("你好，请介绍一下你自己", stream=True)

运行后你会在日志里看到用户的输入内容被记录下来，然后 Agent 正常回复。

关键点：pre_hooks 函数的参数名有讲究。Agno 会根据你函数签名里的参数名自动注入对应的值。可用的参数包括：

run_input – 用户输入（RunInput 对象）
agent – Agent 实例本身
session – 当前会话
run_context – 运行上下文

你不需要全部接收，只写你需要的就行，Agno 会自动过滤。

post_hooks：响应之后做处理

post_hooks 在 Agent 生成响应之后、返回给调用者之前执行。典型场景是收集指标或验证输出质量：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.run.agent import RunOutput
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def log_metrics(run_output: RunOutput) -> None:
    """记录 Agent 的输出指标"""
    if run_output.metrics:
        logger.info(f"模型: {run_output.model}")
        logger.info(f"指标: {run_output.metrics}")


def check_response_length(run_output: RunOutput) -> None:
    """检查输出是否太短"""
    content = str(run_output.content) if run_output.content else ""
    if len(content) < 10:
        logger.warning("输出太短，可能有问题")
    else:
        logger.info(f"输出长度: {len(content)} 字符")


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    post_hooks=[log_metrics, check_response_length],
    instructions="你是一个友好的中文助手",
)
agent.print_response("用三句话解释什么是机器学习", stream=True)

post_hooks 可以传多个，按列表顺序依次执行。上面的例子先记录指标，再检查长度。

post_hooks 函数可用的参数：

run_output – Agent 的输出（RunOutput 对象，包含 content、metrics 等）
agent – Agent 实例
session – 当前会话
run_context – 运行上下文

tool_hooks：给工具调用加中间件

tool_hooks 和前两个不同，它是包裹在每次工具调用外面的中间件。每个 hook 接收工具名、一个 func 回调和参数，你必须调用 func(**args) 才能让工具真正执行 – 这就给了你在工具执行前后插入逻辑的机会。

import time
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools


def timing_hook(function_name: str, func: callable, args: dict):
    """测量每个工具调用的耗时"""
    start = time.time()
    result = func(**args)
    elapsed = time.time() - start
    print(f"[timing] {function_name} 耗时 {elapsed:.3f}s")
    return result


def logging_hook(function_name: str, func: callable, args: dict):
    """记录工具调用信息"""
    print(f"[tool] 调用 {function_name}，参数: {list(args.keys())}")
    return func(**args)


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[DuckDuckGoTools()],
    tool_hooks=[logging_hook, timing_hook],
    instructions="你是一个中文搜索助手",
    markdown=True,
)
agent.print_response("Agno AI 框架是什么？", stream=True)

tool_hooks 的执行顺序是嵌套的，类似洋葱模型：列表中第一个 hook 最外层，最后一个最内层。上面的例子中，logging_hook 先执行，它调用 func(**args) 时会进入 timing_hook，timing_hook 再调用 func(**args) 才真正执行工具。

注意：你必须调用 func(**args) 并返回其结果，否则工具不会执行，Agent 拿不到工具的返回值。

组合使用：搭建监控系统

实际项目里，你往往会把几种 hook 组合起来，搭建一套完整的监控和审计体系：

import time
import json
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.run.agent import RunInput, RunOutput
from agno.tools.duckduckgo import DuckDuckGoTools


# --- pre_hook：审计追踪 ---
def audit_input(run_input: RunInput) -> None:
    """记录所有输入到审计日志"""
    entry = {
        "type": "input",
        "time": time.strftime("%Y-%m-%d %H:%M:%S"),
        "content": run_input.input_content_string()[:200],
    }
    with open("audit.log", "a") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    print(f"[audit] 输入已记录")


# --- post_hook：质量检查 ---
def quality_check(run_output: RunOutput) -> None:
    """基础质量检查"""
    content = str(run_output.content) if run_output.content else ""
    if len(content) < 20:
        print("[quality] 警告：输出太短，可能有问题")
    else:
        print(f"[quality] 输出正常，{len(content)} 字符")


# --- tool_hook：工具计时 ---
def tool_timer(function_name: str, func: callable, args: dict):
    """记录工具调用耗时"""
    start = time.time()
    result = func(**args)
    elapsed = time.time() - start
    print(f"[timer] {function_name}: {elapsed:.3f}s")
    return result


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[DuckDuckGoTools()],
    pre_hooks=[audit_input],
    post_hooks=[quality_check],
    tool_hooks=[tool_timer],
    instructions="你是一个中文搜索助手，回答要详细",
    markdown=True,
)
agent.print_response("Python 3.12 有什么新特性？", stream=True)

这样一套下来，你能清楚看到：输入是什么、工具调用花了多久、输出质量如何。生产环境里把 print 换成正经的日志系统就行。

Hooks vs Guardrails：别搞混了

你可能注意到了，pre_hooks 和 post_hooks 不仅能放普通函数，还能放 Guardrail（守卫）对象。它们都用同一个参数传入，但职责不同：

	Hooks（普通函数）	Guardrails
目的	观察和记录（日志、指标、审计）	拦截和阻断（安全校验、内容过滤）
能否阻止执行	不应该（虽然技术上可以抛异常）	可以，抛出 `InputCheckError` 或 `OutputCheckError`
典型用法	记日志、收集指标、发通知	检查输入是否安全、输出是否合规

举个 Guardrail 阻断的例子，对比看一下：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.run.agent import RunInput
from agno.exceptions import InputCheckError, CheckTrigger


def block_sensitive_topics(run_input: RunInput) -> None:
    """作为 guardrail 使用 - 阻断敏感话题"""
    content = run_input.input_content_string().lower()
    blocked_words = ["密码", "信用卡号", "身份证"]
    for word in blocked_words:
        if word in content:
            raise InputCheckError(
                f"检测到敏感信息：{word}，请勿在对话中提供个人隐私数据",
                check_trigger=CheckTrigger.INPUT_NOT_ALLOWED,
            )


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    pre_hooks=[block_sensitive_topics],
    instructions="你是一个中文助手",
)

# 正常请求 - 通过
agent.print_response("今天天气怎么样？")

# 包含敏感信息 - 被拦截
try:
    agent.print_response("我的信用卡号是 1234-5678-9012-3456")
except InputCheckError as e:
    print(f"请求被拦截: {e}")

核心区别：普通 hook 安安静静做自己的事，guardrail 会大喊一声”不行”然后把请求挡回去。它们可以共存在同一个 pre_hooks 列表里。

参数自动注入的小细节

Agno 的 hook 系统有个很贴心的设计：它会检查你函数的签名，只传入你声明了的参数。这意味着你可以写很简单的 hook，也可以写很复杂的：

# 极简 - 只关心输入
def simple_hook(run_input: RunInput) -> None:
    print(f"收到: {run_input.input_content_string()[:50]}")

# 完整 - 拿到所有上下文
def detailed_hook(run_input: RunInput, agent: Agent, session, run_context) -> None:
    print(f"Agent: {agent.name}")
    print(f"Session: {session.session_id}")
    print(f"输入: {run_input.input_content_string()[:50]}")

两种写法都能直接放进 pre_hooks，不会报参数不匹配的错误。

今天学了什么

回顾一下关键概念：

pre_hooks 在模型处理前执行，用于记录输入、做前置校验
post_hooks 在响应生成后执行，用于记录输出、验证质量、收集指标
tool_hooks 包裹每次工具调用，像洋葱模型一样嵌套执行，必须调用 func(**args) 传递控制权
Hook 函数的参数名决定了注入什么，只写你需要的参数即可
Guardrails 和 Hooks 共用同一个参数（pre_hooks/post_hooks），但 Guardrails 能通过抛异常阻断执行，Hooks 更适合做观察和副作用
三种 hook 可以自由组合，搭建出完整的监控、审计、质量保障体系

下一篇预告

10 - 多 Agent 协作：Team 的四种模式

一个 Agent 干不完所有事。下一篇我们看看怎么把多个 Agent 组成团队，让它们分工协作 – Agno 提供了四种 Team 模式，从简单的轮询到智能的路由，覆盖了大部分多 Agent 场景。

上一篇安全护栏：给 Agent 加上"红线" 下一篇多 Agent 协作：Team 的四种模式