Hook 生命周期

掌握 EngineHook 生命周期，在 QitOS 运行的每个阶段边界上进行观察、日志记录与扩展。

Hook 是 QitOS 中首要的运行时扩展点。它们让你在不修改智能体本身的情况下，观察并对 Engine 内部的每个阶段转换做出反应。本教程覆盖完整的 Hook 生命周期、Hook 接收的上下文对象，以及如何编写步骤级和工具级的 Hook。

​

Hook 与 Critic 的区别

QitOS 有两种运行在不同层面的扩展机制：

机制	角色	能否修改流程？
Hook	观察与副作用	不能 — Hook 只被通知，不参与控制
Critic	控制与门控	能 — Critic 可以否决或修改决策

当你需要记录日志、追踪、采集指标或触发外部通知时，使用 Hook。当你需要约束或覆盖智能体行为时，使用 Critic。

​

Step 1: EngineHook 基类

每个 Hook 都继承自 EngineHook。基类为每个生命周期回调定义了空操作方法，因此你只需覆盖需要的方法。

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 可能需要的所有信息：

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 基础上扩展了工具特有的字段：

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 枚举：

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：

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 围绕单次工具调用触发，让你可以精细地观察智能体调用了哪些工具以及它们的返回值。

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 实例上：

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：

print(engine.hooks)  # list of EngineHook instances

移除某个 Hook：

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

了解 Critic 与 Hook 的区别，以及如何使用 Critic 进行控制流和决策门控。

下一篇教程：多智能体系统

构建包含协调者和工作者智能体的系统，实现任务的并行调度。