> ## Documentation Index
> Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 评估器与停止条件

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

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

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

## 评估器

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

### 评估器契约

```python theme={null}
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()` 时传入：

```python theme={null}
result = agent.run(
    task="...",
    max_steps=10,
    critics=[MyScoreCritic(), MyGroundingCritic()],
    return_state=True,
)
```

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

```python theme={null}
from qitos.engine.engine import Engine

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

### 自定义评估器示例

```python theme={null}
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 契约

```python theme={null}
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` 被设置为非空字符串时停止。它是默认唯一启用的条件。

```python theme={null}
from qitos.engine.stop_criteria import FinalResultCriteria
```

**`MaxStepsCriteria`**

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

```python theme={null}
from qitos.engine.stop_criteria import MaxStepsCriteria

criterion = MaxStepsCriteria(max_steps=15)
```

**`MaxRuntimeCriteria`**

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

```python theme={null}
from qitos.engine.stop_criteria import MaxRuntimeCriteria

criterion = MaxRuntimeCriteria(max_runtime_seconds=120.0)
```

**`StagnationCriteria`**

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

```python theme={null}
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 消耗。

```python theme={null}
from qitos.engine.stop_criteria import MaxTokensCriteria

criterion = MaxTokensCriteria(max_tokens=500_000)
```

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

```python theme={null}
from qitos.harness import resolve_family_preset

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

### 给运行传入停止条件

```python theme={null}
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,
)
```

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

### 自定义条件示例

```python theme={null}
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` 同时表达三种预算：

```python theme={null}
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`    | 连续多步状态未发生有效变化                                     |