第 4 章 / 共 21 章
钩子系统:把扩展挂在循环外面
上一章留下的问题:权限检查被直接写进了 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.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 个事件点,够用来理解模式。一个成熟的实现会有二十来个,覆盖工具、会话、用户交互、子代理、压缩、团队、基础设施等多个类别——因为随着机制增多(后面十六章你会一个个加上去),需要插手的时刻自然变多。
设计事件点时有两条经验:
- 事件要落在”决策之前”和”结果之后”这两类时刻上,而不是随便找个位置埋点。
- 同一事件上的多个钩子要能顺序执行、互不干扰,并且明确规定”谁先返回非
None,谁说了算”。