Gemini 3 Hooks实战：给Agent生命周期装上拦截器

如果你已经跟到这份指南的第九篇，手上大概率已经有一个能跑通的 Agent 了——能调模型、能记住上下文、能做工具调用。但你可能跟我一样，在某个周五下午被一个需求恶心到：给线上的 Agent 加鉴权、加限流、加审计、加记忆注入……每一样都必须在业务主循环里插一段逻辑。插完之后整个 main loop 跟打满补丁的旧棉袄一样，改一个功能要连带测三遍。Gemini 3 的 Hooks 机制，就是为了把这堆"横切逻辑"从业务代码里拆出去，让你在 Agent 生命周期上直接挂拦截器，干净、可控、可观测。

这篇我会从生命周期本身讲起，拆解 Hooks 的事件模型、上下文传递、短路语义和异常行为，再带你把一个带鉴权、限流、记忆、审计、安全过滤的问答 Agent 完整跑通。后面还有我线上踩过的几个坑：并发串数据、Hook 超时把 P95 拖爆、重试导致重复日志。内容偏实战，建议打开编辑器跟着敲。

1. 为什么 Agent 框架需要 Hooks：从“屎山回调”到“生命周期拦截”

1.1 Agent 的一次完整旅程：生命周期到底是什么

先说清楚一个前提：Agent 不是一次性的 LLM 调用，而是一条有状态的事件链。以我常用的问答 Agent 为例，一次用户请求大致要经过这么几个阶段：

1. 请求进来，解析用户身份与会话信息
2. 组装上下文：系统提示词、历史消息、检索结果
3. 调用大模型，拿到响应
4. 如果模型决定调用工具，执行工具并回填结果，再回到第 3 步
5. 所有循环结束后，生成最终回复
6. 更新会话记忆，写审计日志，返回结果

这个过程中任何一步都可能抛异常，也随时可能被中断。所谓"生命周期拦截"，就是在第 1 步到第 6 步之间，给开发者留出若干个"钩子点"——你能在某个节点前后挂一段业务逻辑，这段逻辑可以观察、可以改写、也可以直接终止整个流程。

我见过不少团队把 Hooks 理解成"回调函数的集合"，这方向是对的，但不够。Hooks 的关键不是"能回调"，而是回调发生在固定的生命周期节点上，并且能共享同一个上下文对象。你写 React 时应该能感受到，Hooks 之所以好用，是因为它和组件渲染生命周期绑定死了，你不能随便在循环里调用它，也不能脱离渲染时机。Gemini 3 的 Agent Hooks 也是同样道理：注册时机、执行顺序、上下文传递都是框架约束好的，你只要往既定位置上挂逻辑。

1.2 没有 Hooks 的日子：横切逻辑如何变成屎山

我上一版 Agent 是用最原始的方式写的：主循环里裸奔，所有附加逻辑全靠硬编码。举个具体例子，当时要加三样东西：

• 鉴权：每个请求先检查 token，查用户表
• 限流：同一个 session 一分钟内最多 10 次提问
• 爬取敏感词：模型输出里不能出现手机号和身份证号

于是主循环代码长这样（伪代码）：

python复制async def run_agent(request):
    if not check_token(request.token):          # 鉴权
        return error_response("unauthorized")
    if not rate_limit(request.session_id):      # 限流
        return error_response("too many requests")
    context = build_context(request.messages)   # 组装上下文
    response = await llm.call(context)          # 模型调用
    response = mask_sensitive(response)         # 敏感词过滤
    await save_history(request.session_id, response)  # 存记忆
    return response

看起来还行？等需求迭代到第十个版本时就不是这样了。你要加 A/B 测试、要加模型路由、要加拨测日志、要加企业租户隔离……每加一个横切逻辑，主函数的行数就膨胀一次，而所有横切逻辑还在彼此竞争代码位置。更麻烦的是，只要有一处逻辑想提前 return，后面所有逻辑都受影响，改一个功能要回归整条链路。

后来我意识到一个本质问题：鉴权、限流、安全过滤、审计这四件事，本来就不该是 Agent 业务逻辑的一部分。它们是"环绕在 Agent 周围的系统能力"，应该长在 Agent 的外面，而不是长在里面。这就是 Hooks 机制要解决的问题。

