首页
/ learn-claude-code s04 实战:Hooks 机制——把扩展逻辑挂在 Agent 循环上,而不是写进循环里

learn-claude-code s04 实战:Hooks 机制——把扩展逻辑挂在 Agent 循环上,而不是写进循环里

2026-09-04 15:43:30作者:裴锟轩Denise

learn-claude-code 这个从零搭建的 Claude Code 风格 agent harness 教程中,s04 一课解决一个核心工程问题:当 Agent 需要不断叠加"记录日志、权限检查、注入上下文、自动收尾"等扩展行为时,如何避免把主循环 agent_loop 改得面目全非。读完本文,你将掌握 4 个 hook 事件(UserPromptSubmit / PreToolUse / PostToolUse / Stop)的触发时机与返回值语义,能独立实现一个 hook 注册表 + trigger_hooks() 的最小扩展框架,并理解该模式如何成为后续所有章节(subagent、任务系统、后台任务等)的公共底座。

Hooks Overview: 用户输入 → UserPromptSubmit → LLM → PreToolUse → 工具执行 → PostToolUse → Stop 的完整 agent 循环

一、问题:循环正在被扩展逻辑淹没

s03(权限章节,见 s03_permission/code.py)给 Agent 加上了权限检查。但问题在于:每加一个新检查——"记录每次 bash 调用"、"写文件后自动 git add"、"通知 Slack"——都得直接修改 agent_loop 函数体。循环很快就变成这样:

def agent_loop(messages):
    while True:
        # ... LLM call ...
        for block in response.content:
            if block.type != "tool_use":
                continue
            log_to_file(block)          # 加一行
            check_permission(block)     # 加一行
            notify_slack(block)         # 又加一行
            output = execute(block)
            auto_git_add(block)         # 再加一行
            # ... 很快循环就认不出来了

你想扩展的是 Agent 的行为,但你改的却是循环本身。s04 的口号是:"挂在循环上,不写进循环里"(Hang on the loop, don't write into it)——循环应该是一个稳定核心,扩展逻辑应该挂在它外面。

二、四个 hook 事件:覆盖一个完整的 agent 周期

s04 的设计原则是:s03 的循环结构和权限逻辑完全保留,唯一变化是把 check_permission() 从循环体内移到 hook 上。循环不再直接调用任何检查函数,而是调用 trigger_hooks("PreToolUse", block),由注册表决定跑什么。

四个事件覆盖输入、执行前后、退出的关键节点(引自 s04_hooks/README.md):

事件 触发时机 典型用途
UserPromptSubmit 用户输入提交后、进入 LLM 前 输入验证、注入上下文
PreToolUse 工具执行前 权限检查、日志记录
PostToolUse 工具执行后 副作用(自动 git add 等)、输出检查
Stop 循环即将退出时 收尾清理、决定是否继续循环

扩展通过 register_hook() 添加,循环只调用 trigger_hooks()。这是典型的"注册表 + 触发器"模式:注册是 O(1) 的追加,触发是按注册顺序的串行执行。

三、hook 注册表:约 10 行的最小实现

核心实现只有三个部分,位于 s04_hooks/code.py

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:  # A hook result blocks this tool call.
            return result
    return None

返回值语义是这套机制的精髓trigger_hooks() 按注册顺序执行所有回调,一旦某个回调返回非 None,立即短路并把这个返回值交还给调用方;全部返回 None 则整体返回 None(放行)。但"非 None 意味着什么"因事件而异:

  • PreToolUse:返回非 None → 本次工具执行被阻止,返回值(通常是一段拒绝说明字符串)会被写进 tool_result 回传给模型;
  • Stop:返回非 None(一段消息)→ 循环不退出,消息作为新的 user 消息注入,强制继续;
  • UserPromptSubmit / PostToolUse:返回值不参与控制流,仅作观察/副作用。

这种"同一个触发器、不同事件各自解释返回值"的设计,让循环里的调用点保持极少,而行为决策全部收敛在 hook 回调内。

四、逐个事件看实际回调

UserPromptSubmit:进入 LLM 前的输入拦截

code.py 中,示例 hook 只打印当前工作目录,返回 None 放行:

def context_inject_hook(query: str):
    print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
    return None

register_hook("UserPromptSubmit", context_inject_hook)

触发位置在主 REPL 循环里、消息写入 history 之前(code.py 主循环):

query = input("\033[36ms04 >> \033[0m")
if query.strip().lower() in ("q", "exit", ""):
    break
trigger_hooks("UserPromptSubmit", query)   # ← 进入 LLM 之前
history.append({"role": "user", "content": query})
agent_loop(history)

注意它是在会话级主循环而非 agent_loop 内触发的:每次用户输入只触发一次,而不是每个 LLM 往返都触发。

PreToolUse:s03 的权限管道被整体"搬"进 hook

s03 中权限检查是硬编码在循环里的三段式管道(deny list → 规则匹配 → 用户确认),见 s03_permission/code.py 的 check_permission 管道。s04 把这套逻辑整体包装成一个 PreToolUse 回调 permission_hook,并新增了日志 hook:

# PreToolUse: 权限检查(s03 的逻辑,从循环移到 hook)
DENY_LIST = ["rm -rf /", "sudo", "shutdown", "reboot", "mkfs", "dd if="]
DESTRUCTIVE = ["rm ", "> /etc/", "chmod 777"]

def permission_hook(block):
    """PreToolUse: s03 check_permission() logic moved here."""
    if block.name == "bash":
        for pattern in DENY_LIST:
            if pattern in block.input.get("command", ""):
                print(f"\n\033[31m[blocked] '{pattern}'\033[0m")
                return "Permission denied by deny list"
        for kw in DESTRUCTIVE:
            if kw in block.input.get("command", ""):
                print(f"\n\033[33m[permission] Potentially destructive command\033[0m")
                print(f"   Tool: {block.name}({block.input})")
                choice = input("   Allow? [y/N] ").strip().lower()
                if choice not in ("y", "yes"):
                    return "Permission denied by user"
    if block.name in ("read_file", "write_file", "edit_file"):
        path = block.input.get("path", "")
        if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
            choice = input("   Allow? [y/N] ").strip().lower()
            if choice not in ("y", "yes"):
                return "Permission denied by user"
    return None

# PreToolUse: 日志
def log_hook(block):
    """PreToolUse: log every tool call."""
    args_preview = str(list(block.input.values())[:2])[:60]
    print(f"\033[90m[HOOK] {block.name}({args_preview})\033[0m")
    return None

对比 s03 可以看到两个细节差异:一是 s04 的 DENY_LIST 去掉了 "> /dev/sda"(s03 为 7 项),s04 精简为 6 项并单独用 DESTRUCTIVE 列表承载"需人工确认"的关键词;二是 s04 的 hook 直接返回拒绝原因字符串(如 "Permission denied by deny list"),而 s03 的 check_permission() 返回 bool、拒绝消息固定为 "Permission denied."——hook 化之后,拒绝的具体理由可以透传给模型,信息量更大。

注册顺序决定了执行顺序(code.py 注册区):

register_hook("UserPromptSubmit", context_inject_hook)
register_hook("PreToolUse", permission_hook)   # 先权限、后日志
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)

