Agent Harness 工程：Hook 机制与可扩展循环

上一篇我们给 Agent Loop 加上了权限与审批：模型请求工具调用，Harness 先解析参数，再根据策略判断 allow、ask 或 deny，最后才决定是否执行真实工具。

这已经比“模型想调什么就调什么”安全很多了。

但继续往下做，很快会遇到另一个工程问题：权限检查只是工具调用前后的一个横切逻辑。真实 Agent 里还会有很多类似需求：

调用工具前记录日志；
调用工具前做权限检查；
调用工具前做参数规范化；
调用工具后截断超长结果；
调用工具后写审计记录；
调用工具失败后统一包装错误；
每轮循环结束后记录耗时和 token 使用情况。

如果每增加一个需求，就直接改 run_loop() 或 run_tool_call_with_permission()，核心循环会越来越臃肿。到最后，Agent Loop 里混着模型调用、消息维护、工具分发、权限、日志、审计、指标、错误处理，任何一个小改动都会影响主路径。

所以这一篇继续做一次抽象：把这些横切逻辑做成 Hook。

Hook 要解决什么问题

先看上一章的工具执行流程：

解析工具参数；
检查权限；
必要时请求用户确认；
执行真实工具；
把工具结果写回 messages。

这个流程没问题，但权限检查已经开始侵入工具执行函数。如果下一步再加日志、审计、结果截断，就会继续往里面塞代码。

Hook 的目标是让主流程保持稳定，把可插拔逻辑放到固定的生命周期点上：

Hook 是 Harness 内部的扩展点，位于模型工具请求和真实工具执行之间，以及工具执行之后。

可以把 Hook 理解成 Harness 里的“拦截器”：

工具执行前，可以检查、拒绝、审批、记录；
工具执行后，可以观察、截断、审计、统计；
主循环不需要知道每个 Hook 具体做了什么。

这里要注意一个边界：

工具是暴露给模型的能力，Hook 是 Harness 自己的工程扩展点。

模型知道有哪些工具，但不需要知道有哪些 Hook。Hook 不应该变成模型可以自由调用的东西，它是系统内部控制逻辑。

定义 Hook 上下文

Hook 要工作，首先需要拿到上下文。

对工具调用来说，Hook 至少需要知道：

当前是第几轮 Agent Loop；
这次 tool_call 的 id；
工具名；
解析后的参数；
当前消息历史；
工具执行结果。

我们先定义一个 ToolContext：

from dataclasses import dataclass
from typing import Any, Callable


@dataclass
class ToolContext:
    step: int
    tool_call_id: str
    tool_name: str
    arguments: dict[str, Any]
    messages: list[dict[str, Any]]
    result: str | None = None

这里把 result 设成可选，是因为工具执行前还没有结果，工具执行后才会填进去。

接下来定义一个 Hook 返回值：

@dataclass
class HookResult:
    block: bool = False
    content: str | None = None

这个结构先保持简单。

如果某个执行前 Hook 认为工具不应该继续执行，就返回：

HookResult(block=True, content="Error: permission denied.")

Harness 看到 block=True，就不再执行真实工具，而是把 content 当作工具结果写回 messages。这和上一章的权限拒绝逻辑保持一致：拒绝不是静默失败，而是变成模型能看懂的 tool 消息。

定义 Hook 类型和注册表

为了保持简单，我们先只做两类 Hook：

before_tool：真实工具执行前运行，可以阻断工具调用；
after_tool：真实工具执行后运行，可以修改或记录工具结果。

BeforeToolHook = Callable[[ToolContext], HookResult | None]
AfterToolHook = Callable[[ToolContext], None]


BEFORE_TOOL_HOOKS: list[BeforeToolHook] = []
AFTER_TOOL_HOOKS: list[AfterToolHook] = []

这两个列表就是 Hook 注册表。

顺序很重要。BEFORE_TOOL_HOOKS 会按列表顺序执行。如果第一个 Hook 已经阻断了工具调用，后面的 Hook 就不会继续执行。这样可以避免一个已经被拒绝的危险动作继续进入后续流程。

如果你希望“被拒绝的动作也要进入审计”，就应该把审计 Hook 放在权限 Hook 前面。Hook 的顺序就是策略的一部分。

Agent Loop 只调用 Hook 注册表；新增日志、权限、审计、截断逻辑时，不需要继续改主循环。

执行 before_tool Hooks

先写一个执行 before_tool 的函数：

def run_before_tool_hooks(context: ToolContext) -> str | None:
    for hook in BEFORE_TOOL_HOOKS:
        outcome = hook(context)
        if outcome and outcome.block:
            return outcome.content or "Error: tool call blocked by hook."
    return None

返回值是 str | None：

