Agent 犯了错还继续错？反思机制的设计与工程实现

一、Agent 的"一根筋"问题：没有纠错的自主决策是危险的

当前大多数 Agent 系统采用"感知-规划-执行"的直线流程：接收用户输入，调用 LLM 生成行动计划，按计划执行工具调用，返回结果。这个流程在简单任务上表现良好，但在复杂多步任务中，一旦中间步骤出错，后续所有步骤都会基于错误前提继续推进，错误被逐级放大。

典型场景：Agent 被要求"查询 Q3 财报并计算毛利率"。第一步，Agent 调用搜索工具，但关键词不够精确，召回了 Q2 的财报数据；第二步，基于 Q2 数据计算毛利率；第三步，将结果返回给用户。整个过程没有任何环节质疑"这个数据是 Q3 的吗？"。Agent 不是不知道 Q3 和 Q2 的区别，而是缺乏"回头看"的机制。

反思机制（Reflection）的核心思想是：在 Agent 执行链路中插入"自我评估"环节，让 Agent 在关键节点检查当前状态是否偏离目标，如果偏离则修正计划。这不是让模型"更聪明"，而是给 Agent 加一个"刹车"——在错误积累到不可挽回之前停下来纠偏。

二、反思循环的架构设计：从执行-评估到执行-评估-修正

反思机制不是简单地在执行后加一步"检查结果"，而是一个结构化的循环：执行、评估、修正、再执行。理解这个循环的控制流，才能设计出可靠的反思系统。

graph TD
    INPUT[用户输入] --> PLAN[规划: 生成行动计划]
    PLAN --> EXEC[执行: 调用工具/生成中间结果]
    EXEC --> EVAL[评估: 检查执行结果是否满足约束]

    EVAL --> |通过| NEXT[进入下一步]
    EVAL --> |未通过| REFLECT[反思: 分析失败原因]

    REFLECT --> |可修正| REVISE[修正: 调整计划/参数]
    REVISE --> EXEC

    REFLECT --> |不可修正| ESCALATE[升级: 请求人工介入<br/>或告知用户限制]

    NEXT --> |还有步骤| PLAN
    NEXT --> |全部完成| OUTPUT[输出最终结果]

    style EVAL fill:#fff3e0
    style REFLECT fill:#e1f5fe
    style ESCALATE fill:#ffcdd2

评估（Evaluation）：判断当前步骤的执行结果是否满足预期约束。评估标准需要明确——不是模糊的"好不好"，而是可量化的检查项。例如：数据时效性（是否是 Q3 数据）、格式正确性（是否是数值）、逻辑一致性（毛利率是否在合理范围内）。

反思（Reflection）：当评估未通过时，分析失败原因并生成修正方案。反思的关键是"归因"——准确识别错误发生在哪个环节。是搜索关键词不够精确？是工具返回了错误数据？还是计算逻辑有误？不同的归因对应不同的修正策略。

修正（Revision）：基于反思结论调整行动计划。可能是修改工具调用参数、换一个工具、调整中间步骤的顺序，或者放弃当前路径选择替代方案。

升级（Escalation）：当反思次数超过阈值仍无法修正时，不再盲目重试，而是升级处理——请求人工介入或向用户说明限制。这是防止无限循环的安全阀。

三、反思机制的生产级实现

3.1 核心数据结构

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import time

class StepStatus(Enum):
    PENDING = "pending"
    EXECUTING = "executing"
    EVALUATING = "evaluating"
    REFLECTING = "reflecting"
    REVISING = "revising"
    COMPLETED = "completed"
    FAILED = "failed"
    ESCALATED = "escalated"

@dataclass
class ExecutionStep:
    """Agent 执行步骤"""
    step_id: str
    description: str              # 步骤描述
    tool_name: str = ""           # 调用的工具名称
    tool_input: dict = field(default_factory=dict)  # 工具输入参数
    tool_output: str = ""         # 工具输出
    status: StepStatus = StepStatus.PENDING
    reflection_count: int = 0     # 本步骤已反思次数

@dataclass
class EvaluationResult:
    """评估结果"""
    passed: bool
    checks: list[dict]            # 各检查项的结果
    failure_reason: str = ""      # 失败原因（如果未通过）

@dataclass
class ReflectionResult:
    """反思结果"""
    root_cause: str               # 根因分析
    revision_type: str            # 修正类型: reparam / retool / replan
    revised_input: dict = field(default_factory=dict)  # 修正后的参数
    confidence: float = 0.0       # 修正的信心度 [0, 1]

3.2 反思循环引擎

