评估器与停止条件

使用评估器、停止条件与恢复策略来约束智能体运行。

Engine 提供两套互补机制来控制运行的生命周期：

• 评估器（critic）：在归约（reduce，将观测结果和决策归约回状态的钩子）之后验证每一步，可以允许继续、强制停止，或要求重试本步。
• 停止条件：在评估器通过后进行统一停止判断，根据状态与运行时指标决定是否结束循环。

​

评估器

评估器会接收当前状态、本步决策（decision，智能体每步的结构化决策）和动作（action，标准化工具调用）结果，然后返回一个结构化裁定字典。

​

评估器契约

from abc import ABC, abstractmethod
from typing import Any, Dict
from qitos.engine.critic import Critic
from qitos.core.decision import Decision

class Critic(ABC):
    @abstractmethod
    def evaluate(
        self,
        state: Any,
        decision: Decision[Any],
        results: list[Any],
    ) -> Dict[str, Any]:
        """Return a critic decision dict.

        Supported keys:
        - action: "continue" | "stop" | "retry"
        - reason: str
        - score: float
        - details: dict
        """

​

支持的评估器动作

action	Engine 行为
"continue"	接受本步，继续运行
"stop"	把 state.stop_reason 设为 StopReason.CRITIC_STOP 并结束
"retry"	重试当前步骤，保留观测结果（observation，每步后智能体接收的结构化观察结果）并递增步数计数器

除 "stop" 和 "retry" 外，其他值都会被视作 "continue"。

​

如何给运行添加评估器

可以直接在 agent.run() 时传入：

result = agent.run(
    task="...",
    max_steps=10,
    critics=[MyScoreCritic(), MyGroundingCritic()],
    return_state=True,
)

也可以在直接构造 Engine 时传入：

from qitos.engine.engine import Engine

engine = Engine(agent=agent, critics=[MyScoreCritic()])
result = engine.run("my task")

​

自定义评估器示例

from typing import Any, Dict
from qitos.engine.critic import Critic
from qitos.core.decision import Decision


class VerificationCritic(Critic):
    """Stop or retry when the verification command fails."""

    def evaluate(
        self,
        state: Any,
        decision: Decision[Any],
        results: list[Any],
    ) -> Dict[str, Any]:
        for result in results:
            if isinstance(result, dict):
                returncode = int(result.get("returncode", 0))
                if returncode != 0:
                    return {
                        "action": "retry",
                        "reason": f"command failed with returncode={returncode}",
                        "score": 0.0,
                    }
        return {"action": "continue", "score": 1.0}

评估器输出会被记录进追踪记录（trace）的 step.critic_outputs，并可在 qita 中查看。

---

​

停止条件

停止条件会在每步的评估器阶段之后执行。每个条件都会接收当前状态、步数与运行时信息。

​

StopCriteria 契约

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional, Tuple
from qitos.engine.stop_criteria import StopCriteria
from qitos.core.errors import StopReason

class StopCriteria(ABC):
    @abstractmethod
    def should_stop(
        self,
        state: Any,
        step_count: int,
        runtime_info: Optional[Dict[str, Any]] = None,
    ) -> Tuple[bool, Optional[StopReason], Optional[str]]:
        """Return (should_stop, reason, detail)."""

​

内置条件

QitOS 在 qitos.engine.stop_criteria 中提供四个常用实现：

FinalResultCriteria（默认）

当 state.final_result 被设置为非空字符串时停止。它是默认唯一启用的条件。

from qitos.engine.stop_criteria import FinalResultCriteria

MaxStepsCriteria

当步数达到 max_steps 时停止。一般由 RuntimeBudget 自动注入，不需要手动创建。

from qitos.engine.stop_criteria import MaxStepsCriteria

criterion = MaxStepsCriteria(max_steps=15)

MaxRuntimeCriteria

当墙钟时间超过阈值时停止。

from qitos.engine.stop_criteria import MaxRuntimeCriteria

criterion = MaxRuntimeCriteria(max_runtime_seconds=120.0)

StagnationCriteria

当状态连续若干步没有有效变化时停止。它使用一个签名函数来判断是否有变化。

from qitos.engine.stop_criteria import StagnationCriteria

criterion = StagnationCriteria(
    max_stagnant_steps=3,
    signature_fn=lambda s: (s.final_result, getattr(s, "cursor", None)),
)

MaxTokensCriteria

当累计 token 用量超过预算时停止。Engine 在每个步骤中通过 runtime_info 传递 total_tokens，因此该准则追踪整个运行的实际 token 消耗。

from qitos.engine.stop_criteria import MaxTokensCriteria

criterion = MaxTokensCriteria(max_tokens=500_000)

当你需要独立于步数或时间的硬性 token 预算时使用它。可以搭配黄金预设的推荐值：

from qitos.harness import resolve_family_preset

preset = resolve_family_preset("qwen")
criterion = MaxTokensCriteria(max_tokens=preset.recommended_max_tokens or 500_000)

​

给运行传入停止条件

from qitos.engine.stop_criteria import FinalResultCriteria, MaxRuntimeCriteria, StagnationCriteria

result = agent.run(
    task="...",
    max_steps=20,
    stop_criteria=[
        FinalResultCriteria(),
        MaxRuntimeCriteria(max_runtime_seconds=300.0),
        StagnationCriteria(max_stagnant_steps=4),
    ],
    return_state=True,
)

一旦你手动传入 stop_criteria，就会替换默认的 FinalResultCriteria。如果仍希望在 final_result 被设置时自动停止，请把 FinalResultCriteria() 也放进去。

​

自定义条件示例

from typing import Any, Dict, Optional, Tuple
from qitos.engine.stop_criteria import StopCriteria
from qitos.core.errors import StopReason


class MinEvidenceCriteria(StopCriteria):
    """Only stop after at least N evidence items are collected."""

    def __init__(self, min_evidence: int = 3):
        self.min_evidence = min_evidence

    def should_stop(
        self,
        state: Any,
        step_count: int,
        runtime_info: Optional[Dict[str, Any]] = None,
    ) -> Tuple[bool, Optional[StopReason], Optional[str]]:
        evidence = getattr(state, "evidence", [])
        final_result = getattr(state, "final_result", None)
        if final_result and len(evidence) >= self.min_evidence:
            return True, StopReason.FINAL, f"evidence={len(evidence)} >= min={self.min_evidence}"
        return False, None, None

---

​

用 TaskBudget 表达预算（budget）停止

对于结构化任务，通常用 TaskBudget 同时表达三种预算：

from qitos.core.task import Task, TaskBudget

task = Task(
    id="research-001",
    objective="Summarize the article at the given URL.",
    budget=TaskBudget(
        max_steps=15,
        max_runtime_seconds=180.0,
        max_tokens=8192,
    ),
)

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

当传入 Task 时，Engine 会在运行开始前把 TaskBudget 应用到内部 RuntimeBudget 上。

​

StopReason 值

StopReason 是 qitos.core.errors 中的字符串枚举。每次运行结束后都会写入 state.stop_reason。

值	含义
success	运行成功完成
final	state.final_result 被设置，触发 FinalResultCriteria
budget_steps	达到 max_steps
budget_time	超过 max_runtime_seconds
budget_tokens	超过 max_tokens
critic_stop	评估器返回了 action: "stop"
stagnation	连续多步状态未发生有效变化