Critic 系统

Engine 每一步的执行都流经一条管线：decide、act、reduce，然后是 critic。Critic 位于 reduce 与 check_stop 之间，为你提供了一个结构化的钩子，可以审查智能体的最新决策及其结果、分配质量评分，并在必要时强制重试或终止执行。本教程涵盖完整的 Critic API：CriticResult 契约、内置 Critic、@critic 装饰器、指令补丁与状态补丁，以及组合多个 Critic 的方式。

Step 1: CriticResult 契约

每个 Critic 都返回一个 CriticResult——一个告知 Engine 下一步操作的数据类。

from qitos.engine.critic_result import CriticResult

各字段说明：

字段	类型	默认值	说明
`action`	`str`	`"continue"`	取值为 `"continue"`、`"stop"` 或 `"retry"`。
`reason`	`str`	`""`	人类可读的解释。
`score`	`float`	`1.0`	质量评分，范围 0.0 到 1.0，越高越好。
`details`	`dict`	`{}`	额外的结构化数据，用于日志或遥测。
`instruction_patch`	`str \| None`	`None`	在下一次迭代时追加到智能体系统提示词末尾（仅在 `action="retry"` 时生效）。
`state_patch`	`dict \| None`	`None`	在下一次迭代前合并到智能体状态中的键值对（仅在 `action="retry"` 时生效）。
`modified_prompt`	`str \| None`	`None`	下一次迭代的完整替换系统提示词（仅在 `action="retry"` 时生效）。

三种 action 控制循环行为：

"continue"——正常进入下一步。
"stop"——立即终止 Engine。状态的停止原因会被设为 StopReason.CRITIC_STOP。
"retry"——丢弃当前步骤并重新运行 decide。任何 instruction_patch 或 state_patch 会在重试前应用。

最简的 continue 结果：

result = CriticResult(action="continue", reason="all good", score=1.0)

带指令补丁的重试：

result = CriticResult(
    action="retry",
    reason="output was incomplete",
    score=0.3,
    instruction_patch="Make sure to include all required fields in your answer.",
)

强制停止：

result = CriticResult(
    action="stop",
    reason="safety policy violation",
    score=0.0,
)

Step 2: 使用内置 Critic

QitOS 在 qitos.kit.critic 中提供了开箱即用的 Critic。

PassThroughCritic

最简单的 Critic——始终以满分继续。

from qitos.kit.critic import PassThroughCritic

critic = PassThroughCritic()
# evaluate() 始终返回 {"action": "continue", "reason": "pass", "score": 1.0}

创建 Engine 时若不指定 critics，这就是默认行为。它可用作空操作占位符，或作为组合链中由后续 Critic 执行真正工作的基础。

SelfReflectionCritic

检查工具结果中是否包含错误，并在可配置的重试上限内自动重试。

from qitos.kit.critic import SelfReflectionCritic

critic = SelfReflectionCritic(max_retries=2)

行为逻辑：

如果任何工具结果包含 {"error": ...} 键且重试次数未耗尽，返回 action="retry"、score=0.2。
如果错误持续超过 max_retries，返回 action="stop"、score=0.0。
如果没有发现错误，返回 action="continue"、score=1.0。

将其传入 Engine：

from qitos.engine import Engine

engine = Engine(
    agent=my_agent,
    critics=[SelfReflectionCritic(max_retries=3)],
    # ... 其他 Engine 配置
)

ReActSelfReflectionCritic

专为 ReAct 智能体设计的增强版本。遇到错误时，它会构建一条结构化的反思笔记，描述失败的动作与观察到的错误，并将其追加到状态的 metadata 中，以便 LLM 在下次重试时能够从失败中学习。

from qitos.kit.critic import ReActSelfReflectionCritic

critic = ReActSelfReflectionCritic(max_retries=2)

函数式等价物

每个内置 Critic 都同时提供了装饰器函数版本：

from qitos.kit.critic import pass_through_critic, self_reflection_critic

# 这些是 @critic 装饰的函数——详见 Step 3

Step 3: 使用 @critic 装饰器编写自定义 Critic

@critic 装饰器可将任意普通函数转换为 Critic 实例。该函数接收 (state, decision, results) 参数，可返回简写形式或完整的 CriticResult。

from qitos.engine.critic_decorator import critic

快速返回简写：

返回值	含义
`"continue"`	正常继续
`("stop", "reason")`	带原因的终止
`("retry", "reason")`	使用相同提示词重试
`("retry", "reason", instruction_patch)`	追加指令后重试
`CriticResult(...)`	完整结构化结果

无参数装饰器：

@critic
def no_empty_answers(state, decision, results):
    """Stop if the final answer is empty."""
    if decision.mode == "final" and not decision.final_answer:
        return "stop", "empty final answer"
    return "continue"

带关键字参数——设置自定义名称和默认评分：

