RSS

第 4 章 / 共 21 章

钩子系统:把扩展挂在循环外面

约 7 分钟 更新于

上一章留下的问题:权限检查被直接写进了 agent_loop。现在再来三个需求——每次工具调用要打日志、写文件后要自动 git add、任务结束要发个通知。照上一章的写法,这三个需求会各自往循环里塞几行。加到第五个需求时,你已经认不出那段 while True 了。

问题不在于需求多,而在于它们全都是横切关注点:日志、权限、审计、通知——每一个都跟”循环”本身无关,却都要在循环的某个时刻插一脚。

4.1 挂在循环上,不要写进循环里

解法是给循环定义几个事件点,允许外部代码注册回调:

事件触发时机典型用途
UserPromptSubmit收到用户输入之后、调模型之前输入校验、注入上下文
PreToolUse执行工具之前权限检查、日志、拦截
PostToolUse执行工具之后副作用、输出体检
Stop循环准备退出之前清理、统计、强制续跑

注册表本身只有十几行:

HOOKS = {"UserPromptSubmit": [], "PreToolUse": [], "PostToolUse": [], "Stop": []}

def register_hook(event: str, callback):
    HOOKS[event].append(callback)

def trigger_hooks(event: str, *args):
    for callback in HOOKS[event]:
        result = callback(*args)
        if result is not None:      # 有意见就中断,并把意见交回去
            return result
    return None                     # 全体沉默 = 正常放行

**None 的语义是”我没有意见,照常走”。**返回非 None,表示这个钩子要干预——在 PreToolUse 上就是”拦截,并把这段文字作为拒绝理由”,在 Stop 上就是”别退出,把这段文字当作新输入继续跑”。同一个”返回值即决策”的约定,在不同事件点有不同含义,这是这套设计最需要记牢的一点。

第4章:内核与外挂
图 4.1:循环内核与挂在它周围的扩展层。要看的是那条虚线边界——边界以内是第1章那三十行,永远不动;所有新需求都活在边界以外。判断你的 Agent 架构是否健康,就看新需求往边界的哪一侧落。

4.2 把第3章的权限搬出去

上一章那段 check_permission 现在变成一个普通的钩子:

def permission_hook(block):
    if block.name == "bash":
        for pattern in DENY_LIST:
            if pattern in block.input.get("command", ""):
                return f"权限拒绝:命中硬拒绝名单({pattern})"
    reason = check_rules(block.name, block.input)
    if reason and ask_user(block.name, block.input, reason) == "deny":
        return "用户拒绝了此操作。"
    return None

def log_hook(block):
    print(f"[HOOK] {block.name}({block.input})")
    return None

def large_output_hook(block, output):
    if len(str(output)) > 100_000:
        print(f"[HOOK] ⚠ {block.name} 返回了超大输出({len(str(output))} 字符)")
    return None

def summary_hook(messages: list) -> str | None:
    tool_calls = sum(
        1 for m in messages
        for b in (m["content"] if isinstance(m.get("content"), list) else [])
        if isinstance(b, dict) and b.get("type") == "tool_result"
    )
    print(f"[HOOK] 本次会话共执行了 {tool_calls} 次工具调用")
    return None

register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)

循环里的改动,只有这么多:

        if response.stop_reason != "tool_use":
            force = trigger_hooks("Stop", messages)
            if force:                                  # 钩子要求继续
                messages.append({"role": "user", "content": force})
                continue
            return

        for block in response.content:
            if block.type == "tool_use":
                blocked = trigger_hooks("PreToolUse", block)
                if blocked:
                    results.append({"type": "tool_result", "tool_use_id": block.id,
                                    "content": str(blocked)})
                    continue
                output = TOOL_HANDLERS[block.name](**block.input)
                trigger_hooks("PostToolUse", block, output)
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

对比一下第3章:那一版里 check_permission(block) 是写死的函数调用;这一版里 trigger_hooks("PreToolUse", block) 是一个开放的插槽。**同样的行为,完全不同的可扩展性。**从此以后新需求不再需要动循环。

4.3 走一遍:Stop 钩子的一个真实用法

Stop 事件最有意思的能力是”强制续跑”。举个具体场景:你希望 Agent 在声称完成之前,必须跑一次测试。

def must_run_tests_hook(messages) -> str | None:
    text = str(messages[-1])
    if "pytest" not in str(messages) and "测试" in str(messages[0]):
        return "你还没有运行测试。请执行测试套件,确认通过后再给出结论。"
    return None

模型说”我改完了”,stop_reason 变成 end_turn,循环准备退出——钩子拦下它,往 messages 里塞一句”先跑测试”,continue,循环又转起来了。这就是”让策略约束模型行为”的一个非常朴素、也非常有效的做法:你不需要模型自觉,你需要一个在出口处等着它的钩子。

4.4 工程现实:从 4 个事件到 27 个

教学版给了 4 个事件点,够用来理解模式。一个成熟的实现会有二十来个,覆盖工具、会话、用户交互、子代理、压缩、团队、基础设施等多个类别——因为随着机制增多(后面十六章你会一个个加上去),需要插手的时刻自然变多。

设计事件点时有两条经验:

  1. 事件要落在”决策之前”和”结果之后”这两类时刻上,而不是随便找个位置埋点。
  2. 同一事件上的多个钩子要能顺序执行、互不干扰,并且明确规定”谁先返回非 None,谁说了算”。

4.5 常见坑

4.6 把新需求挂到正确的事件上

4.7 把三个需求归到正确的事件