> ## Documentation Index > Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Hook 生命周期 > 掌握 EngineHook 生命周期，在 QitOS 运行的每个阶段边界上进行观察、日志记录与扩展。 Hook 是 QitOS 中首要的运行时扩展点。它们让你在不修改智能体本身的情况下，观察并对 Engine 内部的每个阶段转换做出反应。本教程覆盖完整的 Hook 生命周期、Hook 接收的上下文对象，以及如何编写步骤级和工具级的 Hook。 ## Hook 与 Critic 的区别 QitOS 有两种运行在不同层面的扩展机制： | 机制 | 角色 | 能否修改流程？ | | ---------- | ------ | --------------------- | | **Hook** | 观察与副作用 | 不能 -- Hook 只被通知，不参与控制 | | **Critic** | 控制与门控 | 能 -- Critic 可以否决或修改决策 | 当你需要记录日志、追踪、采集指标或触发外部通知时，使用 Hook。当你需要约束或覆盖智能体行为时，使用 Critic。 ## Step 1: EngineHook 基类 每个 Hook 都继承自 `EngineHook`。基类为每个生命周期回调定义了空操作方法，因此你只需覆盖需要的方法。 ```python theme={null} from qitos.engine.hooks import EngineHook ``` 在一次运行中，完整的回调集合按执行顺序排列如下： | 回调 | 触发时机 | | ----------------------------------- | --------------------- | | `on_run_start(task, state, engine)` | 第一个步骤开始之前 | | `on_before_step(ctx, engine)` | 步骤开始之前 | | `on_before_decide(ctx, engine)` | 调用 LLM 之前 | | `on_after_decide(ctx, engine)` | LLM 返回决策之后 | | `on_before_act(ctx, engine)` | 执行动作之前 | | `on_after_act(ctx, engine)` | 动作执行完毕之后 | | `on_before_critic(ctx, engine)` | Critic 运行之前 | | `on_after_critic(ctx, engine)` | Critic 运行完毕之后 | | `on_before_reduce(ctx, engine)` | `reduce()` 将结果归约到状态之前 | | `on_after_reduce(ctx, engine)` | `reduce()` 完成之后 | | `on_before_check_stop(ctx, engine)` | 评估停止条件之前 | | `on_after_check_stop(ctx, engine)` | 评估停止条件之后 | | `on_after_step(ctx, engine)` | 步骤完全结束后 | | `on_recover(ctx, engine)` | Engine 从错误中恢复时 | | `on_run_end(result, engine)` | 最终步骤完成之后 | 在步骤循环之外触发的额外生命周期回调： | 回调 | 触发时机 | | ---------------------------------------- | --------------------- | | `on_session_start(ctx, engine)` | 交互式会话开始时 | | `on_session_end(ctx, engine)` | 交互式会话结束时 | | `on_before_compact(ctx, engine)` | 上下文压缩运行之前 | | `on_after_compact(ctx, engine)` | 上下文压缩完成之后 | | `on_event(event, state, record, engine)` | 每次发射 `RuntimeEvent` 时 | | `on_step_end(record, state, engine)` | `StepRecord` 完成归档时 | 所有空操作方法返回 `None`。只需覆盖你关心的回调。 ## Step 2: HookContext 与 ToolHookContext 每个步骤级回调都接收一个 `HookContext` 数据类，它携带了 Hook 可能需要的所有信息： ```python theme={null} from qitos.engine.hooks import HookContext # HookContext 字段： # task: str -- 原始任务字符串 # step_id: int -- 当前步骤索引 # phase: RuntimePhase -- 当前 FSM 阶段 # state: StateSchema -- 活跃的智能体状态 # env_view: dict | None -- 环境快照 # observation: Any -- 最新观测 # decision: Any -- 最新决策 # model_response: dict -- 原始模型响应 # action_results: list -- 动作执行结果 # record: StepRecord -- 当前步骤记录 # payload: dict -- 阶段特定的任意数据 # error: Exception | None -- 阶段失败时的错误 # stop_reason: str | None -- 运行停止的原因 # run_id: str -- 唯一运行标识符 # ts: str -- ISO 时间戳 ``` 工具级回调接收 `ToolHookContext`，它在 `HookContext` 基础上扩展了工具特有的字段： ```python theme={null} from qitos.engine.hooks import ToolHookContext # ToolHookContext 额外字段： # tool_name: str -- 被调用的工具名称 # tool_args: dict -- 传给工具的参数 # tool_result: Any -- 工具返回的结果（on_after_tool_use 中可用） # permission_decision: str -- "allowed" 或 "denied"（on_permission_denied 中可用） ``` `phase` 字段来自 `RuntimePhase` 枚举： ```python theme={null} from qitos.engine.states import RuntimePhase # RuntimePhase 值： # INIT, DECIDE, ACT, CRITIC, REDUCE, CHECK_STOP, END, # DECIDE_ERROR, ACT_ERROR, RECOVER, # DELEGATE_START, DELEGATE_END, # HANDOFF_START, HANDOFF_END, # INTERRUPT, FANOUT_START, FANOUT_END, # COMPACT, SESSION_START, SESSION_END ``` ## Step 3: 编写自定义日志 Hook 一个常见用例是记录每个阶段转换，以便后续分析。下面是一个 `LifecycleRecorderHook`，它为每个回调记录时间戳和步骤 ID： ```python theme={null} import logging from qitos.engine.hooks import EngineHook, HookContext from qitos.engine.states import RuntimePhase logger = logging.getLogger("qitos.lifecycle") class LifecycleRecorderHook(EngineHook): """Records every engine lifecycle event into a structured log.""" def on_run_start(self, task, state, engine): logger.info("run_start | task=%s", task[:80]) def on_before_step(self, ctx: HookContext, engine): logger.info("before_step | step=%d phase=%s", ctx.step_id, ctx.phase.value) def on_after_step(self, ctx: HookContext, engine): logger.info("after_step | step=%d", ctx.step_id) def on_before_decide(self, ctx: HookContext, engine): logger.info("before_decide | step=%d", ctx.step_id) def on_after_decide(self, ctx: HookContext, engine): decision = ctx.decision logger.info("after_decide | step=%d decision=%s", ctx.step_id, type(decision).__name__) def on_before_act(self, ctx: HookContext, engine): logger.info("before_act | step=%d", ctx.step_id) def on_after_act(self, ctx: HookContext, engine): logger.info("after_act | step=%d results=%d", ctx.step_id, len(ctx.action_results)) def on_before_critic(self, ctx: HookContext, engine): logger.info("before_critic | step=%d", ctx.step_id) def on_after_critic(self, ctx: HookContext, engine): logger.info("after_critic | step=%d", ctx.step_id) def on_before_reduce(self, ctx: HookContext, engine): logger.info("before_reduce | step=%d", ctx.step_id) def on_after_reduce(self, ctx: HookContext, engine): logger.info("after_reduce | step=%d", ctx.step_id) def on_before_check_stop(self, ctx: HookContext, engine): logger.info("before_check_stop | step=%d", ctx.step_id) def on_after_check_stop(self, ctx: HookContext, engine): logger.info("after_check_stop | step=%d stop_reason=%s", ctx.step_id, ctx.stop_reason) def on_recover(self, ctx: HookContext, engine): logger.warning("recover | step=%d error=%s", ctx.step_id, ctx.error) def on_run_end(self, result, engine): logger.info("run_end | steps=%d", result.state.current_step if hasattr(result, 'state') else '?') ``` 因为 Hook 不能修改流程，所以将 `LifecycleRecorderHook` 添加到任何运行都是安全的，不会产生副作用。 ## Step 4: 工具级 Hook 工具级 Hook 围绕单次工具调用触发，让你可以精细地观察智能体调用了哪些工具以及它们的返回值。 ```python theme={null} from qitos.engine.hooks import EngineHook, ToolHookContext class ToolAuditHook(EngineHook): """Audits every tool call: name, args, result, and permission decisions.""" def on_before_tool_use(self, ctx: ToolHookContext, engine): print(f"[TOOL CALL] {ctx.tool_name}({ctx.tool_args})") def on_after_tool_use(self, ctx: ToolHookContext, engine): result_str = str(ctx.tool_result)[:200] # truncate for display print(f"[TOOL RESULT] {ctx.tool_name} -> {result_str}") def on_permission_denied(self, ctx: ToolHookContext, engine): print( f"[PERMISSION DENIED] {ctx.tool_name}({ctx.tool_args}) " f"decision={ctx.permission_decision}" ) ``` 三个工具级回调： | 回调 | 上下文 | 用途 | | ---------------------- | ------------------------------------------- | ---------- | | `on_before_tool_use` | `ToolHookContext`（尚无 `tool_result`） | 在执行前记录或校验 | | `on_after_tool_use` | `ToolHookContext`（包含 `tool_result`） | 检查或记录结果 | | `on_permission_denied` | `ToolHookContext`（包含 `permission_decision`） | 追踪被拒绝的工具调用 | 使用 `on_permission_denied` 来监控安全边界，而无需修改权限系统本身。 ## Step 5: 在 Engine 上注册 Hook 在调用 `run()` 之前，将 Hook 注册到 `Engine` 实例上： ```python theme={null} from qitos import Engine engine = Engine(agent=my_agent) engine.add_hook(LifecycleRecorderHook()) engine.add_hook(ToolAuditHook()) result = engine.run(task="Fix the bug in buggy_module.py") ``` 你可以注册多个 Hook。它们在每个回调中按注册顺序触发。因为 Hook 仅用于观察，顺序不影响控制流程 -- 但它会影响日志输出的顺序，这在调试时很重要。 查看当前已注册的 Hook： ```python theme={null} print(engine.hooks) # list of EngineHook instances ``` 移除某个 Hook： ```python theme={null} engine.remove_hook(my_hook_instance) ``` ## 完整生命周期图 ``` on_run_start | v on_before_step |---> on_before_decide |---> on_after_decide |---> on_before_act | |---> on_before_tool_use (每个工具调用) | |---> on_after_tool_use (每个工具调用) | |---> on_permission_denied (如被拒绝) |---> on_after_act |---> on_before_critic |---> on_after_critic |---> on_before_reduce |---> on_after_reduce |---> on_before_check_stop |---> on_after_check_stop v on_after_step | ... (循环直到停止) | v on_run_end ``` 当发生错误恢复时，`on_recover` 代替该步骤的其余回调触发，Engine 可能根据配置决定重试或中止。 了解 Critic 与 Hook 的区别，以及如何使用 Critic 进行控制流和决策门控。 构建包含协调者和工作者智能体的系统，实现任务的并行调度。