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 如何与停止条件和重试预算交互。

Documentation Index

​Step 1: CriticResult 契约

​Step 2: 使用内置 Critic

​PassThroughCritic

​SelfReflectionCritic

​ReActSelfReflectionCritic

​函数式等价物

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

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

​Step 5: 组合多个 Critic