None 表示没有 Hook 阻断，可以继续执行真实工具；
str 表示某个 Hook 阻断了工具调用，这个字符串就是要写回模型的工具结果。

这种写法让 Hook 的阻断语义非常明确，不需要抛异常，也不需要在主循环里写一堆特殊分支。

执行 after_tool Hooks

执行后的 Hook 不需要阻断工具，因为工具已经执行完了。它更适合做观察和后处理。

def run_after_tool_hooks(context: ToolContext) -> None:
    for hook in AFTER_TOOL_HOOKS:
        hook(context)

after_tool Hook 可以读取 context.result，也可以修改它。

例如，工具输出太长时，可以截断结果，避免把 messages 撑爆：

MAX_TOOL_RESULT_CHARS = 50000


def truncate_tool_result_hook(context: ToolContext) -> None:
    if context.result is None:
        return

    if len(context.result) > MAX_TOOL_RESULT_CHARS:
        omitted = len(context.result) - MAX_TOOL_RESULT_CHARS
        context.result = (
            context.result[:MAX_TOOL_RESULT_CHARS]
            + f"\n... ({omitted} characters truncated)"
        )

这个 Hook 对模型是透明的。模型只会在下一轮看到被截断后的 tool 消息，但 Agent Loop 不需要关心截断细节。

把权限检查改造成 Hook

上一章我们写过：

permission = check_permission(tool_name, arguments)

现在可以把它包装成一个 before_tool Hook：

def permission_hook(context: ToolContext) -> HookResult | None:
    permission = check_permission(context.tool_name, context.arguments)

    if permission.action == "deny":
        return HookResult(
            block=True,
            content=f"Error: permission denied. {permission.reason}",
        )

    if permission.action == "ask":
        allowed = ask_user_for_permission(
            context.tool_name,
            context.arguments,
            permission.reason,
        )
        if not allowed:
            return HookResult(
                block=True,
                content=f"Error: user denied permission. {permission.reason}",
            )

    return None

这样权限检查就不再是工具执行函数里的硬编码逻辑，而是一个可插拔 Hook。

它仍然保留上一章的关键语义：

deny 直接阻断；
ask 请求用户确认；
用户拒绝也会变成工具结果；
被拒绝的结果会通过 tool_call_id 写回模型。

加一个审计 Hook

再写两个最简单的审计 Hook。

工具执行前记录模型想做什么：

def audit_before_tool_hook(context: ToolContext) -> HookResult | None:
    print(
        "[tool:start]",
        {
            "step": context.step,
            "tool_call_id": context.tool_call_id,
            "tool_name": context.tool_name,
            "arguments": context.arguments,
        },
    )
    return None

工具执行后记录结果长度：

def audit_after_tool_hook(context: ToolContext) -> None:
    result_length = len(context.result or "")
    print(
        "[tool:end]",
        {
            "step": context.step,
            "tool_call_id": context.tool_call_id,
            "tool_name": context.tool_name,
            "result_length": result_length,
        },
    )

教学代码里先用 print()。真实工程里，这里更可能是结构化日志、数据库审计表、OpenTelemetry span，或者专门的事件流。

注册 Hook

有了 Hook 之后，注册顺序就变得很重要。

BEFORE_TOOL_HOOKS = [
    audit_before_tool_hook,
    permission_hook,
]

AFTER_TOOL_HOOKS = [
    truncate_tool_result_hook,
    audit_after_tool_hook,
]

这里把 audit_before_tool_hook 放在 permission_hook 前面，是为了记录所有工具请求，包括后来被权限拒绝的请求。

truncate_tool_result_hook 放在 audit_after_tool_hook 前面，则意味着审计里看到的是最终写回模型的结果长度，而不是原始结果长度。如果你想同时记录原始长度和截断后长度，也可以再拆一个 Hook。

这就是 Hook 的一个特点：它很灵活，但顺序会影响语义。

改造工具执行函数

现在把 Hook 接进工具执行流程。

上一篇的 run_tool_call() 仍然保留，只负责调用真实工具：

def run_tool_call(tool_name: str, arguments: dict[str, Any]) -> str:
    handler = TOOL_HANDLERS.get(tool_name)
    if handler is None:
        return f"Error: unknown tool {tool_name}"

    try:
        return handler(**arguments)
    except TypeError as exc:
        return f"Error: invalid arguments for {tool_name} - {exc}"
    except Exception as exc:
        return f"Error: tool {tool_name} failed - {exc}"

然后新增一个带 Hook 的执行函数：