class ReflectiveAgent:
    """带反思机制的 Agent 引擎"""

    def __init__(
        self,
        llm_client,
        tool_registry: dict,
        max_reflections: int = 3,      # 每步最大反思次数
        max_total_reflections: int = 10, # 全局最大反思次数
    ):
        self.llm = llm_client
        self.tools = tool_registry
        self.max_reflections = max_reflections
        self.max_total_reflections = max_total_reflections
        self.total_reflections = 0

    def run(self, user_input: str) -> str:
        """执行带反思的 Agent 循环"""
        # 1. 生成初始计划
        plan = self._generate_plan(user_input)
        steps = [ExecutionStep(step_id=f"s{i}", description=desc)
                 for i, desc in enumerate(plan)]

        # 2. 逐步执行，每步带反思
        for i, step in enumerate(steps):
            step.status = StepStatus.EXECUTING

            # 执行工具调用
            step = self._execute_step(step)

            # 评估执行结果
            step.status = StepStatus.EVALUATING
            eval_result = self._evaluate_step(step, user_input)

            if eval_result.passed:
                step.status = StepStatus.COMPLETED
                continue

            # 评估未通过，进入反思循环
            step = self._reflection_loop(step, eval_result, user_input)

            if step.status == StepStatus.ESCALATED:
                return self._format_escalation(step, user_input)

        # 3. 所有步骤完成，生成最终答案
        return self._generate_final_answer(steps, user_input)

    def _reflection_loop(
        self,
        step: ExecutionStep,
        eval_result: EvaluationResult,
        user_input: str,
    ) -> ExecutionStep:
        """反思循环：评估 → 反思 → 修正 → 重新执行"""
        while step.reflection_count < self.max_reflections:
            if self.total_reflections >= self.max_total_reflections:
                step.status = StepStatus.ESCALATED
                return step

            # 反思：分析失败原因
            step.status = StepStatus.REFLECTING
            reflection = self._reflect(step, eval_result, user_input)
            step.reflection_count += 1
            self.total_reflections += 1

            if reflection.confidence < 0.3:
                # 信心度过低，继续反思无意义，升级处理
                step.status = StepStatus.ESCALATED
                return step

            # 修正：调整参数或工具
            step.status = StepStatus.REVISING
            step = self._apply_revision(step, reflection)

            # 重新执行
            step.status = StepStatus.EXECUTING
            step = self._execute_step(step)

            # 重新评估
            step.status = StepStatus.EVALUATING
            eval_result = self._evaluate_step(step, user_input)

            if eval_result.passed:
                step.status = StepStatus.COMPLETED
                return step

        # 反思次数耗尽，升级处理
        step.status = StepStatus.ESCALATED
        return step

    def _evaluate_step(self, step: ExecutionStep, user_input: str) -> EvaluationResult:
        """评估步骤执行结果"""
        eval_prompt = f"""
请评估以下步骤的执行结果是否满足要求。

用户目标：{user_input}
步骤描述：{step.description}
工具名称：{step.tool_name}
执行结果：{step.tool_output}

请逐项检查：
1. 结果是否与用户目标相关
2. 结果格式是否正确
3. 结果内容是否完整（非空、非错误信息）
4. 结果是否满足时效性要求（如需要最新数据）

以 JSON 格式返回：
{{"passed": true/false, "checks": [{{"name": "检查项", "passed": true/false}}], "failure_reason": "失败原因"}}
"""
        response = self.llm.chat(eval_prompt)
        # 解析 LLM 返回的评估结果
        return self._parse_evaluation(response)

    def _reflect(
        self,
        step: ExecutionStep,
        eval_result: EvaluationResult,
        user_input: str,
    ) -> ReflectionResult:
        """反思：分析失败根因并生成修正方案"""
        reflect_prompt = f"""
步骤执行失败，请分析原因并提出修正方案。

用户目标：{user_input}
步骤描述：{step.description}
工具名称：{step.tool_name}
工具输入：{step.tool_input}
执行结果：{step.tool_output}
失败原因：{eval_result.failure_reason}
已反思次数：{step.reflection_count}

请分析：
1. 根因是什么？（参数不当/工具选错/计划有误）
2. 如何修正？
3. 修正的信心度（0-1）

以 JSON 格式返回：
{{"root_cause": "根因", "revision_type": "reparam/retool/replan", "revised_input": {{}}, "confidence": 0.8}}
"""
        response = self.llm.chat(reflect_prompt)
        return self._parse_reflection(response)

    def _apply_revision(self, step: ExecutionStep, reflection: ReflectionResult) -> ExecutionStep:
        """应用修正方案"""
        if reflection.revision_type == "reparam":
            # 修改工具参数，保持工具不变
            step.tool_input.update(reflection.revised_input)
        elif reflection.revision_type == "retool":
            # 更换工具
            step.tool_name = reflection.revised_input.get("tool_name", step.tool_name)
            step.tool_input = reflection.revised_input.get("tool_input", step.tool_input)
        elif reflection.revision_type == "replan":
            # 重新规划后续步骤（最重的修正）
            new_plan = reflection.revised_input.get("new_steps", [])
            # 生产环境中需要更新全局步骤列表
        return step

    def _execute_step(self, step: ExecutionStep) -> ExecutionStep:
        """执行工具调用"""
        tool = self.tools.get(step.tool_name)
        if tool is None:
            step.tool_output = f"错误：工具 {step.tool_name} 不存在"
            return step

        try:
            step.tool_output = tool.execute(step.tool_input)
        except Exception as e:
            step.tool_output = f"工具执行异常: {str(e)}"
        return step