1.3 中间件、装饰器、Hooks：为什么最终选了事件钩子

有人可能会说，这不就是中间件或者装饰器吗？确实，三者目标一致，都是把横切逻辑从核心业务里抽出来，但机制差别很大：

装饰器的问题是粒度太粗，你只能在函数外层包一层，没法精确地插在"llm 调用之后、工具调用之前"。中间件是常见的 Web 框架方案，但它天然是为请求/响应模型设计的，对 Agent 这种"内部有多轮循环"的执行体来说，颗粒度依然不够——你想在工具返回后、下一次模型调用前做点事情，中间件表达能力就很别扭。

Gemini 3 选择的是"生命周期事件钩子"：把 Agent 一次完整执行拆成一组有序事件，每个事件上有若干监听函数。Hook 可以拿到带全部运行状态的上下文对象，可以修改上下文来影响后续流程，也可以通过返回值语义提前中断。这比装饰器细、比中间件准、比自己在主循环里写判断条件清爽得多。

2. Gemini 3 Hooks 的核心机制拆解：事件、上下文与短路

2.1 生命周期事件全集：从 agent_start 到 agent_end

我基于自己项目里实际用到的版本，把核心事件整理成一张表。不同 SDK 版本名字可能略有差异，但这几个节点基本是所有 Agent 框架的公共骨架：

注意这里有几个容易忽略的点。

第一，llm_call_start 和 llm_call_end 在一次请求里可能触发多次——因为 Agent 存在"模型-工具"循环，只要模型请求调用工具，就会进入下一轮循环。如果你要写计费逻辑，必须以这两个事件为准做累加，而不是在 agent_start 里预估。

第二，agent_error 是全局兜底事件，它捕获的覆盖面很广。但它本身不会自动决定是否吞掉异常，具体行为取决于你怎么处理，后面我会专门讲。

第三，流式模式下，agent_end 的时机是在所有 chunk 都发完之后、最终响应封装之前。你仍然可以在 agent_end 里拿到完整文本摘要做后处理，前提是你在上下文里缓存了拼接结果。

2.2 注册的两种姿势：装饰器与注册中心

Gemini 3 提供两种注册 Hooks 的方式。我平时两种混着用：静态能力用装饰器，动态开关用注册中心。

装饰器写法，直观且不容易漏：

python复制from gemini_agent import Agent
from gemini_agent.hooks import HookContext

agent = Agent(name="qa-assistant", model="gemini-3-pro")

@agent.hook("agent_start")
async def on_agent_start(ctx: HookContext):
    # 做鉴权之类的逻辑
    if not ctx.request.token:
        ctx.abort("unauthorized")

注册中心写法，适合在运行时按配置动态挂载：

python复制agent.hooks.register("agent_end", write_audit_log, priority=10)
agent.hooks.register("agent_end", notify_webhook, priority=20)

装饰器和寄存器本质上是同一套底层，只是入口不同。个人经验：装饰器负责"这个 Agent 天生就有"的能力，注册中心负责"某个租户/某个场景才需要"的能力，这样代码读起来非常清晰，功能开关也能做到按需加载。

2.3 上下文对象：哪些字段能读，哪些字段能改

Hook 函数统一接收一个 HookContext 对象，这就是钩子和主流程之间的"会话单"。字段大体分三类：

我对比过不少 Agent 框架，Gemini 3 的上下文设计有一点做得特别好：它允许你在 Hook 里改写 messages 本身。这意味着 context_build 钩子里注入历史记忆，不是"拼接一段字符串"，而是直接往消息列表里插入条目，模型真的会把它当成对话的一部分，而不是一段生硬塞进去的文本。这一点对后续做动态记忆、行为约束非常重要。

需要提醒的是，run_id、agent_name、model_name 这类字段是只读的，SDK 做了保护，但你仍然能读取。真正要小心的是 metadata：它默认是一个空字典，框架不会自动往里塞东西，但它是 Hook 间传递数据的官方通道。我见过有人图省事用模块级全局变量传数据，并发一高立刻出问题，这个坑后面细说。