@critic(name="safety", score=0.8)
def safety_check(state, decision, results):
    """Retry if any result looks unsafe."""
    for r in results:
        if isinstance(r, dict) and r.get("flagged"):
            return "retry", "unsafe output detected"
    return "continue"

当函数未显式设置评分时，装饰器的 score 参数将用作默认值。在上面的示例中，"continue" 返回会获得 score=0.8，而非通常的 1.0。将 Critic 接入 Engine：

engine = Engine(
    agent=my_agent,
    critics=[no_empty_answers, safety_check],
    # ... 其他 Engine 配置
)

Step 4: 带指令补丁和状态补丁的 Critic

当 Critic 返回 action="retry" 时，可以通过两种方式指导下一次迭代：

instruction_patch——追加到智能体系统提示词末尾的字符串，为 LLM 提供额外指导。
state_patch——一个字典，其键值通过 setattr 合并到智能体的状态对象中。

这些补丁在 Engine 再次调用 decide 之前就已应用，因此当重试开始时，LLM 和状态都已经更新完毕。

from qitos.engine.critic_result import CriticResult
from qitos.engine.critic_decorator import critic

@critic(name="format_enforcer")
def enforce_json_output(state, decision, results):
    """Retry with guidance if the output is not valid JSON."""
    if decision.mode == "final":
        answer = decision.final_answer or ""
        if not answer.strip().startswith("{"):
            return CriticResult(
                action="retry",
                reason="output is not JSON",
                score=0.2,
                instruction_patch=(
                    "You MUST output valid JSON. "
                    "Do not include any text outside the JSON object."
                ),
            )
    return "continue"

使用 state_patch 注入追踪数据：

@critic(name="retry_tracker")
def track_retries(state, decision, results):
    """Count retries and halt after the limit."""
    metadata = getattr(state, "metadata", {}) or {}
    retries = int(metadata.get("custom_retries", 0))

    has_error = any(isinstance(r, dict) and r.get("error") for r in results)
    if has_error and retries < 5:
        metadata["custom_retries"] = retries + 1
        state.metadata = metadata
        return CriticResult(
            action="retry",
            reason="tool error, retrying",
            score=0.3,
            state_patch={"last_error_type": "tool_failure"},
        )
    if has_error:
        return "stop", "exceeded retry limit"

    return "continue"

也可以使用元组简写配合指令补丁：

@critic
def be_concise(state, decision, results):
    if decision.mode == "final" and len(decision.final_answer or "") > 2000:
        return "retry", "answer too long", "Keep your final answer under 2000 characters."
    return "continue"

Step 5: 组合多个 Critic

Engine 接受一个 Critic 列表。它们按顺序求值，第一个非 continue 的结果生效。这让你可以从最严格到最宽松地堆叠 Critic。

from qitos.kit.critic import SelfReflectionCritic
from qitos.engine.critic_decorator import critic

@critic(name="safety_gate")
def safety_gate(state, decision, results):
    """Hard stop on safety violations -- nothing overrides this."""
    for r in results:
        if isinstance(r, dict) and r.get("safety_violation"):
            return "stop", "safety violation detected"
    return "continue"

@critic(name="format_check")
def format_check(state, decision, results):
    """Retry on formatting issues."""
    if decision.mode == "final":
        answer = decision.final_answer or ""
        if not answer.strip().startswith("{"):
            return "retry", "not JSON", "Output valid JSON only."
    return "continue"

将它们接入 Engine——安全优先，然后是格式，最后是内置的自省 Critic 作为兜底：

from qitos.engine import Engine

engine = Engine(
    agent=my_agent,
    critics=[
        safety_gate,        # 最严格的终止——最先检查
        format_check,       # 较宽松——带指导的重试
        SelfReflectionCritic(max_retries=2),  # 捕获工具错误
    ],
    # ... 其他 Engine 配置
)

按照这个顺序：

如果 safety_gate 返回 "stop"，Engine 立即终止。后续 Critic 不会被调用。
如果 safety_gate 返回 "continue" 但 format_check 返回 "retry"，Engine 带着指令补丁重试。
如果两者都返回 "continue"，SelfReflectionCritic 有机会捕获工具错误。

这种分层方式让每个 Critic 保持小巧专注，而顺序则赋予你对优先级的完全控制。

Hooks 生命周期

了解 Critic 参与的完整 Engine 事件生命周期。

Agent Module

深入了解 Critic 审查的 AgentModule 接口。

Critic 与停止条件

了解 Critic 如何与停止条件和重试预算交互。

​Step 1: CriticResult 契约

​Step 2: 使用内置 Critic

​PassThroughCritic

​SelfReflectionCritic

​ReActSelfReflectionCritic

​函数式等价物

​Step 3: 使用 @critic 装饰器编写自定义 Critic

​Step 4: 带指令补丁和状态补丁的 Critic

​Step 5: 组合多个 Critic