def run_tool_call_with_hooks(
    tool_call: Any,
    messages: list[dict[str, Any]],
    step: int,
) -> str:
    tool_name = tool_call.function.name
    arguments, error = parse_tool_arguments(tool_call)
    if error:
        return error

    context = ToolContext(
        step=step,
        tool_call_id=tool_call.id,
        tool_name=tool_name,
        arguments=arguments,
        messages=messages,
    )

    blocked_result = run_before_tool_hooks(context)
    if blocked_result is not None:
        return blocked_result

    context.result = run_tool_call(tool_name, arguments)
    run_after_tool_hooks(context)

    return context.result or ""

这段代码的结构很稳定：

解析参数；
构造 Hook 上下文；
执行 before_tool Hook；
如果被阻断，直接返回阻断结果；
执行真实工具；
执行 after_tool Hook；
返回最终工具结果。

权限、日志、审计、截断都不再散落在主流程里。

Agent Loop 继续保持稳定

接下来再看 run_loop()。

它和前几篇几乎一样，只是工具执行函数换成了 run_tool_call_with_hooks()：

SYSTEM_PROMPT = f"""
你是一个编程助手，工作在当前目录：{WORKDIR}。
你可以使用工具读取文件、列出目录、写入文件、编辑文件和执行必要的 shell 命令。
如果工具调用被拒绝，请根据拒绝原因调整方案，不要反复请求同一个危险操作。
"""


def run_loop(user_message: str, max_steps: int = 8) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": user_message}]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "system", "content": SYSTEM_PROMPT}] + messages,
            tools=TOOLS,
        )

        assistant_message = response.choices[0].message

        if not assistant_message.tool_calls:
            final_answer = assistant_message.content or ""
            messages.append({"role": "assistant", "content": final_answer})
            return final_answer

        messages.append(
            {
                "role": "assistant",
                "content": assistant_message.content or "",
                "tool_calls": [
                    {
                        "id": tool_call.id,
                        "type": "function",
                        "function": {
                            "name": tool_call.function.name,
                            "arguments": tool_call.function.arguments,
                        },
                    }
                    for tool_call in assistant_message.tool_calls
                ],
            }
        )

        for tool_call in assistant_message.tool_calls:
            tool_result = run_tool_call_with_hooks(tool_call, messages, step)
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result,
                }
            )

    return "达到最大循环次数，任务仍未完成。"

这就是 Hook 机制的价值。

如果后面要加一个“记录每次工具调用耗时”的能力，不需要改 run_loop()。新增一个 Hook，注册进去就行。

如果后面要加一个“禁止读取 .env 文件”的策略，也不需要改 run_loop()。新增一个权限 Hook，或者扩展现有 permission_hook()。

Agent Loop 继续只做三件事：

调模型；
维护消息历史；
把工具调用交给 Harness 执行。

Hook 不要变成新的混乱源

Hook 很有用，但也容易被滥用。

如果所有事情都往 Hook 里塞，最后会变成另一种隐式控制流：代码表面看起来很干净，但实际行为藏在一堆 Hook 里，排查问题反而更难。

所以 Hook 设计要注意几条原则。

命名清楚。permission_hook、audit_after_tool_hook 比 hook1、process_hook 好得多。Hook 名字应该说明它在做什么。

顺序明确。Hook 顺序会影响行为，所以注册表应该集中定义，不要散落在多个文件里偷偷追加。

职责单一。一个 Hook 最好只做一类事。权限 Hook 不要顺手写审计，审计 Hook 不要顺手改权限。

结果可观察。阻断工具调用时，要把原因写回 tool 消息。执行后修改结果时，最好能在日志里留下痕迹。

不要绕过主协议。Hook 可以阻断工具，可以修改工具结果，但不要直接往 messages 里乱塞结构不明的内容。消息协议仍然要保持稳定。

小结

这篇把权限与审批继续抽象成 Hook 机制，让 Agent Harness 多了一层可扩展能力：

用 ToolContext 描述一次工具调用的上下文；
用 before_tool Hook 在工具执行前做权限、审计和参数处理；
用 after_tool Hook 在工具执行后做截断、日志和观测；
把权限检查改造成 permission_hook；
让 Agent Loop 继续保持稳定，只调用 run_tool_call_with_hooks()。

到这里，我们的 Agent Harness 已经有了越来越清晰的分层：

TOOLS 告诉模型有哪些工具；
TOOL_HANDLERS 把工具名映射到真实函数；
权限策略决定哪些动作能执行；
Hook 机制承载横切逻辑；
Agent Loop 只负责维持观察、行动、反馈的循环。

下一篇继续往“可完成任务”的方向走：TodoWrite。当任务不再是一两步就能完成时，Agent 应该先把目标拆成清晰的待办步骤，执行过程中持续更新进度，再根据当前 Todo 决定下一步行动。也就是说，让 Agent 不只是会调用工具，还要学会先列计划、按步骤推进、及时标记完成。