2.4 返回值语义：continue、break 与消息改写

Hook 不是"触发完就结束"那么简单，你通过返回值或上下文操作，能直接干预主流程方向。我实际使用中总结为三种语义：

• 放行：什么都不做，或返回 CONTINUE，流程继续往后走
• 短路：调用 ctx.abort(reason) 或返回 BREAK，流程立即终止，agent_end 和 agent_error 都不触发，直接向调用方返回 abort 原因
• 改写：修改 ctx.messages、ctx.system_prompt 等字段，后续节点拿到的就是改写后的数据

这里有一个至关重要的细节：短路与异常的差别。ctx.abort() 是"主动终止"，语义上是业务决定拒绝放行，比如鉴权失败；而异常上抛是"被动故障"，意味着系统出问题了。设计 Hooks 时一定要区分这两个场景，否则日志系统会被大量"exception"刷屏，而真正需要人为关注的故障反而被淹没了。

我项目里的约定是：鉴权失败、限流超限、敏感词触发，全部走 ctx.abort()，它们被记在业务日志里；真正的连接超时、模型限流异常、数据格式错误，才让异常往上抛，由 agent_error 接住并告警。

3. 实战：在问答 Agent 上挂满五个 Hook

3.1 目标拆解：鉴权、限流、记忆、审计、安全过滤

光讲机制太虚，我带你做一个完整例子。需求来自我线上真实场景的简化版：我要做一个内部问答助手，要求有五个横切能力——用户鉴权、调用限流、历史记忆注入、全链路审计、输出敏感信息脱敏。

如果用老办法，这五个逻辑全要写进主循环，代码会非常难维护。用 Hooks 的规划是：

3.2 Agent 骨架搭建与前置检查

先建一个最小 Agent，不挂任何 Hook，确保主流程能跑通：

python复制from gemini_agent import Agent

agent = Agent(
    name="qa-assistant",
    model="gemini-3-pro",
    system_prompt="你是一个严谨的问答助手。",
)

async def main():
    response = await agent.run(
        user_id="u_1001",
        session_id="s_2001",
        message="我的项目预算还剩多少？",
    )
    print(response.text)

这个阶段不要急着挂东西。先把 Agent 跑通，确认 SDK 版本、模型调用、基础输出正常。我见过太多人一上来就写十来个 Hook，结果主流程还没跑通，排查时根本分不清是 Hook 的问题还是模型的问题。

3.3 逐步挂 Hook：从请求进门到结果出门

第一步，挂鉴权和限流，都放在 agent_start。

python复制import time
from gemini_agent.hooks import HookContext

RATE_LIMIT_MAX = 10

@agent.hook("agent_start")
async def check_auth(ctx: HookContext):
    # ctx.request.token 是 SDK 从调用参数里透传进来的
    if ctx.request.token != "valid_token":
        ctx.abort("unauthorized", code=401)

@agent.hook("agent_start")
async def check_rate_limit(ctx: HookContext):
    # 用一个简单的内存计数器做演示，生产环境请用 Redis
    session_key = f"rate:{ctx.session_id}"
    current = ctx.get("rate_count", 0) + 1
    ctx.set("rate_count", current)
    if current > RATE_LIMIT_MAX:
        ctx.abort("rate_limit_exceeded", code=429)

注意这里我用 ctx.set("rate_count", ...) 而不是模块级全局变量，就是让状态跟着本次请求走，而不是跟着进程走。

第二步，挂记忆注入。关键是选 context_build 而不是 agent_start，因为此时消息列表还没有最终定型，你插入的内容能直接参与上下文组装：

python复制@agent.hook("context_build")
async def inject_memory(ctx: HookContext):
    memo = load_user_memo(ctx.user_id)  # 从数据库/缓存读取用户备注
    if memo:
        ctx.messages.insert(0, {
            "role": "system",
            "content": f"用户备注：{memo}。在回答时请参考这些偏好。"
        })

第三步，挂审计日志和安全脱敏。审计放 agent_end，能拿到一次请求的完整生命周期信息；脱敏也放这里，保证用户看到的永远是脱敏后的结果：

