> ## 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.

# Engine

> Engine 是执行内核，负责驱动智能体循环、执行动作、管理追踪记录，并检查停止条件。

`Engine` 是 QitOS 所有智能体工作流共享的执行内核（kernel）。它运行步骤循环，协调 `AgentModule` 的钩子，分发工具调用，执行评估器，检查停止条件，并写出追踪记录产物。只有当你需要比 `agent.run()` 更细粒度的控制时，才会直接和它交互。

<Note>
  QitOS 强制遵守**单内核规则**：一次运行只有一个 `Engine`。解析器、评估器、记忆适配器、工具集等扩展都挂接在这条主流水线上，而不是再引入第二个执行循环。
</Note>

## 循环是如何工作的

每一步都遵循固定顺序：

```
prepare → decide → act → reduce → critics → check_stop → trace
```

1. **prepare**：`agent.prepare(state)` 把状态格式化成模型可直接消费的提示词文本。
2. **decide**：先调用 `agent.decide(state, observation)`；若返回 `None`，再走 Engine 的默认模型调用路径。
3. **act**：把 `Decision.actions` 中的工具调用交给 `ToolRegistry` 执行。
4. **reduce**：`agent.reduce(state, observation, decision)` 用新的观测结果更新状态。
5. **评估器**：所有已注册的 `Critic` 在此步后评估当前结果，必要时可停止或重试。
6. **check\_stop**：检查预算、`FinalResultCriteria`、`agent.should_stop()` 以及自定义 `StopCriteria`。
7. **追踪记录**：把本步的步骤记录与运行时事件写入 `TraceWriter`。

## 构造函数

```python theme={null}
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`      | 步数、时间与令牌限制。默认 `RuntimeBudget(max_steps=10)`。                |
| `parser`          | `Parser \| None`             | 把原始模型输出解析成 `Decision` 的解析器，必须与提示词格式匹配。                      |
| `critics`         | `list[Critic] \| None`       | 每步后执行的评估器，可停止或重试。                                           |
| `stop_criteria`   | `list[StopCriteria] \| None` | 停止条件列表。默认 `[FinalResultCriteria()]`。                        |
| `env`             | `Env \| None`                | 提供 `reset/observe/step/is_terminal/close` 生命周期的环境。          |
| `trace_writer`    | `TraceWriter \| None`        | 为本次运行写出追踪记录产物：`manifest.json`、`events.jsonl`、`steps.jsonl`。 |
| `hooks`           | `list[EngineHook] \| None`   | 生命周期钩子。                                                     |
| `render_hooks`    | `list \| None`               | 渲染钩子，会在内部合并进 `hooks`。                                       |
| `history_policy`  | `HistoryPolicy \| None`      | 控制运行内对话历史如何被组装。                                             |
| `recovery_policy` | `RecoveryPolicy \| None`     | 控制步骤失败时如何恢复。                                                |

<Tip>
  单次运行的普通场景优先用 `agent.run()`。当你需要跨多次任务复用同一个 Engine，或在运行间动态调整钩子时，再直接使用 `Engine`。
</Tip>

## `Engine.run(task)`

```python theme={null}
result = engine.run(task)
```

它接受普通字符串任务，也接受结构化 `Task` 对象，返回一个 `EngineResult`。

当传入 `Task` 时，Engine 会读取 `task.budget` 并覆盖自身默认预算，同时自动管理资源预检、环境重置/观测/关闭等生命周期。

## `EngineResult`

```python theme={null}
@dataclass
class EngineResult(Generic[StateT]):
    state: StateT
    records: List[StepRecord]
    events: List[RuntimeEvent]
    step_count: int
    task_result: Optional[TaskResult]
```

| 字段            | 说明                                                             |
| ------------- | -------------------------------------------------------------- |
| `state`       | 运行结束后的最终强类型状态。重点查看 `state.final_result` 与 `state.stop_reason`。 |
| `records`     | 每步一个 `StepRecord`，包含决策、观测结果与状态差异。                              |
| `events`      | 本次运行发出的全部 `RuntimeEvent`。                                      |
| `step_count`  | 实际执行的步数，即 `len(records)`。                                      |
| `task_result` | 结构化任务结果，包含成功标志与条件结果。                                           |

