事件与调用控制

通过 Agent 事件，可以观察运行过程，也可以在模型或工具调用前作出决定。每个 Agent 都会创建 EventBus 并订阅 TapeRecorder，你可以通过 agent.events 添加处理器。

三类信息各有用途：

类型	用途	访问方式
AgentEvent	内部生命周期通知和调用拦截。	agent.events.on(...)
StreamEvent	交给调用方的输出，例如文本和工具进度。	迭代 agent.run() 返回的流。
TapeEntry	保存的会话及部分运行记录。	查询 Tape。

Agent 循环控制执行过程，并在指定位置直接调用事件处理器。TapeRecorder 保存选定的交互和生命周期记录；其他需要保留的事件，可以由你的处理器写入 Tape。

注册处理器

下面使用 SDK 指南中的 agent：

from bub.core.events import StepFinished, ToolCallFinished


@agent.events.on(ToolCallFinished)
async def report_tool(event: ToolCallFinished) -> None:
    print(event.call.tool, event.result.duration_ms, "ms")


def report_step(event: StepFinished) -> None:
    print("step", event.step, event.status)


agent.events.on(StepFinished, report_step)

使用 agent.events.off(StepFinished, report_step) 移除注册。注册到基类的处理器也会接收子类事件；注册到 AgentEvent 的处理器可以接收所有被派发到它的 Agent 事件。

每次派发时，匹配的处理器按注册顺序执行。总线会等待异步处理器完成，再继续下一个。处理器的返回值会被忽略，异常则传播给 emit() 的调用方，并可能导致运行失败。如果希望观察器失败后工作仍能继续，应在观察器内部处理异常。

这个顺序针对单次派发。并发工具调用或不同运行可能同时发出事件，因此处理器共享的状态仍需要适当协调。

修改或停止调用

要停止调用，给事件的 decision 字段赋值：

from bub.core.events import ToolCallDecision, ToolCallStarted


@agent.events.on(ToolCallStarted)
def require_refund_permission(event: ToolCallStarted) -> None:
    if event.call.tool == "refund_order" and not event.state.get("refunds_allowed"):
        event.decision = ToolCallDecision.deny("Refunds require approval.")

deny() 跳过工具并产生工具错误；replace(result) 跳过工具并提供替代结果。两者都会停止把当前 ToolCallStarted 交给后续处理器，执行器仍会为结果发出 ToolCallFinished。模型循环会把错误作为工具结果交给模型；直接命令调用则会把错误传播给调用方。

需要修改参数并继续执行时，用 dataclasses.replace(event.call, arguments=...) 替换 event.call。需要在模型调用前结束时，在 LLMCallStarted 上设置 event.decision = LLMCallDecision.finish(text)。这会跳过供应商调用和后续处理器；被跳过的供应商调用不会产生 LLMCallFinished。

事件	发生时机	支持的操作
RunStarted	模型运行开始。	观察 prompt、模型和 metadata。
RunFinished	对应的流关闭。	观察状态、步骤数、耗时和错误。
StepStarted	模型步骤开始。	观察进度。
StepFinished	模型步骤结束。	观察进度；status == "continue" 时，可设置 next_prompt 为下一步添加用户消息。
MessagesPending	步骤构建模型请求时，或步骤结束且没有调用工具时。	向 messages 追加用户消息，排在下一个 prompt 之前；步骤结束时提供的消息会让运行继续。
LLMCallStarted	调用供应商之前。	替换 request，或设置 LLMCallDecision.finish(text)。
LLMCallFinished	供应商调用完成或抛出异常。	观察累积结果；这里不报告取消或消费方关闭流。
ToolCallStarted	执行工具之前。	替换 call、拒绝调用或提供替代结果。
ToolCallFinished	工具成功、失败、被拒绝或被替代。	赋值 event.result.result，改变返回结果。
ChatFinished	本步骤的交互已准备好记录。	观察交互。
CommandFinished	直接命令完成或失败。	观察命令、状态和输出。

命令 prompt 直接执行工具，产生工具事件和 CommandFinished。详见命令。

所有 agent 事件都携带本轮 state。继承 TapeEvent 的事件还携带当前 Tape。工具调用具有上下文时，可以通过 event.call.context 访问。

了解默认记录范围

Agent 在你的处理器之前注册 TapeRecorder。它记录 RunStarted、RunFinished、StepStarted、StepFinished、ChatFinished 和 CommandFinished。其中 ChatFinished 处理器保存消息与工具交互，供后续模型调用构建上下文。

记录属于 Agent 的正常运行过程，写入你配置的存储；未指定存储时写入内存。记录功能独立于 tracing 配置。

要保存自定义事件，可以添加处理器，将它写入 Tape。修改请求或结果应使用调用拦截事件；默认 Recorder 已经保存交互后，才会执行你的 ChatFinished 观察器。

记录业务事件

定义继承 AgentEvent 的 dataclass；如果处理器需要 Tape，就继承 TapeEvent。下面显式记录一项业务事件：

from dataclasses import dataclass

from bub.core.events import TapeEvent


@dataclass(kw_only=True)
class OrderReviewed(TapeEvent):
    order_id: str


@agent.events.on(OrderReviewed)
async def record_review(event: OrderReviewed) -> None:
    await event.tape.append_event("order.reviewed", {"order_id": event.order_id})


async def mark_reviewed() -> None:
    tape = agent.tape("customer-42")
    await agent.events.emit(
        OrderReviewed(state=tape.context.state, tape=tape, order_id="A123")
    )

emit() 返回处理器执行后的事件。自定义可变事件可以把结果带回发出方；如果结果确定后就应结束派发，可以覆写 stopped 属性。工具可以通过 context.events 发出事件，并传入相同的本轮状态。

从应用插件接入事件

AgentPlugins 将 before_llm_call、after_llm_call、before_tool_call、after_tool_call hooks 接到调用事件上，将 continue_prompt hook 接到 StepFinished 上，将 steering inbox 接到 MessagesPending 上。插件按 hook 要求返回值，由适配器更新对应事件。BubAgent 使用这个适配器，核心 Agent 同样可以使用。

该适配器遵循插件 hook 顺序，捕获单个拦截 hook 的异常，记录后跳过。拒绝调用应返回对应 decision。其他应用 hooks 遵循各自的错误处理规则。

应用 Hooks 指南提供插件示例，Hooks 参考列出顺序与签名。EventBus 包含在基础 SDK 安装中。

framework.events 通过 bub.application 中的 CommandChanged、WorkChanged 和 PromptChanged 发布命令、通道工作和用户提示，沿用相同的 on() 与 off() 接口订阅和取消订阅。