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_factorydict) # 工具输入参数 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_factorydict) # 修正后的参数 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_idfs{i}, descriptiondesc) 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 step3.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 更适合生产环境。
Agent 犯了错还继续错？反思机制的设计与工程实现
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_factorydict) # 工具输入参数 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_factorydict) # 修正后的参数 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_idfs{i}, descriptiondesc) 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 step3.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 更适合生产环境。
