跳转到主要内容

​
StateSchema

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

​
内置字段

@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_resultagent 的最终答案。设置后会触发默认的 FinalResultCriteria。
stop_reason运行结束原因，由 state.set_stop() 或 Engine 在预算耗尽时写入。
metadata任意附加运行上下文。
metrics跟踪数值型指标的自由字典。

​
继承 StateSchema

通过继承 StateSchema 来增加自己的字段。Engine 会在每个 step record 中通过 state.to_dict() 记录 state diff，因此这些字段也会自动进入 trace。 
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)),
    )

​
让 run 停下来

你可以在 reduce 中调用 state.set_stop()，让 run 在下一次 stop check 时终止。它接受 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 应完成什么、需要哪些资源、以及受什么约束。任何支持 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 字段

字段类型说明
idstr唯一任务 ID，会写入 trace artifacts。
objectivestr传给 init_state 的任务文本。
resourceslist[TaskResource]任务依赖的文件、目录、URL 或 artifacts。
env_specEnvSpec | None声明 environment 类型与能力。
constraintsdict[str, Any]传给 agent 的自由约束信息。
success_criterialist[str]在 TaskResult 中记录的人类可读成功标准。
budgetTaskBudget覆盖 Engine 默认 budget 的任务级预算。
inputsdict[str, Any]提供给 agent 的结构化输入。
metadatadict[str, Any]写入 trace manifest 的自由元数据。

​
TaskBudget

TaskBudget 用来指定单个任务的步数、时间和 token 限制。传入 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 或先前 run 的 artifact。Engine 会在循环开始前验证所有 required resources。 
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)

# 来自先前运行的 artifact
TaskResource(kind="artifact", path="runs/agent_20260407/steps.jsonl")
合法的 kind 值包括："file"、"dir"、"url"、"artifact"。

​
Decision

Decision 表示一次 decide step 的输出，也就是 agent 下一步想做什么。它有四种模式，每种模式都对应一个工厂方法。
执行一个或多个工具动作。
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.",
)
推荐使用 Decision.act()、Decision.final() 等工厂方法，而不是直接构造 Decision。它们会帮你校验必需字段并正确设置 mode。