State 与 Task

理解 QitOS 中的 StateSchema、Task、TaskBudget、TaskResource 和 Decision。

​

StateSchema

StateSchema 是智能体强类型状态的基类。一次运行中，所有钩子都围绕它读取与写入，它是整个运行期间的单一事实来源。

​

内置字段

@dataclass
class StateSchema:
    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)

字段	说明
task	传给 init_state 的任务文本，通常由 Engine 自动设置。
current_step	当前步索引。每步结束后由 Engine 递增。
max_steps	最大允许步数，必须 > 0 且 >= current_step。
final_result	智能体的最终答案。设置后会触发默认的 FinalResultCriteria。
stop_reason	运行结束原因，由 state.set_stop() 或 Engine 在预算耗尽时写入。
metadata	任意附加运行上下文。
metrics	跟踪数值型指标的自由字典。

​

继承 StateSchema

通过继承 StateSchema 来增加自己的字段。Engine 会在每个步骤记录中通过 state.to_dict() 记录状态差异，因此这些字段也会自动进入追踪记录。

from dataclasses import dataclass, field
from typing import Any

from qitos import StateSchema


@dataclass
class ResearchState(StateSchema):
    scratchpad: list[str] = field(default_factory=list)
    sources_visited: list[str] = field(default_factory=list)
    confidence: float = 0.0

然后在 init_state 中初始化：

def init_state(self, task: str, **kwargs: Any) -> ResearchState:
    return ResearchState(
        task=task,
        max_steps=int(kwargs.get("max_steps", 20)),
    )

​

让运行停下来

你可以在 reduce 中调用 state.set_stop()，让运行在下一次停止检查时终止。它接受 StopReason 枚举或等价字符串。

from qitos import StopReason

def reduce(self, state, observation, decision):
    if some_terminal_condition:
        state.set_stop(StopReason.MAX_STEPS_REACHED, final_result="Done.")
    return state

如果是成功完成，更推荐直接设置 state.final_result：

state.final_result = "The answer is 42."

---

​

Task

Task 是结构化任务包，用来描述智能体应完成什么、需要哪些资源、以及受什么约束。任何支持 agent.run() 或 Engine.run() 的地方都可以传入 Task。

from qitos import Task, TaskBudget, TaskResource

task = Task(
    id="research-001",
    objective="Summarize the findings in paper.pdf",
    resources=[
        TaskResource(kind="file", path="paper.pdf", required=True),
    ],
    budget=TaskBudget(max_steps=25, max_runtime_seconds=300.0),
    success_criteria=["Summary covers abstract, methods, and conclusions"],
    constraints={"output_format": "markdown"},
)

result = agent.run(task, return_state=True)

​

Task 字段

字段	类型	说明
id	str	唯一任务标识，会写入追踪记录产物。
objective	str	传给 init_state 的任务文本。
resources	list[TaskResource]	任务依赖的文件、目录、URL 或产物。
env_spec	EnvSpec | None	声明环境类型与能力。
constraints	dict[str, Any]	传给智能体的自由约束信息。
success_criteria	list[str]	在 TaskResult 中记录的人类可读成功标准。
budget	TaskBudget	覆盖 Engine 默认预算的任务级预算。
inputs	dict[str, Any]	提供给智能体的结构化输入。
metadata	dict[str, Any]	写入追踪记录清单的自由元数据。

​

TaskBudget

TaskBudget 用来指定单个任务的步数、时间和令牌限制。传入 Task 时，它会优先于 Engine 默认的 RuntimeBudget。

from qitos import TaskBudget

budget = TaskBudget(
    max_steps=30,
    max_runtime_seconds=180.0,
    max_tokens=100_000,
)

留空的字段会继承 Engine 默认值。

​

TaskResource

TaskResource 描述任务依赖的文件、目录、URL 或先前运行的产物。Engine 会在循环开始前验证所有必需资源。

from qitos import TaskResource

# 必需的本地文件
TaskResource(kind="file", path="data/corpus.txt", required=True)

# 可选目录
TaskResource(kind="dir", path="output/", required=False)

# 远程 URL
TaskResource(kind="url", uri="https://arxiv.org/abs/2501.12345", required=True)

# 来自先前运行的产物
TaskResource(kind="artifact", path="runs/agent_20260407/steps.jsonl")

合法的 kind 值包括：file、dir、url、artifact。

---

​

Decision

Decision 表示一次决策步骤的输出，也就是智能体下一步想做什么。它有四种模式，每种模式都对应一个工厂方法。

act

执行一个或多个工具动作。

from qitos import Decision, Action

decision = Decision.act(
    actions=[Action(name="read_file", args={"path": "paper.pdf"})],
    rationale="Need to read the file before summarizing.",
)

final

表示智能体已经得到最终答案，运行应结束。

decision = Decision.final(
    answer="The paper proposes a novel attention mechanism that reduces memory usage by 40%.",
    rationale="Finished reading all sections.",
)

wait

本步暂时不执行动作，常用于等待外部事件或环境信号。

decision = Decision.wait(rationale="Waiting for environment to reset.")

branch

提出多个候选决策，交给 BranchSelector 评估。

decision = Decision.branch(
    candidates=[
        Decision.act([Action(name="search", args={"query": "attention mechanism"})]),
        Decision.act([Action(name="read_file", args={"path": "intro.txt"})]),
    ],
    rationale="Exploring two strategies in parallel.",
)

推荐使用 Decision.act()、Decision.final() 等工厂方法，而不是直接构造 Decision。它们会帮你校验必需字段并正确设置 mode。