Agent 与执行

AgentModule 管理策略与状态变更，Engine 驱动 observe、decide、act、reduce 与停止。Decision.act 声明动作，Decision.final 结束策略。EngineResult 保存状态及 typed 记录，reduce 收到字典观察。完整组合见自定义 Agent 教程。RuntimeComposition 默认 store 只在进程内；AgentModule.run 保留为程序化便捷路径。完整可运行教程 / Complete tutorial · API index 以下签名和字段由固定源码提取；签名是参考，不是可直接运行的程序。每个条目的源码链接绑定同一 runtime baseline。类型中的 Any 不代表任意对象均受支持，应结合上述行为契约和教程使用。

AgentModule

from qitos import AgentModule

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

# NotesAgent subclasses AgentModule in the complete tutorial.
agent = NotesAgent()
result = Engine(agent, runtime=RuntimeComposition()).session("Index notes").run()

Canonical policy contract for step-based agents.

AgentModule(tool_registry: Any=None, toolset: Any=None, llm: Any=None, model_parser: Any=None, model_protocol: Any=None, memory: Memory | None=None, history: History | None=None, mcp_servers: List[Any] | None=None, **config: Any) -> Any (see behavior contract)

Parameter	Type	Default
`tool_registry`	`Any`	`None`
`toolset`	`Any`	`None`
`llm`	`Any`	`None`
`model_parser`	`Any`	`None`
`model_protocol`	`Any`	`None`
`memory`	`Memory \| None`	`None`
`history`	`History \| None`	`None`
`mcp_servers`	`List[Any] \| None`	`None`

Field	Type	Default
`name`	`str`	`'agent'`
`handoff_targets`	`List[str] \| None`	`None`

AgentModule.init_state

init_state(task: str, **kwargs: Any) -> StateT

Parameter	Type	Default
`task`	`str`	`required`

Create and return the initial typed state for a run.

Source

AgentModule.decide

decide(state: StateT, observation: ObservationT) -> Optional[Decision[ActionT]]

Parameter	Type	Default
`state`	`StateT`	`required`
`observation`	`ObservationT`	`required`

Optional custom decision hook. Return None to use Engine model decision.

Source

AgentModule.reduce

reduce(state: StateT, observation: ObservationT, decision: Decision[ActionT]) -> StateT

Parameter	Type	Default
`state`	`StateT`	`required`
`observation`	`ObservationT`	`required`
`decision`	`Decision[ActionT]`	`required`

Reduce observation (including action/env outputs) into next state.

Source

AgentModule.should_stop

should_stop(state: StateT) -> bool

Parameter	Type	Default
`state`	`StateT`	`required`

Optional additional stop condition.

Source

AgentModule.run

run(task: str | Task, return_state: bool=False, hooks: List[Any] | None=None, render_hooks: List[Any] | None=None, engine_kwargs: Dict[str, Any] | None=None, workspace: str | None=None, max_steps: int | None=None, env: Any=None, parser: Any=None, protocol: Any=None, search: Any=None, critics: List[Any] | None=None, stop_criteria: List[Any] | None=None, history_policy: Any=None, context_config: Any=None, trace: Any=None, render: Any=None, trace_logdir: str='./runs', trace_prefix: str | None=None, theme: str='research', run_spec: RunSpec | Dict[str, Any] | None=None, experiment_spec: ExperimentSpec | Dict[str, Any] | None=None, **state_kwargs: Any) -> Any

Parameter	Type	Default
`task`	`str \| Task`	`required`
`return_state`	`bool`	`False`
`hooks`	`List[Any] \| None`	`None`
`render_hooks`	`List[Any] \| None`	`None`
`engine_kwargs`	`Dict[str, Any] \| None`	`None`
`workspace`	`str \| None`	`None`
`max_steps`	`int \| None`	`None`
`env`	`Any`	`None`
`parser`	`Any`	`None`
`protocol`	`Any`	`None`
`search`	`Any`	`None`
`critics`	`List[Any] \| None`	`None`
`stop_criteria`	`List[Any] \| None`	`None`
`history_policy`	`Any`	`None`
`context_config`	`Any`	`None`
`trace`	`Any`	`None`
`render`	`Any`	`None`
`trace_logdir`	`str`	`'./runs'`
`trace_prefix`	`str \| None`	`None`
`theme`	`str`	`'research'`
`run_spec`	`RunSpec \| Dict[str, Any] \| None`	`None`
`experiment_spec`	`ExperimentSpec \| Dict[str, Any] \| None`	`None`