3.3 领域约束注入：结构化评估

class FinancialAgent(ReflectiveAgent):
    """财务场景 Agent：注入领域约束评估"""

    def _evaluate_step(self, step: ExecutionStep, user_input: str) -> EvaluationResult:
        """在通用评估基础上增加领域约束检查"""
        # 先执行通用评估
        result = super()._evaluate_step(step, user_input)

        if not result.passed:
            return result

        # 领域约束：财务数据时效性检查
        if "财报" in user_input or "季度" in user_input:
            # 检查返回数据是否包含正确的季度标识
            quarter = self._extract_quarter(user_input)
            if quarter and quarter not in step.tool_output:
                result.passed = False
                result.failure_reason = f"数据时效性不匹配：期望 {quarter}，结果中未找到"
                result.checks.append({
                    "name": "季度匹配",
                    "passed": False,
                    "detail": f"期望 {quarter}，实际数据未包含"
                })

        # 领域约束：数值合理性检查
        if "毛利率" in user_input or "利润率" in user_input:
            value = self._extract_percentage(step.tool_output)
            if value is not None and (value < -50 or value > 100):
                result.passed = False
                result.failure_reason = f"数值异常：毛利率 {value}% 超出合理范围 [-50, 100]"
                result.checks.append({
                    "name": "数值合理性",
                    "passed": False,
                    "detail": f"毛利率 {value}% 不在合理范围"
                })

        return result

四、反思机制的代价：Token 消耗、延迟膨胀与过度反思

反思机制在提升 Agent 准确率的同时，引入了显著的资源开销和新的工程风险。

Token 消耗倍增。每次反思需要调用 LLM 做评估和根因分析，至少额外消耗 2 次 LLM 调用。3 次反思意味着 6 次额外调用，Token 成本增加 3-6 倍。对于成本敏感的场景，需要严格控制反思触发条件——不是每一步都反思，只在关键步骤（数据查询、数值计算）上启用反思。

延迟线性增长。反思循环是串行的——评估、反思、修正、重新执行，每轮反思增加 2-5 秒延迟。3 轮反思可能让单步执行时间从 2 秒膨胀到 15 秒。对于实时交互场景，这个延迟不可接受。解决方案是：设定反思超时上限，超时后直接升级处理。

过度反思的陷阱。LLM 生成的评估和反思本身可能有误。评估可能误判正确结果为错误（假阳性），导致不必要的反思循环；反思可能给出错误的根因分析，导致修正方向偏离，越改越错。解决方案是：评估标准尽量用确定性规则（正则匹配、数值范围检查），减少对 LLM 评估的依赖。

反思与执行的耦合。当前实现中，反思逻辑和执行逻辑紧耦合。如果工具接口变更，反思逻辑也需要同步修改。建议将评估标准抽象为声明式规则（如 JSON Schema 校验），与执行逻辑解耦。

适用边界：简单单步任务不需要反思，直接执行即可；多步推理任务在关键步骤启用反思；高风险决策任务（财务、法律、医疗）必须启用反思 + 人工升级；实时对话场景限制反思次数为 1 次，避免延迟过大。

五、总结

反思机制的核心价值是：在 Agent 执行链路中插入"自我纠错"环节，防止错误在多步推理中被逐级放大。评估、反思、修正三步循环构成了 Agent 的"免疫系统"，让 Agent 具备了从错误中恢复的能力。

落地路线建议：第一步，在 Agent 的关键步骤（数据查询、数值计算、外部 API 调用）后插入评估环节，评估标准优先使用确定性规则；第二步，评估未通过时触发反思，调用 LLM 分析根因并生成修正方案；第三步，设定反思次数上限（每步 3 次、全局 10 次），防止无限循环；第四步，反思次数耗尽时升级处理，请求人工介入或向用户说明限制；第五步，监控反思触发率和修正成功率，持续优化评估标准和反思 Prompt。

反思机制不是让 Agent 变得更聪明，而是让 Agent 变得更可靠。一个会纠错的 Agent，比一个从不犯错但也不知纠错的 Agent 更适合生产环境。