> ## 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. # Critic 系统 > 通过 Critic 系统为智能体循环添加运行时质量控制——在决策提交前进行检查、评分、重试或终止。 Engine 每一步的执行都流经一条管线：decide、act、reduce，然后是 **critic**。Critic 位于 `reduce` 与 `check_stop` 之间，为你提供了一个结构化的钩子，可以审查智能体的最新决策及其结果、分配质量评分，并在必要时强制重试或终止执行。 本教程涵盖完整的 Critic API：`CriticResult` 契约、内置 Critic、`@critic` 装饰器、指令补丁与状态补丁，以及组合多个 Critic 的方式。 ## Step 1: CriticResult 契约 每个 Critic 都返回一个 `CriticResult`——一个告知 Engine 下一步操作的数据类。 ```python theme={null} 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 结果： ```python theme={null} result = CriticResult(action="continue", reason="all good", score=1.0) ``` 带指令补丁的重试： ```python theme={null} result = CriticResult( action="retry", reason="output was incomplete", score=0.3, instruction_patch="Make sure to include all required fields in your answer.", ) ``` 强制停止： ```python theme={null} result = CriticResult( action="stop", reason="safety policy violation", score=0.0, ) ``` ## Step 2: 使用内置 Critic QitOS 在 `qitos.kit.critic` 中提供了开箱即用的 Critic。 ### PassThroughCritic 最简单的 Critic——始终以满分继续。 ```python theme={null} from qitos.kit.critic import PassThroughCritic critic = PassThroughCritic() # evaluate() 始终返回 {"action": "continue", "reason": "pass", "score": 1.0} ``` 创建 Engine 时若不指定 critics，这就是默认行为。它可用作空操作占位符，或作为组合链中由后续 Critic 执行真正工作的基础。 ### SelfReflectionCritic 检查工具结果中是否包含错误，并在可配置的重试上限内自动重试。 ```python theme={null} 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： ```python theme={null} from qitos.engine import Engine engine = Engine( agent=my_agent, critics=[SelfReflectionCritic(max_retries=3)], # ... 其他 Engine 配置 ) ``` ### ReActSelfReflectionCritic 专为 ReAct 智能体设计的增强版本。遇到错误时，它会构建一条结构化的反思笔记，描述失败的动作与观察到的错误，并将其追加到状态的 metadata 中，以便 LLM 在下次重试时能够从失败中学习。 ```python theme={null} from qitos.kit.critic import ReActSelfReflectionCritic critic = ReActSelfReflectionCritic(max_retries=2) ``` ### 函数式等价物 每个内置 Critic 都同时提供了装饰器函数版本： ```python theme={null} from qitos.kit.critic import pass_through_critic, self_reflection_critic # 这些是 @critic 装饰的函数——详见 Step 3 ``` ## Step 3: 使用 @critic 装饰器编写自定义 Critic `@critic` 装饰器可将任意普通函数转换为 `Critic` 实例。该函数接收 `(state, decision, results)` 参数，可返回简写形式或完整的 `CriticResult`。 ```python theme={null} from qitos.engine.critic_decorator import critic ``` 快速返回简写： | 返回值 | 含义 | | ---------------------------------------- | --------- | | `"continue"` | 正常继续 | | `("stop", "reason")` | 带原因的终止 | | `("retry", "reason")` | 使用相同提示词重试 | | `("retry", "reason", instruction_patch)` | 追加指令后重试 | | `CriticResult(...)` | 完整结构化结果 | 无参数装饰器： ```python theme={null} @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" ``` 带关键字参数——设置自定义名称和默认评分： ```python theme={null} @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： ```python theme={null} 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 和状态都已经更新完毕。 ```python theme={null} 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` 注入追踪数据： ```python theme={null} @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" ``` 也可以使用元组简写配合指令补丁： ```python theme={null} @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。 ```python theme={null} 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 作为兜底： ```python theme={null} from qitos.engine import Engine engine = Engine( agent=my_agent, critics=[ safety_gate, # 最严格的终止——最先检查 format_check, # 较宽松——带指导的重试 SelfReflectionCritic(max_retries=2), # 捕获工具错误 ], # ... 其他 Engine 配置 ) ``` 按照这个顺序： 1. 如果 `safety_gate` 返回 `"stop"`，Engine 立即终止。后续 Critic 不会被调用。 2. 如果 `safety_gate` 返回 `"continue"` 但 `format_check` 返回 `"retry"`，Engine 带着指令补丁重试。 3. 如果两者都返回 `"continue"`，`SelfReflectionCritic` 有机会捕获工具错误。 这种分层方式让每个 Critic 保持小巧专注，而顺序则赋予你对优先级的完全控制。 import { CardGroup, Card } from "nextra/components"; 了解 Critic 参与的完整 Engine 事件生命周期。 深入了解 Critic 审查的 AgentModule 接口。 了解 Critic 如何与停止条件和重试预算交互。