Execute task with Engine using plain text objective or structured Task.

Source

StateSchema

from qitos import StateSchema

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

from dataclasses import dataclass
@dataclass
class MyState(StateSchema):
    completed: int = 0
state = MyState(task="Index notes", max_steps=3)

Canonical typed state base for AgentModule.

Field	Type	Default
`schema_version`	`int`	`1`
`task`	`str`	`''`
`current_step`	`int`	`0`
`max_steps`	`int`	`10`
`final_result`	`Optional[str]`	`None`
`stop_reason`	`Optional[str]`	`None`
`metadata`	`Dict[str, Any]`	`field(default_factory=dict)`
`metrics`	`Dict[str, Any]`	`field(default_factory=dict)`
`migration_registry`	`ClassVar[StateMigrationRegistry]`	`StateMigrationRegistry()`

Task

from qitos import Task

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

task = Task(objective="Index the notes")
print(task.objective)

Task package with objective, resources, and environment requirements.

Field	Type	Default
`id`	`str`	`required`
`objective`	`str`	`required`
`inputs`	`Dict[str, Any]`	`dc_field(default_factory=dict)`
`resources`	`List[TaskResource]`	`dc_field(default_factory=list)`
`env_spec`	`Optional[EnvSpec]`	`None`
`constraints`	`Dict[str, Any]`	`dc_field(default_factory=dict)`
`success_criteria`	`List[str]`	`dc_field(default_factory=list)`
`budget`	`TaskBudget`	`dc_field(default_factory=TaskBudget)`
`metadata`	`Dict[str, Any]`	`dc_field(default_factory=dict)`

Decision

from qitos import Decision

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

decision = Decision.act([Action(name="summarize_note", args={"index": 0})])
finished = Decision.final("Session")

Field	Type	Default
`mode`	`DecisionMode`	`required`
`actions`	`List[ActionT]`	`field(default_factory=list)`
`final_answer`	`Optional[str]`	`None`
`rationale`	`Optional[str]`	`None`
`meta`	`Dict[str, Any]`	`field(default_factory=dict)`
`candidates`	`List['Decision[ActionT]']`	`field(default_factory=list)`

Decision.act

act(actions: List[ActionT], rationale: Optional[str]=None, meta: Optional[Dict[str, Any]]=None) -> 'Decision[ActionT]'

Parameter	Type	Default
`actions`	`List[ActionT]`	`required`
`rationale`	`Optional[str]`	`None`
`meta`	`Optional[Dict[str, Any]]`	`None`

Source

Decision.final

final(answer: str, rationale: Optional[str]=None, meta: Optional[Dict[str, Any]]=None) -> 'Decision[ActionT]'

Parameter	Type	Default
`answer`	`str`	`required`
`rationale`	`Optional[str]`	`None`
`meta`	`Optional[Dict[str, Any]]`	`None`

Source

Action

from qitos import Action

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

action = Action(name="summarize_note", args={"index": 0})
print(action.name)

Normalized action contract emitted by policy and consumed by executor.

Field	Type	Default
`name`	`str`	`required`
`args`	`Dict[str, Any]`	`field(default_factory=dict)`
`kind`	`ActionKind`	`ActionKind.TOOL`
`action_id`	`Optional[str]`	`None`
`timeout_s`	`Optional[float]`	`None`
`max_retries`	`int`	`0`
`idempotent`	`bool`	`True`
`classification`	`str`	`'default'`
`metadata`	`Dict[str, Any]`	`field(default_factory=dict)`

Engine

from qitos import Engine

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

engine = Engine(NotesAgent(), runtime=RuntimeComposition())
result = engine.session("Index notes").run()