PostToolUse:执行后的观察点

def large_output_hook(block, output):
    """PostToolUse: warn on large output."""
    if len(str(output)) > 100000:
        print(f"\033[33m[HOOK] Large output from {block.name}: {len(str(output))} chars\033[0m")
    return None

阈值是 100000 字符(code.py)。这里也解释了为什么它必须是 Post 而非 Pre:只有工具真正执行完才有 output 可检查。该回调同时收到 blockoutput 两个参数,体现了 trigger_hooks(event, *args) 变长参数设计——每个事件向回调透传自己上下文的任意参数组合。

Stop:唯一能"否决退出"的事件

当 LLM 返回 stop_reason != "tool_use" 时,循环本应退出。s04 在退出前插入了 Stop hook(code.py):

if response.stop_reason != "tool_use":
    force = trigger_hooks("Stop", messages)
    if force:
        # hook 返回了一段消息 → 注入并强制继续循环
        messages.append({"role": "user", "content": force})
        continue
    return

示例回调遍历全部消息统计 tool_result 块数量,打印会话级统计:

def summary_hook(messages: list):
    tool_count = sum(1 for m in messages
                     for b in (m.get("content") if isinstance(m.get("content"), list) else [])
                     if isinstance(b, dict) and b.get("type") == "tool_result")
    print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
    return None   # return None = allow stop, return string = force continuation

注意返回值契约:返回 None 允许停止;返回一个非空字符串则循环把该字符串包装成 user 消息追加到历史并 continue——这就是"强制继续"的实现方式,hook 甚至可以用返回的内容给模型下达"继续"的指令。

五、循环本体:只改了一处

对比 s03 的循环(循环内直接 if not check_permission(block): ...)与 s04 的 agent_loop,工具处理段的变化只有一处:

for block in response.content:
    if block.type != "tool_use":
        continue

    # s03: if not check_permission(block): ...
    # s04: hook 替代硬编码
    blocked = trigger_hooks("PreToolUse", block)
    if blocked:
        results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": str(blocked)})
        continue

    handler = TOOL_HANDLERS.get(block.name)
    output = handler(**block.input) if handler else f"Unknown: {block.name}"

    trigger_hooks("PostToolUse", block, output)

    results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