python复制@agent.hook("agent_end")
async def write_audit_log(ctx: HookContext):
    log_entry = {
        "run_id": ctx.run_id,
        "session_id": ctx.session_id,
        "user_id": ctx.user_id,
        "token_usage": ctx.usage,
        "latency_ms": ctx.elapsed_ms,
        "message_count": len(ctx.messages),
    }
    append_audit_log(log_entry)

@agent.hook("agent_end")
async def mask_sensitive_output(ctx: HookContext):
    text = ctx.output_text
    text = re.sub(r"1[3-9]\d{9}", "[手机号]", text)   # 手机号
    text = re.sub(r"\b\d{17}[\dXx]\b", "[身份证]", text)  # 身份证
    ctx.output_text = text

3.4 跑一遍测试：从日志看完整事件链

下面这段模拟测试代码，会把整个事件链和上下文变化打印出来：

python复制from gemini_agent.hooks import hook  # 全局事件监听，用于观测

@hook("agent_start")
@hook("context_build")
@hook("llm_call_start")
@hook("llm_call_end")
@hook("agent_end")
async def tracer(ctx: HookContext, event: str):
    print(f"[{event}] run_id={ctx.run_id} session_id={ctx.session_id}")

response = await agent.run(
    user_id="u_1001",
    session_id="s_2001",
    token="valid_token",
    message="我的项目预算还剩多少？",
)

实际输出大致是：

text复制[agent_start] run_id=run_8891 session_id=s_2001
[context_build] run_id=run_8891 session_id=s_2001
[llm_call_start] run_id=run_8891 session_id=s_2001
[llm_call_end] run_id=run_8891 session_id=s_2001
[agent_end] run_id=run_8891 session_id=s_2001

这个顺序本身就是很有价值的调试信息。如果哪一次请求没走 agent_end，说明中间有 Hook 要么 abort 了、要么抛异常了；如果同一个 run_id 下面出现了多组 llm_call_start/end，说明发生了工具调用循环。所以我的经验是：每个 Agent 项目上线前，先挂一个全局 tracer Hook，把所有事件顺序打印出来留档，后面排查问题会省很多力气。

4. 容易翻车的边界场景：并发、异常、重试与超时

4.1 并发请求下共享状态被污染：日志全串了

这是我踩过最典型的一个坑。早期我图省事，在 Hook 里用一个模块级变量存 run_id 到用户的映射：

python复制# 错误示范
_current_user = {}

@agent.hook("agent_start")
async def bad_hook(ctx: HookContext):
    _current_user[ctx.run_id] = ctx.user_id

线上并发一高，诡异的事情出现了：请求 A 的日志里打出了请求 B 的 user_id，审计数据全是乱的。排查了大半天才发现，模块级变量是所有请求共享的，Python 异步协程是并发切换的，两个请求的 Hook 执行过程中，全局字典的写入顺序完全不可控。

正确的做法只有一条：跨 Hook 传数据，用上下文对象，不要用全局变量。这就是为什么 HookContext 提供 set/get 方法——它就是给这次请求独占用的书包。后来我把所有中间状态都迁到 ctx 里，串数据的问题再也没出现过。

还有一个容易忽略的点：同一个 Hook 函数内部尽量不要用类属性或模块级可变对象缓存业务数据。如果你一定要做内存级缓存，请使用带过期时间的线程安全缓存组件，并且缓存 key 要包含上下文里的隔离标识。

4.2 Hook 自身抛异常：中断任务还是放行

框架默认行为是：agent_start、context_build 等节点上的 Hook 如果抛出未捕获异常，整个 Agent 调用会中断，异常最终被 agent_error 接住。如果 agent_error 也没处理，异常会继续传播到你的业务调用方。

这个默认行为其实很合理——鉴权失败、数据组装失败这些情况，本来就不该把流程继续跑下去。但问题在于，有些 Hook 只是"辅助性"的，比如打日志的 Hook、上报指标的 Hook，它们挂了不应该影响主流程。我在一次线上事故里就吃过亏：监控系统临时故障导致埋在 agent_end 的 Hook 抛异常，结果用户正常提问直接返回 500。