Single execution kernel for all AgentModule workflows.

Engine(agent: AgentModule[StateT, ObservationT, ActionT], agent_registry: Optional[Any]=None, budget: Optional[RuntimeBudget]=None, delegate_depth: int=0, shared_memory: Any=None, validation_gate: Optional[StateValidationGate]=None, recovery_handler: Optional[RecoveryHandler]=None, recovery_policy: Optional[RecoveryPolicy]=None, trace_writer: Optional[TraceWriter]=None, parser: Optional[Parser[ActionT]]=None, protocol: Any=None, stop_criteria: Optional[List[StopCriteria]]=None, branch_selector: Optional[BranchSelector[StateT, ObservationT, ActionT]]=None, search: Optional[Search[StateT, ObservationT, ActionT]]=None, critics: Optional[List[Critic]]=None, env: Optional[Env]=None, history_policy: Optional[HistoryPolicy]=None, hooks: Optional[List[EngineHook]]=None, render_hooks: Optional[List[Any]]=None, context_config: Optional[ContextConfig | Dict[str, Any]]=None, cache_backend: Optional[Any]=None, checkpoint_manager: Optional[Any]=None, checkpoint_store: Optional[CheckpointStore]=None, checkpoint_durability: DurabilityMode=DurabilityMode.SYNC, permission_pipeline: Optional[Any]=None, read_before_write_enforcer: Optional[Any]=None, permission_interaction_callback: Optional[Any]=None, loop_detector: Optional[ToolCallLoopDetector]=None, tracing_provider: Optional[Any]=None, interceptors: Optional[List[ToolInterceptor]]=None, auto_approve: bool=False, action_execution_policy: Optional[Any]=None, runtime: Optional[RuntimeComposition]=None) -> Any (see behavior contract)

Parameter	Type	Default
`agent`	`AgentModule[StateT, ObservationT, ActionT]`	`required`
`agent_registry`	`Optional[Any]`	`None`
`budget`	`Optional[RuntimeBudget]`	`None`
`delegate_depth`	`int`	`0`
`shared_memory`	`Any`	`None`
`validation_gate`	`Optional[StateValidationGate]`	`None`
`recovery_handler`	`Optional[RecoveryHandler]`	`None`
`recovery_policy`	`Optional[RecoveryPolicy]`	`None`
`trace_writer`	`Optional[TraceWriter]`	`None`
`parser`	`Optional[Parser[ActionT]]`	`None`
`protocol`	`Any`	`None`
`stop_criteria`	`Optional[List[StopCriteria]]`	`None`
`branch_selector`	`Optional[BranchSelector[StateT, ObservationT, ActionT]]`	`None`
`search`	`Optional[Search[StateT, ObservationT, ActionT]]`	`None`
`critics`	`Optional[List[Critic]]`	`None`
`env`	`Optional[Env]`	`None`
`history_policy`	`Optional[HistoryPolicy]`	`None`
`hooks`	`Optional[List[EngineHook]]`	`None`
`render_hooks`	`Optional[List[Any]]`	`None`
`context_config`	`Optional[ContextConfig \| Dict[str, Any]]`	`None`
`cache_backend`	`Optional[Any]`	`None`
`checkpoint_manager`	`Optional[Any]`	`None`
`checkpoint_store`	`Optional[CheckpointStore]`	`None`
`checkpoint_durability`	`DurabilityMode`	`DurabilityMode.SYNC`
`permission_pipeline`	`Optional[Any]`	`None`
`read_before_write_enforcer`	`Optional[Any]`	`None`
`permission_interaction_callback`	`Optional[Any]`	`None`
`loop_detector`	`Optional[ToolCallLoopDetector]`	`None`
`tracing_provider`	`Optional[Any]`	`None`
`interceptors`	`Optional[List[ToolInterceptor]]`	`None`
`auto_approve`	`bool`	`False`
`action_execution_policy`	`Optional[Any]`	`None`
`runtime`	`Optional[RuntimeComposition]`	`None`

Engine.session

session(task: str | Task, session_id: Any=None) -> Any

