1. 从一次“越写越乱”的 Agent Loop 说起如果你正在做 Agent Harness 相关的工程大概率已经写过一个能跑通的agent_loop()解析模型返回的tool_calls找到对应 handler 执行把结果回填成roletool的消息再进入下一轮。这个循环本身不复杂真正让它变复杂的是后面不断加进来的东西——权限校验、审计日志、输出长度检查、调用次数统计、耗时打点。每加一个需求就在循环里插一个if几轮迭代之后主循环已经看不出哪部分是协议必需哪部分只是可插拔策略。这篇要解决的就是这个结构问题。核心思路是把这些扩展逻辑从 Agent Loop 里搬出去用 Hooks 机制在四个生命周期时点触发回调UserPromptSubmit、PreToolUse、PostToolUse、Stop。其中PreToolUse和PostToolUse是最关键的两个切点一个负责执行前拦截一个负责执行后观察。适合已经写过基础工具调用循环、想让代码结构更干净的人也适合刚接触 Agent Harness、想理解“扩展点”到底怎么落地的人。读完你能拿到一份可复制的 Hooks 注册表骨架、settings.json风格的配置示例以及一套触发验证动作确认扩展逻辑确实在循环外执行。我试过把权限、日志、输出检查全塞进agent_loop()结果是每次改一个策略都要动主循环回归测试范围越来越大。Hooks 的价值不是让代码行数变少而是让“核心流程”和“可变策略”彻底分开。2. 前置准备TaoToken 与运行环境在动手改代码之前先把模型调用这一层准备好。Agent Harness 需要一个能返回tool_calls的兼容接口这里用 TaoToken 作为模型服务入口。它的 API 地址是https://taotoken.net/api兼容 Chat Completions 协议工具调用走标准的tools和tool_calls字段不需要为 Hooks 新增任何协议字段——这一点很重要因为 Hooks 全部位于应用内部模型是看不到的。你需要先拿到一个 API Key。进入控制台创建密钥地址是https://taotoken.net/console/api-keys。创建后复制保存后面配置里会用到。如果你还没决定用哪个模型可以先在模型对话页面试一下工具调用是否正常返回地址是https://taotoken.net/models。环境方面Python 3.9 以上即可依赖只有openai这个包。安装命令pip install openai然后设置环境变量避免把 Key 写死在代码里export TAOTOKEN_API_KEY你的API Key如果你打算长期跑编码类 Agent或者需要多轮工具调用、Agent 编排可以了解一下 Coding Plan地址是https://taotoken.net/coding-plan。它更适合高频、长会话的场景普通调试用按量调用就够了。注意Hooks 是 Harness 内部机制不会改变模型协议。你仍然用 Chat Completions 的tools声明工具模型仍然返回tool_callsHooks 只在应用侧观察或控制这些调用的生命周期。3. 可复制配置Hooks 注册表与 settings.json 骨架先建立注册表。它是一个字典键是事件名值是按注册顺序排列的回调列表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 Noneregister_hook()把回调追加到列表注册顺序就是执行顺序。trigger_hooks()逐个调用一旦某个回调返回非None值立即返回不再执行同事件后面的回调。这个“返回非 None 就短路”是整个机制最重要的控制语义它让PreToolUse可以阻止 handler 执行。四个事件的回调签名并不相同这点必须记清楚事件触发位置回调签名典型用途UserPromptSubmit用户输入进入 messages 之前(query)记录环境、补充上下文PreToolUse参数解析后、handler 之前(block, arguments)权限校验、参数检查、执行前日志PostToolUsehandler 返回后、结果回填前(block, output)输出检查、审计、指标Stop模型不再请求工具、循环准备退出(messages)统计、决定是否强制继续接下来是回调实现。权限 Hook 从第 03 篇的硬编码函数改成注册式DENY_LIST [rm -rf, mkfs, dd if] DESTRUCTIVE [rm , /dev/sda] def permission_hook(block, arguments: dict): name block.function.name if name bash: cmd arguments.get(command, ) for pattern in DENY_LIST: if pattern in cmd: return Permission denied by deny list for keyword in DESTRUCTIVE: if keyword in cmd: choice input(Allow? [y/N] ).strip().lower() if choice not in (y, yes): return Permission denied by user return None def log_hook(block, arguments: dict): print(f[PreToolUse] tool{block.function.name} args{arguments}) return None def large_output_hook(block, output): if len(str(output)) 2000: print(f[PostToolUse] output too large: {len(str(output))} chars) return None def context_inject_hook(query): print(f[UserPromptSubmit] cwd{os.getcwd()}) return None def summary_hook(messages): tool_count sum(1 for m in messages if m.get(role) tool) print(f[Stop] total tool calls: {tool_count}) return None注册顺序决定执行顺序这一点在权限场景下尤其关键register_hook(PreToolUse, log_hook) register_hook(PreToolUse, permission_hook) register_hook(PostToolUse, large_output_hook) register_hook(UserPromptSubmit, context_inject_hook) register_hook(Stop, summary_hook)把log_hook放在permission_hook之前是为了让被拒绝的调用也能进入审计日志。如果反过来权限 Hook 返回拒绝字符串后触发短路日志 Hook 就没机会运行了。如果你希望把注册配置从代码里抽出来可以用一个settings.json风格的配置文件{ hooks: { UserPromptSubmit: [context_inject_hook], PreToolUse: [log_hook, permission_hook], PostToolUse: [large_output_hook], Stop: [summary_hook] }, hook_options: { fail_closed_events: [PreToolUse], timeout_ms: 3000 } }启动时读取这个文件按顺序调用register_hook()完成注册。fail_closed_events表示这些事件上的回调抛异常时默认阻断timeout_ms是单个回调的超时上限。生产环境里这两个字段能避免一个慢 Hook 卡住整个循环。4. 接回主链四个触发点与验证请求注册完成只说明“事件发生时该找谁”还必须由真实流程在正确状态边界主动触发。下面按一次请求的时间顺序把四个调用点接回agent_loop()。主程序在 query 进入历史前触发UserPromptSubmittrigger_hooks(UserPromptSubmit, query) history.append({role: user, content: query}) agent_loop(history)工具执行前参数已解析但 handler 尚未产生副作用此时触发PreToolUse才能真正阻止动作blocked trigger_hooks(PreToolUse, block, arguments) if blocked: messages.append({ role: tool, tool_call_id: block.id, content: str(blocked) }) continuehandler 返回后外部副作用可能已经发生此时触发PostToolUse适合观察输出handler TOOL_HANDLERS.get(name) output handler(**arguments) if handler else fUnknown: {name} trigger_hooks(PostToolUse, block, output) messages.append({ role: tool, tool_call_id: block.id, content: str(output) })模型没有继续请求工具时assistant 最终文本已加入历史但函数还没返回Stop位于这个狭窄边界if not message.tool_calls: force trigger_hooks(Stop, messages) if force: continue return现在做一次触发验证。构造一个命中 deny list 的 bash 调用观察执行顺序# 模拟模型返回的 tool_call block type(Block, (), { id: call_001, function: type(Fn, (), { name: bash, arguments: {command: rm -rf /tmp/test} })() })() arguments {command: rm -rf /tmp/test} result trigger_hooks(PreToolUse, block, arguments) print(blocked:, result)预期输出[PreToolUse] toolbash args{command: rm -rf /tmp/test} blocked: Permission denied by deny list日志 Hook 先执行并打印权限 Hook 返回拒绝字符串trigger_hooks()立即返回handler 和PostToolUse都不运行。主循环把拒绝原因作为工具结果回填进入下一轮模型调用。这条路径验证了短路既阻止 handler也阻止同事件的后续 Hook。再验证一次正常调用确认PostToolUse在 handler 之后触发output file1.txt\nfile2.txt trigger_hooks(PostToolUse, block, output)如果输出超过 2000 字符会看到[PostToolUse] output too large的提醒但output本身没有被修改工具结果照常回填。5. 本篇常见错排查回调签名不匹配导致 TypeError。当前注册表没有类型校验把一个接收(query)的函数注册到PreToolUse只有触发时才会抛错。排查方法是启动阶段加一次签名检查或者用inspect.signature对比参数个数。生产环境建议在注册时校验让错误配置在接收任务前暴露。权限 Hook 放错事件。如果把权限判断放到PostToolUse最多只能发现动作不应执行但副作用已经产生。PreToolUse发生在 handler 之前是唯一能真正阻止动作的时点。事件位置本身就是系统保证的一部分放错位置等于放弃拦截能力。短路顺序导致审计日志丢失。权限 Hook 注册在日志 Hook 之前时拒绝路径不会执行日志 Hook。解决办法是把审计类回调注册在阻塞类回调之前或者区分观察型回调和阻塞型 guardrail明确规定审计回调无论决策结果如何都必须执行。PostToolUse 抛异常导致状态不一致。handler 已经执行外部动作可能完成但工具结果还没加入 messages。此时简单重跑 handler 可能造成重复写入。排查时先持久化调用 ID、参数摘要和执行状态再触发执行后扩展恢复时依据幂等键判断是补写消息还是执行补偿。回调超时卡住整个循环。同步回调没有超时控制一个慢 Hook 会阻塞后续所有逻辑。给每个回调加超时或者把观测型回调放到独立线程池。settings.json里的timeout_ms就是为这个准备的。模型端看不到 Hook。如果你在等模型返回hook字段那方向就错了。Hooks 全部位于应用内部模型仍然通过tool_calls请求工具Harness 仍以工具消息回传 observation。兼容接口不会自动发送 Hook 信息。6. 下一步把扩展逻辑真正移出循环到这里Agent Loop 已经拥有多工具、执行前权限和生命周期扩展点。权限、日志、输出检查、停止统计都不再需要各自在agent_loop()里增加专用分支主循环重新呈现“调用模型—执行工具—回填结果”这条稳定骨架。最小主线可以压缩成一句话注册、触发、回调、可选短路。但当前trigger_hooks()还没有捕获异常也没有区分观察型、转换型和阻塞型返回值。生产系统至少要为不同事件定义失败策略权限 guardrail 失败默认阻断日志指标失败决定降级还是停止PostToolUse失败时仍要保存原始执行结果。回调还需要稳定输入类型、输出类型、优先级和唯一标识注册、卸载和版本升级要可追踪不能只依赖模块导入顺序。如果你准备把这套 Hooks 接到真实模型上跑先去 API Keys 页面确认密钥可用地址是https://taotoken.net/console/api-keys接入细节和字段说明看文档https://taotoken.net/doc。想先验证工具调用是否正常返回tool_calls用模型对话页面最快地址是https://taotoken.net/models。长期跑编码类 Agent、需要多轮工具调用和会话保持的看 Coding Planhttps://taotoken.net/coding-plan。下一篇会继续处理运行状态增长的问题把零散配置与上下文收敛为可组合的运行环境。
