Engine 是 QitOS 所有 agent 工作流共享的执行内核。它运行 step loop，协调 AgentModule 的 hooks，分发工具调用，执行 critics，检查 stop conditions，并写出 trace artifacts。只有当你需要比 agent.run() 更细粒度的控制时，才会直接和它交互。
QitOS 强制遵守 single-kernel rule：一次 run 只有一个 Engine。parser、critic、memory adapter、toolkit 等扩展都挂接在这条主流水线上，而不是再引入第二个执行循环。
循环是如何工作的
每一步都遵循固定顺序：
prepare → decide → act → reduce → critics → check_stop → trace
- prepare：
agent.prepare(state) 把 state 格式化成模型可直接消费的 prompt 文本。
- decide：先调用
agent.decide(state, observation)；若返回 None，再走 Engine 的默认模型调用路径。
- act：把
Decision.actions 中的工具调用交给 ToolRegistry 执行。
- reduce：
agent.reduce(state, observation, decision) 用新的 observation 更新 state。
- critics：所有已注册
Critic 在此步后评估当前结果，必要时可 stop 或 retry。
- check_stop：检查 budget、
FinalResultCriteria、agent.should_stop() 以及自定义 StopCriteria。
- trace：把本步的 step record 与 runtime events 写入
TraceWriter。
构造函数
from qitos import Engine
engine = Engine(
agent=agent,
budget=RuntimeBudget(max_steps=20),
parser=ReActTextParser(),
critics=[my_critic],
stop_criteria=[FinalResultCriteria()],
env=host_env,
trace_writer=trace_writer,
hooks=[my_hook],
)
| 参数 | 类型 | 说明 |
|---|
agent | AgentModule | 必需。要执行的策略模块。 |
budget | RuntimeBudget | None | 步数、时间与 token 限制。默认 RuntimeBudget(max_steps=10)。 |
parser | Parser | None | 把原始模型输出解析成 Decision，必须与 prompt 格式匹配。 |
critics | list[Critic] | None | 每步后执行的评估器，可 stop 或 retry。 |
stop_criteria | list[StopCriteria] | None | 停止条件列表。默认 [FinalResultCriteria()]。 |
env | Env | None | 提供 reset/observe/step/is_terminal/close 生命周期的 environment。 |
trace_writer | TraceWriter | None | 为本次 run 写出 manifest.json、events.jsonl、steps.jsonl。 |
hooks | list[EngineHook] | None | 生命周期 hooks。 |
render_hooks | list | None | 渲染 hooks，会在内部合并进 hooks。 |
history_policy | HistoryPolicy | None | 控制 in-run 对话历史如何被组装。 |
recovery_policy | RecoveryPolicy | None | 控制 step 失败时如何恢复。 |
单次 run 的普通使用场景优先用 agent.run()。当你需要跨多次任务复用同一个 Engine，或在运行间动态调整 hooks 时，再直接使用 Engine。
Engine.run(task)
result = engine.run(task)
Task 对象，返回一个 EngineResult。
当传入 Task 时，Engine 会读取 task.budget 并覆盖自身默认 budget，同时自动管理资源预检、environment reset/observe/close 等生命周期。
EngineResult
@dataclass
class EngineResult(Generic[StateT]):
state: StateT
records: List[StepRecord]
events: List[RuntimeEvent]
step_count: int
task_result: Optional[TaskResult]
| 字段 | 说明 |
|---|
state | 运行结束后的最终强类型 state。重点查看 state.final_result 与 state.stop_reason。 |
records | 每步一个 StepRecord，包含 decision、observation 与 state diff。 |
events | 本次 run 发出的全部 RuntimeEvent。 |
step_count | 实际执行的步数，即 len(records)。 |
task_result | 结构化任务结果，包含 success flag 与 criteria 结果。 |
result = engine.run("summarize the paper")
print(result.state.final_result)
print(result.state.stop_reason)
print(result.step_count)
print(result.task_result.success)
Hooks
hooks 用于在不修改 Engine 内部逻辑的前提下观察或响应生命周期事件。它们通常实现 EngineHook，在 on_before_step 与 on_after_step 等边界被调用。
engine.register_hook(my_hook)
engine.unregister_hook(my_hook)
engine.clear_hooks()
hooks 传入，或在 agent.run(hooks=[...]) 中动态附加。
Budget 耗尽
当 step 数、墙钟时间或 token 预算耗尽时，Engine 会把 state.stop_reason 设置为对应值，并写出 END event。运行会优雅结束，仍然返回 EngineResult，所以你可以通过检查 state.stop_reason 判断是否发生预算耗尽。
from qitos.engine.states import RuntimeBudget
engine = Engine(
agent=agent,
budget=RuntimeBudget(
max_steps=30,
max_runtime_seconds=120.0,
max_tokens=50_000,
),
)
从 AgentModule 构建 Engine
AgentModule.build_engine() 是一个便捷工厂，会创建一个和当前 agent 绑定好的 Engine：
engine = agent.build_engine(
budget=RuntimeBudget(max_steps=15),
trace_writer=trace_writer,
)
result = engine.run(task)
Engine(agent=agent, ...) 等价，也是 agent.run() 在内部采用的路径。