Parameter	Type	Default
`task`	`str \| Task`	`required`
`session_id`	`Any`	`None`

Create one durable Session facade using this Engine composition.

Source

Engine.run

run(task: str | Task, **kwargs: Any) -> EngineResult[StateT]

Parameter	Type	Default
`task`	`str \| Task`	`required`

Source

Engine.restore

restore(session_id: Any, *, resolvers: Any=None, runtime: Optional[RuntimeComposition]=None) -> Any

Parameter	Type	Default
`session_id`	`Any`	`required`
`resolvers`	`Any`	`None`
`runtime`	`Optional[RuntimeComposition]`	`None`

Restore a Session in a fresh Engine through explicit resolvers.

Source

EngineResult

from qitos.engine.engine import EngineResult

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

result = session.run()
print(result.state.final_result, result.state.stop_reason)
for record in result.records:
    for tool_result in record.action_results:
        print(tool_result.tool_name, tool_result.status)

Field	Type	Default
`state`	`StateT`	`required`
`records`	`List[StepRecord]`	`required`
`events`	`List[RuntimeEvent]`	`required`
`step_count`	`int`	`required`
`task_result`	`Optional[TaskResult]`	`None`
`runtime_seconds`	`float`	`0.0`
`total_tokens`	`int`	`0`
`run_id`	`str`	`''`
`critic_traces`	`List[CriticTrace]`	`field(default_factory=list)`
`handoff_traces`	`List[HandoffTrace]`	`field(default_factory=list)`
`failure`	`Optional[Dict[str, Any]]`	`None`

RuntimeBudget

from qitos.engine.states import RuntimeBudget

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

budget = RuntimeBudget(max_steps=3)
engine = Engine(NotesAgent(), budget=budget)

Field	Type	Default
`max_steps`	`int`	`10`
`max_runtime_seconds`	`Optional[float]`	`None`
`max_tokens`	`Optional[int]`	`None`
`max_model_requests`	`Optional[int]`	`None`

StopReason

from qitos import StopReason

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

print([reason.value for reason in StopReason])

RuntimeComposition

from qitos.engine.runtime import RuntimeComposition

Source @ 19f6258 用法与可执行示例用法片段：接续上方完整教程中的对象，不是独立程序。

runtime = RuntimeComposition()
engine = Engine(NotesAgent(), runtime=runtime)
# The default checkpoint store is process-local.

Process-local Engine components plus their serializable description.

Field	Type	Default
`checkpoint_store`	`Optional[CheckpointStore]`	`None`
`resolvers`	`ResolverRegistry`	`field(default_factory=ResolverRegistry)`
`durability_mode`	`DurabilityMode`	`DurabilityMode.SYNC`
`lifecycle_policy`	`LifecyclePolicy`	`field(default_factory=LifecyclePolicy)`
`snapshot_components`	`tuple[RuntimeSnapshotComponent, ...]`	`()`
`event_sink`	`Any`	`None`
`event_sink_failure_policy`	`Any`	`None`
`event_sink_view`	`Any`	`None`
`tool_execution_policy`	`Any`	`None`
`launch_metadata`	`Mapping[str, Any]`	`field(default_factory=dict)`
`context_model_runtime`	`Optional[ContextModelRuntime]`	`None`
`work_runtime`	`Any`	`None`
`event_sink_reports`	`list[Any]`	`field(default_factory=list, init=False)`

​AgentModule

​AgentModule.init_state

​AgentModule.decide

​AgentModule.reduce

​AgentModule.should_stop

​AgentModule.run

​StateSchema

​Task

​Decision

​Decision.act

​Decision.final

​Action

​Engine

​Engine.session

​Engine.run

​Engine.restore

​EngineResult

​RuntimeBudget

​StopReason

​RuntimeComposition

AgentModule

AgentModule.init_state

AgentModule.decide

AgentModule.reduce

AgentModule.should_stop

AgentModule.run

StateSchema

Task

Decision

Decision.act

Decision.final

Action

Engine

Engine.session

Engine.run

Engine.restore

EngineResult

RuntimeBudget

StopReason

RuntimeComposition