常见用法：

```python theme={null}
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)
```

## 钩子

钩子用于在不修改 Engine 内部逻辑的前提下观察或响应生命周期事件。它们通常实现 `EngineHook`，在 `on_before_step` 与 `on_after_step` 等边界被调用。

```python theme={null}
engine.register_hook(my_hook)
engine.unregister_hook(my_hook)
engine.clear_hooks()
```

你也可以在构造时通过 `hooks` 传入，或在 `agent.run(hooks=[...])` 中动态附加。

## 预算耗尽

当步数、墙钟时间或令牌预算耗尽时，Engine 会把 `state.stop_reason` 设置为对应值，并写出 `END` 事件。运行会优雅结束，仍然返回 `EngineResult`，你可以通过检查 `state.stop_reason` 判断是否发生了预算耗尽。

```python theme={null}
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()` 是一个便捷工厂，会创建一个和当前智能体绑定好的 Engine：

```python theme={null}
engine = agent.build_engine(
    budget=RuntimeBudget(max_steps=15),
    trace_writer=trace_writer,
)
result = engine.run(task)
```

这与直接写 `Engine(agent=agent, ...)` 等价，也是 `agent.run()` 在内部采用的路径。

## AsyncEngine

`AsyncEngine` 提供非阻塞的智能体执行能力。它封装了同样的 `Engine` 循环，但把阻塞调用放到线程池中执行，因此在 `asyncio` 事件循环中使用是安全的。

```python theme={null}
from qitos import AsyncEngine

async_engine = AsyncEngine(agent=agent, budget=RuntimeBudget(max_steps=20))
```

### `AsyncEngine.arun(task)`

异步运行智能体循环，返回与 `Engine.run()` 相同的 `EngineResult`。

```python theme={null}
result = await async_engine.arun("分析数据")
print(result.state.final_result)
```

### `AsyncEngine.arun_stream(task)`

异步运行智能体循环，并在事件发生时产出结构化的 `EngineEvent` 对象——非常适合实时 UI 更新或将进度流式推送给客户端。

```python theme={null}
async for event in async_engine.arun_stream("分析数据"):
    print(event.event_type, event.step_id, event.payload)
```

事件在步骤边界（`step_start`、`step_end`）、阶段切换（`decide`、`act`、`reduce`、`critic`、`check_stop`）和多智能体事件（`handoff`、`delegate`、`fanout`）时产出。流总是以 `run_start` 开始、以 `run_end` 结束。

### `EngineEvent`

```python theme={null}
@dataclass
class EngineEvent:
    event_type: EngineEventType   # step_start, step_end, decide, act, ...
    step_id: int
    agent_id: str | None          # 产出此步骤的智能体
    phase: RuntimePhase | None
    ok: bool
    payload: Dict[str, Any]
    error: str | None
    ts: str                       # ISO 8601 时间戳
```

### `EventStream`

`EventStream` 是支撑 `arun_stream()` 的异步队列。你也可以独立使用它，将事件扇出到多个消费者：

```python theme={null}
from qitos.engine.events import EventStream

stream = EventStream()
q1 = stream.subscribe()  # 消费者 1 的 asyncio.Queue
q2 = stream.subscribe()  # 消费者 2 的 asyncio.Queue
```

### 异步模型

当配置的模型实现了 `acall()`（来自 `AsyncModel`）时，`AsyncEngine` 可以在不阻塞事件循环的情况下调用它。内置的异步模型适配器：

```python theme={null}
from qitos.models import AsyncOpenAICompatibleModel, AsyncOpenAIModel

llm = AsyncOpenAICompatibleModel(
    model="qwen-plus",
    api_key="...",
    base_url="https://...",
)
result = await llm.acall([{"role": "user", "content": "你好"}])
```

<Tip>
  `AsyncEngine.arun()` 与任何模型都兼容——同步模型会自动调度到线程池。当你需要真正的非阻塞 I/O（例如在高并发 Web 服务器中）时，才需要使用异步模型适配器。
</Tip>