Engine is the single execution kernel for all QitOS agent workflows. It runs the step loop, coordinates your AgentModule hooks, dispatches tool calls to the ToolRegistry for execution, applies critics (post-step evaluators that can approve, stop, or retry the step), checks stop conditions, and writes trace artifacts (persistent records like steps.jsonl and events.jsonl that capture what happened during the run). You interact with it directly only when you need control beyond what agent.run() provides.
QitOS enforces a single-kernel rule: there is exactly one Engine per run. Extensions like parsers, critics, memory adapters, and toolkits attach to this pipeline — they do not introduce a second execution loop.
How the loop works
Each step follows a fixed sequence:
prepare → decide → act → reduce → critics → check_stop → trace
- prepare —
agent.prepare(state) formats state into the model-ready prompt text.
- decide —
agent.decide(state, observation) is checked first; if it returns None, the Engine calls the LLM via the configured parser.
- act — Tool calls in the decision’s
actions list are executed against the ToolRegistry.
- reduce —
agent.reduce(state, observation, decision) updates state with the new observation.
- critics — Any registered
Critic instances (post-step evaluators that can approve, stop, or retry the step) evaluate the step; they can trigger a stop or retry.
- check_stop — Budget exhaustion,
FinalResultCriteria, agent.should_stop(), and any custom StopCriteria are evaluated.
- trace — The step record and events are written to the
TraceWriter.
Constructor
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],
)
| Parameter | Type | Description |
|---|
agent | AgentModule | Required. The policy module to execute. |
budget | RuntimeBudget | None | Step/time/token limits. Defaults to RuntimeBudget(max_steps=10). |
parser | Parser | None | Parses raw model output into a Decision. Must match your prompt format. |
critics | list[Critic] | None | Post-step evaluators that can stop or retry a step. |
stop_criteria | list[StopCriteria] | None | Stop conditions. Defaults to [FinalResultCriteria()]. |
env | Env | None | Environment providing reset/observe/step/is_terminal/close. |
trace_writer | TraceWriter | None | Writes manifest.json, events.jsonl, steps.jsonl for this run. |
hooks | list[EngineHook] | None | Hook instances called on lifecycle events. |
render_hooks | list | None | Render hooks (e.g., terminal UI). Appended to hooks internally. |
history_policy | HistoryPolicy | None | Controls how the in-run conversation history is managed. |
recovery_policy | RecoveryPolicy | None | Controls how the Engine responds to step failures. |
Prefer calling agent.run() for single-run workflows. Use Engine directly when you need to reuse an Engine across multiple runs or configure hooks dynamically between runs.
Engine.run(task)
result = engine.run(task)
Task object. Returns an EngineResult.
When you pass a Task, the Engine extracts the budget from task.budget and overrides the Engine’s own budget for that run. It also orchestrates resource staging and environment lifecycle (reset, observe, close) automatically.
EngineResult
@dataclass
class EngineResult(Generic[StateT]):
state: StateT
records: List[StepRecord]
events: List[RuntimeEvent]
step_count: int
task_result: Optional[TaskResult]
| Field | Description |
|---|
state | Final typed state after the run. Check state.final_result and state.stop_reason. |
records | One StepRecord per step, containing decision, observation, and diffs. |
events | All RuntimeEvent objects emitted during the run. |
step_count | Number of steps executed (len(records)). |
task_result | Structured outcome including success flag and criterion results. |
result = engine.run("summarize the paper")
print(result.state.final_result) # the agent's final answer
print(result.state.stop_reason) # why the loop ended
print(result.step_count) # how many steps ran
print(result.task_result.success) # whether success criteria passed
Hooks
Hooks observe and react to lifecycle events without modifying Engine internals. They implement EngineHook and are called at on_before_step and on_after_step boundaries.
engine.register_hook(my_hook)
engine.unregister_hook(my_hook)
engine.clear_hooks()
hooks parameter, or at run time via agent.run(hooks=[...]).
Budget exhaustion
When the step budget, wall-clock time limit, or token budget is exceeded, the Engine sets state.stop_reason to the appropriate value and emits an END event. The run terminates gracefully and EngineResult is still returned — inspect state.stop_reason to detect this case.
from qitos.engine.states import RuntimeBudget
engine = Engine(
agent=agent,
budget=RuntimeBudget(
max_steps=30,
max_runtime_seconds=120.0,
max_tokens=50_000,
),
)
Building an Engine from AgentModule
AgentModule.build_engine() is a convenience factory that creates an Engine pre-bound to the agent:
engine = agent.build_engine(
budget=RuntimeBudget(max_steps=15),
trace_writer=trace_writer,
)
result = engine.run(task)
Engine(agent=agent, ...) and is what agent.run() calls internally.
AsyncEngine
AsyncEngine provides non-blocking execution for agent workflows. It wraps the same Engine loop but runs blocking calls in a thread pool, making it safe to use inside asyncio event loops.
from qitos import AsyncEngine
async_engine = AsyncEngine(agent=agent, budget=RuntimeBudget(max_steps=20))
AsyncEngine.arun(task)
Run the agent loop asynchronously. Returns the same EngineResult as Engine.run().
result = await async_engine.arun("analyze the data")
print(result.state.final_result)
AsyncEngine.arun_stream(task)
Run the agent loop and yield structured EngineEvent objects as they occur — ideal for real-time UI updates or streaming progress to a client.
async for event in async_engine.arun_stream("analyze the data"):
print(event.event_type, event.step_id, event.payload)
step_start, step_end), phase transitions (decide, act, reduce, critic, check_stop), and multi-agent events (handoff, delegate, fanout). The stream always begins with run_start and ends with run_end.
EngineEvent
@dataclass
class EngineEvent:
event_type: EngineEventType # step_start, step_end, decide, act, ...
step_id: int
agent_id: str | None # which agent produced this step
phase: RuntimePhase | None
ok: bool
payload: Dict[str, Any]
error: str | None
ts: str # ISO 8601 timestamp
EventStream
EventStream is the async queue that underpins arun_stream(). You can also use it standalone to fan out events to multiple consumers:
from qitos.engine.events import EventStream
stream = EventStream()
q1 = stream.subscribe() # asyncio.Queue for consumer 1
q2 = stream.subscribe() # asyncio.Queue for consumer 2
Async models
When the configured model implements acall() (from AsyncModel), AsyncEngine can invoke it without blocking the event loop. Built-in async model adapters:
from qitos.models import AsyncOpenAICompatibleModel, AsyncOpenAIModel
llm = AsyncOpenAICompatibleModel(
model="qwen-plus",
api_key="...",
base_url="https://...",
)
result = await llm.acall([{"role": "user", "content": "Hello"}])
AsyncEngine.arun() works with any model — sync models are automatically dispatched to a thread pool. Use async model adapters when you need true non-blocking I/O, e.g. in high-concurrency web servers.