现在我的做法是给辅助性 Hook 加保护层：

python复制@agent.hook("agent_end")
async def safe_metric_hook(ctx: HookContext):
    try:
        await report_metric(ctx)
    except Exception:
        # 辅助 Hook 异常只记录，不上抛
        log_error("metric hook failed", exc_info=True)

原则很简单：改业务数据的 Hook 要把异常视为严重故障，往上抛；只做观测和辅助动作的 Hook 要自己兜底，别让它们污染主链路。架构上可以把这两类 Hook 放到不同的注册优先级，方便维护。

4.3 重试场景下 Hook 重复执行：n次重试就n份日志

Agent 层通常有重试机制：模型超时重试、工具调用失败重试、网络抖动重试。问题来了——重试的是整个 agent.run()，但你的 Hook 是按生命周期事件触发的，重试几次，agent_start 和 agent_end 也跟着触发几次。

后果是什么？用户一次提问，如果内部重试了 3 次，你的审计日志会写 3 条重复记录，限流计数器会被扣 3 次，计费金额如果挂在 agent_end 上也会重复计算。这非常坑。

我的解法分两步。第一，使用只读的 run_id 做幂等键。所有关键副作用（审计、扣费、计数）在写入前先检查 run_id 是否已处理过：

python复制@agent.hook("agent_end")
async def dedup_audit(ctx: HookContext):
    if not dedup_store.exists(ctx.run_id):
        dedup_store.mark(ctx.run_id)
        write_audit_log(ctx)

但要注意，如果重试发生在 Agent 内部，而 run_id 也重试时变了，这个幂等键就失效。所以第二，尽量把副作用留在真正"最终成功"的节点上，例如只在 agent_end 做扣费和审计，而不是分散在 llm_call_end、tool_call_end 这些循环节点。如果业务就是要在每次模型调用后即时计费，那就给每次 LLM 调用也生成子 ID 作为幂等键，而不是只依赖 run_id。

4.4 耗时的 Hook 把 P95 拖到 8 秒：超时与异步化

另一个真实事故：我在 agent_start 里挂了一个同步的远程鉴权 RPC 调用，代码是 requests.post(...)，没设超时。某个下午下游系统开始抖动，每次鉴权要花 6 秒，导致线上 Agent 的 P95 延迟从 2 秒直接飙到 8 秒。用户体感就是"这个机器人突然变笨了"。

Hook 是跑在 Agent 主链路上的，它的耗时就是用户请求的耗时。所以有两条铁律：

1. 外部调用必须设置超时，超时时间要远小于整体 Agent 响应预算
2. 能不阻塞的尽量异步化，或者把数据提前缓存到本地再判断

改造后的代码长这样：

python复制import asyncio

@agent.hook("agent_start")
async def check_auth(ctx: HookContext):
    try:
        result = await asyncio.wait_for(
            check_auth_remote(ctx.request.token),
            timeout=0.5  # 最多等 500ms
        )
    except asyncio.TimeoutError:
        ctx.abort("auth_timeout", code=408)

更进一步，如果鉴权结果可以接受"短暂过期"，我会在本地维护一个 token->user 的缓存，命中缓存直接放行，只有缓存没命中才走远程。这样一个 Hook 对 P95 的影响能控制在 10ms 以内，同时兜底逻辑也不复杂：缓存不存在就是 0.5 秒超时拦截，安全性和可用性都有保障。

5. 进阶玩法：把 Hooks 变成你的可观测性基础设施

5.1 全链路 Trace ID：在 agent_start 埋下一个种子

Agent 任务跟普通 Web 请求不一样，它内部有一连串模型调用、工具调用、记忆读写。排障时最大的痛点是"这么多日志之间都是散落的，怎么串起来"。我用 Hooks 做了一件小事：在 agent_start 生成一个 trace_id 放进上下文，并要求所有日志组件都从上下文里取这个 ID。

python复制import uuid

@agent.hook("agent_start")
async def init_trace(ctx: HookContext):
    trace_id = ctx.get("trace_id") or str(uuid.uuid4())
    ctx.set("trace_id", trace_id)
    set_global_trace_id(trace_id)  # 绑定到日志系统的 ThreadLocal/ContextVar

