API 参考 - QitOS

AgentModule

AgentModule 是 QitOS 的策略层。你通过继承它来定义智能体的状态形状、系统提示词、决策逻辑与归约规则；真正的执行循环由 Engine（执行内核）驱动。

class AgentModule(ABC, Generic[StateT, ObservationT, ActionT])

构造函数

def __init__(
    self,
    tool_registry: Any = None,
    llm: Any = None,
    model_parser: Any = None,
    memory: Memory | None = None,
    history: History | None = None,
    **config: Any,
)

参数	类型	说明
`tool_registry`	`ToolRegistry \| None`	智能体可调用的工具注册表
`llm`	`Any`	用于默认模型决策路径的大模型可调用对象
`model_parser`	`Any`	解析器，把输出解析成 `Decision`（智能体每步的结构化决策）
`memory`	`Memory \| None`	可选记忆适配器
`history`	`History \| None`	可选历史适配器
`**config`	`Any`	额外关键字参数，保存在 `self.config`

钩子只要求实现 init_state 与 reduce；其余钩子全部可选。

init_state（必需）

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

创建并返回本次运行的初始状态。

reduce（必需）

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

把当前观测结果（每步后智能体接收的结构化观察结果）与决策（智能体每步的结构化决策）折叠进下一步状态。

build_system_prompt

def build_system_prompt(self, state: StateT) -> str | None

返回动态系统提示词；默认返回 None。

prepare

def prepare(self, state: StateT) -> str

把状态转成模型输入文本；默认是 str(state)。

decide

def decide(
    self,
    state: StateT,
    observation: ObservationT,
) -> Decision[ActionT] | None

自定义决策钩子。返回 Decision 时跳过默认模型调用；返回 None 时继续走 Engine 的模型路径。

should_stop

def should_stop(self, state: StateT) -> bool

额外停止条件；默认返回 False。

.run() 方法

def run(
    self,
    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,
    search: Any = None,
    critics: List[Any] | None = None,
    stop_criteria: List[Any] | None = None,
    history_policy: Any = None,
    trace: Any = None,
    render: Any = None,
    trace_logdir: str = "./runs",
    trace_prefix: str | None = None,
    theme: str = "research",
    **state_kwargs: Any,
) -> Any

这是最常用入口。它会构建 Engine、执行任务，并默认返回 state.final_result；当 return_state=True 时，返回完整 EngineResult。

Engine

Engine 是执行内核，负责阶段循环、工具执行、恢复、追踪与停止条件评估。

class Engine(Generic[StateT, ObservationT, ActionT])

核心构造参数包括：

agent
budget（预算）
parser
stop_criteria
critics（评估器）
env
history_policy
trace_writer
hooks（钩子）

最常用方法：

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

EngineResult

@dataclass
class EngineResult(Generic[StateT]):
    state: StateT
    records: List[StepRecord]
    events: List[RuntimeEvent]
    step_count: int
    task_result: Optional[TaskResult] = None

其中：

state：最终强类型状态
records：每步 StepRecord
events：所有运行时事件
step_count：执行步数
task_result：结构化任务结果

Decision

Decision 是决策阶段的规范输出（智能体每步的结构化决策）。推荐用工厂方法构造：

Decision.act(...)
Decision.final(...)
Decision.wait(...)
Decision.branch(...)

四种模式分别对应：

"act"：执行动作（标准化工具调用）
"final"：给出最终答案并结束
"wait"：本步不执行动作
"branch"：提出多个候选决策

Task / State / Action

常用基础数据结构包括：

StateSchema
Task
TaskBudget
TaskResource
Action
StopReason

它们共同定义了一次运行的输入、状态与输出语义。

Documentation Index