被阻止时,拒绝字符串被原样放进 tool_result 回传——模型因此能"看到"为什么失败并自行调整,这与 s03 里固定回复 "Permission denied." 相比是明显的行为改进。完整的 s04 相对 s03 变更清单:

组件 之前 (s03) 之后 (s04)
扩展方式 check_permission() 硬编码在循环里 HOOKS 注册表 + trigger_hooks()
新函数 register_hook, trigger_hooks
hook 回调 context_inject_hook, permission_hook, log_hook, large_output_hook, summary_hook
循环 直接调用 check_permission() 调用 trigger_hooks("PreToolUse", ...)
退出控制 trigger_hooks("Stop", ...) 可阻止退出
输入拦截 trigger_hooks("UserPromptSubmit", ...) 可注入上下文

六、运行与观察

环境准备

依赖声明在 requirements.txtanthropic>=0.25.0python-dotenv>=1.0.0pyyaml>=6.0code.py 入口 通过 load_dotenv(override=True) 加载 .env,需要两个环境变量:

  • MODEL_ID:必填,指定使用的模型;
  • ANTHROPIC_BASE_URL:可选,指向兼容 Anthropic API 的网关;设置后代码会 popANTHROPIC_AUTH_TOKEN,交由 base_url 对应的鉴权方式处理。
cd learn-claude-code
python s04_hooks/code.py

三条验证 prompt

  1. Read the file README.md —— 应该直接通过,观察 [HOOK] read_file(...) 日志是否出现(PreToolUse 的 log_hook);
  2. Create a file called test.txt —— 创建成功后观察 PostToolUse 链路是否被走过;
  3. Delete all temporary files in /tmp —— 命令含 rm 关键词,会触发 permission_hook 的 DESTRUCTIVE 分支,提示 Allow? [y/N],拒绝后模型会收到 "Permission denied by user" 作为 tool_result。

观察重点:每次工具执行前是否出现 [HOOK] 日志?权限被拒时是 hook 拦截的还是循环里硬编码的?如果答案都指向 hook,说明"循环干净、扩展外挂"的目标达成。

七、源码层面的佐证:hook 系统成为后续章节的公共内核

s04 不是一课一弃的玩具,仓库测试明确把它定位为后续所有进阶章节的内核(kernel):

  • tests/test_s06_subagent.py 断言 s06 在 s04 内核上仅新增 task 工具,并验证 lesson.large_output_hook in lesson.HOOKS["PostToolUse"],即 PostToolUse 注册表原样继承;
  • tests/test_task_system.py 命名为 test_s10_keeps_the_s04_kernel_and_adds_task_tools,并断言 lesson.permission_hook in lesson.HOOKS["PreToolUse"]
  • tests/test_background_tasks.pytests/test_cron_scheduler.py 分别以 keeps_the_s04_kernel 命名,确认 s11/s12 同样建立在 s04 内核之上;
  • tests/test_goal_loop.py 直接调用 session.trigger_hooks("PreToolUse", block) 验证越权路径被拒(返回包含 "outside" 的拒绝串),证明 Stop/PreToolUse 的返回值契约在更复杂的运行时里仍然成立。

从这些测试的结构可以推断:本仓库后续的 subagent、任务系统、后台任务、cron 调度、目标循环等章节,都复用了同一套 HOOKS 注册表 + trigger_hooks() 触发器,hook 回调集合则按需增删。这正是 s04 设计要达到的工程效果——循环代码保持稳定,能力增长只发生在注册表和回调里。

小结:为什么"挂"比"写"好

s04 给出的答案可以用三句话概括:

  1. 循环是稳定核心agent_loop 只保留 LLM 调用、工具分派和三个 trigger_hooks() 调用点(PreToolUse / PostToolUse / Stop),加一个主循环里的 UserPromptSubmit;
  2. 行为决策外置:权限、日志、输出检查、收尾统计全部是可插拔回调,注册顺序即执行顺序,任一回调可用返回值短路后续回调;
  3. 返回值即协议:同一套 trigger_hooks 返回值语义在不同事件上被解释为"阻止执行"或"强制继续",循环代码因此无需为每种扩展行为新增分支。

掌握了这一层,你就有了继续上层的钥匙:s05 开始讨论的是 Agent 拿到工具之后"先规划再执行"的问题,可以接着阅读 s05_todo_write 章节;而 s04 的 hook 机制会一直伴随到仓库最后的集成 harness 章节。

登录后查看全文
热门项目推荐
相关项目推荐