这样模型调用日志、工具调用日志、外部告警全部自动带上 trace_id。一次用户提问就是一张完整的时间线，哪里慢、哪里失败，一眼就能看出来。我强烈建议每个 Agent 项目上线前，先把这个 Hooks 挂上，它不解决功能问题，但它能决定你在生产环境排障的效率。

5.2 动态记忆注入：在上下文组装阶段改写消息列表

前面实战里已经简单演示过记忆注入，这里展开说一下进阶思路。记忆注入的难点是：不是所有历史消息都有用，而且记忆也分时效性。我把记忆分为长期偏好和短期上下文两类，在 context_build 里做分层处理：

python复制@agent.hook("context_build", priority=5)
async def inject_long_term_memory(ctx: HookContext):
    prefs = await load_user_prefs(ctx.user_id)
    if prefs:
        ctx.messages.append({
            "role": "system",
            "content": f"用户长期偏好：{prefs}",
        })

@agent.hook("context_build", priority=10)
async def inject_short_term_session(ctx: HookContext):
    recent = await load_session_memory(ctx.session_id, limit=5)
    if recent:
        ctx.messages.append({
            "role": "system",
            "content": f"最近对话：{recent}",
        })

两个 Hook 用 priority 控制先后顺序，避免长期偏好和短期上下文拼在一起时语义混乱。优先级低（数字小）的先执行，先放长期偏好打底，再放短期上下文补充，模型看到的信息结构是稳定的，输出质量会明显更稳。

5.3 输出安全过滤：chunk 级拦截与全文脱敏

安全过滤有两个维度。一是全文脱敏，在 agent_end 做一次整体处理，适合手机号、身份证号这种有明确正则的模式。二是流式场景下的实时拦截，你需要在 llm_chunk 事件里对每个 chunk 做增量检测，一旦发现敏感词的起点，立即决定是否切断后续生成。

举个简化的 chunk 级过滤思路：

python复制sensitive_buffer = ""

@agent.hook("llm_chunk")
async def filter_streaming_chunk(ctx: HookContext):
    global sensitive_buffer
    chunk_text = ctx.chunk_text
    blocked = contains_sensitive_prefix(chunk_text, sensitive_buffer)
    if blocked:
        ctx.abort("sensitive_content_blocked", code=400)
    sensitive_buffer = smax_match(chunk_text)

注意这里有一个工程细节：敏感词可能是跨 chunk 边界分割的，比如第一个 chunk 是"身份"，第二个 chunk 是"证号"，如果只对单个 chunk 做匹配，永远匹配不到。所以需要维护一个短窗口 buffer，把最近几个 chunk 的尾巴拼起来再检测。这个 buffer 要在 agent_end 或 agent_error 里清掉，否则会污染下一次请求。

5.4 真实收益：横切逻辑搬走后，迭代速度快了一倍

最后说点实在的。这套 Hooks 机制我用下来最直观的收益，不是代码变少了，而是新需求的交付节奏变了。以前加一个"只对 VIP 用户开放工具调用"的需求，我要改 Agent 主循环、加开关、跑回归。现在只需要挂一个 context_build 或者 tool_call_start 的 Hook，按用户身份改写系统提示词或直接 abort，十几行代码就能交付。

举一个实际项目里的例子：我们要上线一个"每个企业租户可以自定义模型温度"的功能。没有 Hooks 时，得去主流程里找模型调用参数赋值的地方，加一个租户配置查询，改动风险很高。有 Hooks 后，一行注册的 llm_call_start 里改写 ctx.llm_params.temperature，新功能完全不触碰主流程代码，上线当天就完成灰度，后来改成按租户套餐动态调整也没有任何额外成本。

另一个意外收获出现在多 Agent 协同场景：因为 run_id 和上下文是标准化的，父子 Agent 之间的参数传递、结果回填都能做到通过 Hook 自动注入。我是准备把这块写进下一期指南的——如果你已经在用 Gemini 3 的多 Agent 组合，建议提前把 Hook 事件链和 trace_id 埋好，后面接任何监控和管理系统都会顺畅得多。