Hooks

在模型请求和工具执行的生命周期上插入钩子，用于观测、改写输入、包裹执行和后处理结果。

钩子做什么

Hook 让你在不修改 engine 的前提下，介入一次 run 的两条关键链路：模型请求和工具执行。每条链路都有三个钩子点：

阶段	模型请求	工具执行
before	`before_model_request`	`before_tool_execute`
around	`around_model_request`	`around_tool_execute`
after	`after_model_request`	`after_tool_execute`

before 用于在调用前改写输入，around 用于包裹整个调用（计时、重试观测、短路），after 用于在调用后改写或观测结果。

Mermaid

Rendering diagram...

钩子点由 HookType 枚举定义：

from we0agent.hooks.hooks import HookType

# before_model_request / around_model_request / after_model_request
# before_tool_execute  / around_tool_execute  / after_tool_execute

一个钩子就是一个对象

钩子不是函数，而是实现了对应协议方法的对象。同一个对象可以同时实现模型和工具的多个钩子方法，engine 会用 isinstance 按协议分别触发。你只需实现关心的方法。

模型相关协议（来自 we0agent.hooks.protocol）：

协议	方法	作用
`BeforeModelRequestHook`	`before_model_request(request)`	返回改写后的 `HookModelRequest`。
`AroundModelRequestHook`	`around_model_request(request, handler)`	调用 `handler.run(request)` 得到 `HookModelResult` 并返回。
`AfterModelRequestHook`	`after_model_request(result)`	返回改写后的 `HookModelResult`。

工具相关协议：

协议	方法	作用
`BeforeToolExecuteHook`	`before_tool_execute(request)`	返回改写后的 `HookToolExecuteRequest`。
`AroundToolExecuteHook`	`around_tool_execute(request, handler)`	调用 `handler.run(request)` 得到 `HookToolExecuteResult` 并返回。
`AfterToolExecuteHook`	`after_tool_execute(result)`	返回改写后的 `HookToolExecuteResult`。

所有方法都是 async，且必须返回与入参对应的类型。

注册到 We0Agent

把钩子对象组成列表传给 We0Agent(hooks=...)。engine 内部用 Hooks 容器持有它们，并在每个钩子点按列表顺序触发。

from we0agent.agent.agent import We0Agent

agent = We0Agent(
    name="demo",
    model=model,
    tools=tools,
    system_prompt=system_prompt,
    hooks=[ConsoleModelHook(), ConsoleToolHook()],
)

before 和 after 类钩子按列表顺序依次执行，每个钩子的返回值作为下一个钩子的输入，形成链式改写。around 类钩子则按列表顺序逐层嵌套：列表中靠前的钩子在最外层包裹靠后的钩子，最内层才是真正的模型调用或工具执行。

注册后无需额外调用 hook。后续 stream() 和 invoke() 都会经过同一组 hook：

await agent.invoke(
    abort=asyncio.Event(),
    session_id="ses_hook",
    messages=[user_message("读取项目状态并总结。")],
)

如果本轮触发模型请求，会依次进入模型 hook；如果模型调用工具，会继续进入工具 hook。

控制台观测示例

下面是一个只做打印、不改写数据的最小钩子。它同时实现模型和工具两条链路：

from we0agent.hooks.models import (
    HookModelResult,
    HookModelRequest,
    HookToolExecuteResult,
    HookToolExecuteRequest,
)
from we0agent.hooks.protocol import ModelRequestHookHandler, ToolExecuteHookHandler
from we0agent.domain.types.hooks import We0Hook


class ConsoleModelHook:
    async def before_model_request(self, request: HookModelRequest) -> HookModelRequest:
        print(f"[hook] before model request: messages={len(request.messages)}")
        return request

    async def around_model_request(
        self,
        request: HookModelRequest,
        handler: ModelRequestHookHandler,
    ) -> HookModelResult:
        print("[hook] around model request: start")
        result = await handler.run(request)  # 必须且只能调用一次
        print(f"[hook] around model request: end result={result.result}")
        return result

    async def after_model_request(self, result: HookModelResult) -> HookModelResult:
        print(f"[hook] after model request: result={result.result}")
        return result


class ConsoleToolHook:
    async def before_tool_execute(self, request: HookToolExecuteRequest) -> HookToolExecuteRequest:
        print(f"[hook] before tool execute: {request.tool_call['name']}")
        return request

    async def around_tool_execute(
        self,
        request: HookToolExecuteRequest,
        handler: ToolExecuteHookHandler,
    ) -> HookToolExecuteResult:
        print(f"[hook] around tool execute: start {request.tool_call['name']}")
        result = await handler.run(request)  # 必须且只能调用一次
        print(f"[hook] around tool execute: end {request.tool_call['name']}")
        return result

    async def after_tool_execute(self, result: HookToolExecuteResult) -> HookToolExecuteResult:
        print(f"[hook] after tool execute: {result.title}")
        return result


hooks: list[We0Hook] = [ConsoleModelHook(), ConsoleToolHook()]

改写输入和结果

钩子的载荷模型都提供 override() 方法，返回带有指定字段覆盖的副本，没传的字段保持原值。推荐用 override() 而不是直接改字段，避免影响原始对象。

HookModelRequest 可覆盖的字段：

字段	类型	说明
`messages`	`We0Messages`	本次模型请求的消息列表。
`tool_choice`	`str \| None`	工具选择策略。
`skip_cache_write`	`bool`	是否跳过 prompt cache marker 写入。

async def before_model_request(self, request: HookModelRequest) -> HookModelRequest:
    # 强制本次请求不写 cache marker
    return request.override(skip_cache_write=True)

HookToolExecuteRequest 持有一个 tool_call，可用 override(args=...) 改写工具入参：

async def before_tool_execute(self, request: HookToolExecuteRequest) -> HookToolExecuteRequest:
    if request.tool_call["name"] == "write_file":
        new_args = {**request.tool_call["args"], "dry_run": True}
        return request.override(args=new_args)
    return request

HookToolExecuteResult 可覆盖 title、metadata、output、attachments：

async def after_tool_execute(self, result: HookToolExecuteResult) -> HookToolExecuteResult:
    return result.override(title=f"[审计] {result.title}")

around 钩子的约束

around 钩子拿到的 handler 是下一层调用的入口。你必须且只能调用一次 await handler.run(request)：

不调用会让 engine 报错，因为底层模型调用或工具执行被跳过。
调用多于一次会抛出 RuntimeError，提示 handler.run() must be called exactly once。

这条约束保证 around 链路的嵌套语义清晰：每一层只包裹一次真实调用。在 handler.run() 前后可以自由插入计时、日志、异常捕获等逻辑，但调用本身不能省略也不能重复。