AI Agent 规划与反思：从 ReAct 到 Reflexion 的推理增强实践

cover

一、Agent 推理的瓶颈：从"行动优先"到"先想后做"

当前主流的 AI Agent 框架普遍采用 ReAct（Reasoning + Acting）模式：模型先输出一段思考，再调用工具执行动作，循环往复直到任务完成。这种模式在简单任务上表现良好，但在需要多步推理和策略调整的复杂场景中暴露出明显短板——Agent 容易陷入"行动惯性"，一旦选择了错误的推理路径，就会沿着错误方向持续执行，无法自我纠正。

某自动化运维团队在生产环境中部署了基于 ReAct 的故障诊断 Agent，在一次数据库主从切换事故中，Agent 错误地将延迟归因于网络抖动，连续执行了 6 次网络重试操作，每次重试都加剧了主从同步延迟，最终导致服务中断时间从预期的 30 秒延长到 12 分钟。根本原因在于：ReAct 模式缺乏"回头看"的机制，Agent 无法在执行过程中评估自身推理的正确性。

反思增强（Reflexion）机制正是为解决这一瓶颈而设计。它在 ReAct 的基础上引入了自我评估和策略修正环节，使 Agent 能够在推理链中发现错误、总结教训并调整后续行为。

二、ReAct 到 Reflexion 的推理架构演进

flowchart TB
    subgraph ReAct["ReAct 模式"]
        R1[观察环境] --> R2[推理思考]
        R2 --> R3[执行动作]
        R3 --> R4{任务完成?}
        R4 -->|否| R1
        R4 -->|是| R5[输出结果]
    end

    subgraph Reflexion["Reflexion 模式"]
        F1[观察环境] --> F2[推理思考]
        F2 --> F3[执行动作]
        F3 --> F4{任务完成?}
        F4 -->|否| F5{自我评估}
        F5 -->|推理正确| F1
        F5 -->|推理偏差| F6[生成反思]
        F6 --> F7[更新策略记忆]
        F7 --> F1
        F4 -->|是| F8[输出结果 + 经验总结]
    end

    ReAct -.->|演进| Reflexion

    style ReAct fill:#f9f,stroke:#333
    style Reflexion fill:#9ff,stroke:#333

Reflexion 的核心差异在于三个新增组件：

自我评估器（Self-Evaluator）：在每次动作执行后，Agent 不仅观察环境反馈，还对自身推理链进行评估。评估维度包括：推理是否与观察一致、动作是否朝着目标推进、是否存在更优路径。

反思生成器（Reflection Generator）：当自我评估发现推理偏差时，生成结构化的反思文本，明确指出错误原因和修正方向。反思文本不是简单的"我错了"，而是包含具体错误定位和改进策略的分析。

策略记忆（Strategy Memory）：将反思经验持久化存储，在后续任务中作为上下文注入，避免重复犯同类错误。这是 Reflexion 区别于单次纠错的关键——它实现了跨任务的策略学习。

三、生产级 Reflexion Agent 的代码实现

以下实现基于 Python，使用结构化的反思机制和策略记忆管理：

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

class EvaluationResult(Enum):
    """自我评估结果枚举"""
    ON_TRACK = "on_track"       # 推理方向正确
    DEVIATION = "deviation"     # 轻微偏差，可继续
    WRONG_PATH = "wrong_path"  # 严重偏差，需要反思

@dataclass
class Reflection:
    """结构化反思记录"""
    step: int                           # 出错步骤
    observation: str                    # 实际观察
    expected: str                       # 预期结果
    deviation_type: str                 # 偏差类型：premise_error / action_error / goal_drift
    root_cause: str                     # 根因分析
    correction: str                     # 修正策略
    confidence: float                   # 修正置信度

@dataclass
class StrategyMemory:
    """策略记忆：跨任务的反思经验库"""
    reflections: list[Reflection] = field(default_factory=list)

    def add_reflection(self, reflection: Reflection):
        self.reflections.append(reflection)

    def get_relevant_reflections(self, context: str, top_k: int = 3) -> list[Reflection]:
        """根据当前上下文检索相关反思经验"""
        # 基于偏差类型和根因关键词的简单匹配策略
        # 生产环境中应替换为向量检索
        scored = []
        for r in self.reflections:
            score = 0.0
            if r.deviation_type in context:
                score += 0.4
            # 根因关键词重叠度
            cause_words = set(r.root_cause.split())
            ctx_words = set(context.split())
            overlap = len(cause_words & ctx_words)
            score += min(overlap / max(len(cause_words), 1), 1.0) * 0.6
            scored.append((score, r))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [r for _, r in scored[:top_k]]

    def format_for_prompt(self, context: str) -> str:
        """将策略记忆格式化为 Prompt 注入文本"""
        relevant = self.get_relevant_reflections(context)
        if not relevant:
            return ""
        lines = ["## 历史反思经验（请避免同类错误）"]
        for i, r in enumerate(relevant, 1):
            lines.append(
                f"{i}. 步骤{r.step}：{r.deviation_type} - "
                f"根因：{r.root_cause}，修正：{r.correction}"
            )
        return "\n".join(lines)


class ReflexionAgent:
    """带反思增强的 Agent 实现"""

    def __init__(self, llm_client, tools: dict, max_retries: int = 3):
        self.llm = llm_client
        self.tools = tools
        self.max_retries = max_retries
        self.strategy_memory = StrategyMemory()
        self.trajectory: list[dict] = []

    def _evaluate_step(self, step_result: dict, goal: str) -> EvaluationResult:
        """自我评估：判断当前推理是否偏离目标"""
        eval_prompt = f"""
        目标：{goal}
        当前步骤：{step_result['thought']}
        执行动作：{step_result['action']}
        观察结果：{step_result['observation']}

        请评估当前推理方向：
        - on_track：推理正确，朝着目标推进
        - deviation：轻微偏差，需要微调
        - wrong_path：严重偏差，需要反思并调整策略

        仅输出评估结果和一句话理由，格式：评估结果|理由
        """
        response = self.llm.generate(eval_prompt)
        result_str, reason = response.strip().split("|", 1)
        try:
            return EvaluationResult(result_str.strip())
        except ValueError:
            return EvaluationResult.DEVIATION

    def _generate_reflection(self, step_result: dict, goal: str) -> Reflection:
        """生成结构化反思"""
        reflection_prompt = f"""
        目标：{goal}
        失败步骤：{step_result['thought']}
        执行动作：{step_result['action']}
        实际观察：{step_result['observation']}
        预期结果：{step_result.get('expected', '未明确')}

        请生成结构化反思，分析偏差原因并给出修正策略。
        输出 JSON 格式：
        {{
            "deviation_type": "premise_error/action_error/goal_drift",
            "root_cause": "根因分析",
            "correction": "修正策略",
            "confidence": 0.0-1.0
        }}
        """
        response = self.llm.generate(reflection_prompt)
        data = json.loads(response)
        return Reflection(
            step=len(self.trajectory),
            observation=step_result['observation'],
            expected=step_result.get('expected', ''),
            deviation_type=data['deviation_type'],
            root_cause=data['root_cause'],
            correction=data['correction'],
            confidence=data['confidence']
        )

    def run(self, task: str) -> str:
        """执行任务，带反思增强的推理循环"""
        for attempt in range(self.max_retries):
            # 注入策略记忆到上下文
            memory_context = self.strategy_memory.format_for_prompt(task)
            result = self._execute_with_reflection(task, memory_context)

            if result['success']:
                return result['answer']

            # 任务失败，触发反思并更新策略记忆
            reflection = self._generate_reflection(
                result['last_step'], task
            )
            self.strategy_memory.add_reflection(reflection)

        return f"任务在 {self.max_retries} 次尝试后仍未完成"

    def _execute_with_reflection(self, task: str, memory: str) -> dict:
        """带反思检查的执行循环"""
        self.trajectory = []
        for step in range(20):  # 单次尝试最大步数
            # 构建包含策略记忆的 Prompt
            thought, action = self._think_and_act(task, memory)
            observation = self._execute_action(action)

            step_result = {
                'thought': thought, 'action': action,
                'observation': observation
            }
            self.trajectory.append(step_result)

            # 每 3 步进行一次自我评估
            if step > 0 and step % 3 == 0:
                evaluation = self._evaluate_step(step_result, task)
                if evaluation == EvaluationResult.WRONG_PATH:
                    # 严重偏差：生成本次反思并提前终止本轮
                    reflection = self._generate_reflection(step_result, task)
                    self.strategy_memory.add_reflection(reflection)
                    return {'success': False, 'last_step': step_result}

        return {'success': True, 'answer': self._extract_answer()}

关键设计决策说明：

  • 评估频率控制：不是每步都评估，而是每 3 步评估一次。过于频繁的评估会显著增加 Token 消耗和延迟，而间隔太长则无法及时纠偏。3 步是一个在成本和效果间取得平衡的经验值。

  • 反思持久化：StrategyMemory 将反思经验跨任务保存，新任务启动时自动检索相关经验注入上下文。这比简单的重试机制更高效——Agent 不需要在同一类错误上反复试错。

  • 偏差分类：将推理偏差分为前提错误（premise_error）、动作错误（action_error）和目标漂移（goal_drift）三类，不同类型对应不同的修正策略。

四、Reflexion 的 Trade-offs 与适用边界

Token 成本显著增加。自我评估和反思生成环节需要额外的 LLM 调用，单次任务的 Token 消耗约为纯 ReAct 模式的 1.5-2.5 倍。在高频低价值任务（如简单信息查询）中，这种成本增加是不合理的。

反思质量依赖评估器能力。如果自我评估器本身判断失误——将正确推理误判为偏差，或忽略真正的错误——反思反而会引导 Agent 走向更错误的方向。在模型能力较弱的场景下，建议用规则引擎替代 LLM 评估器。

策略记忆的时效性问题。历史反思经验可能过时——环境变化后，曾经的"错误"策略可能变为正确策略。需要在策略记忆中加入时间衰减机制，或定期清理过时的反思记录。

适用场景：多步推理、策略探索、需要试错纠偏的复杂任务。禁用场景：单步查询、确定性操作、对延迟敏感的实时交互。

五、总结

Reflexion 机制通过自我评估、反思生成和策略记忆三个核心组件，将 Agent 从"行动优先"的 ReAct 模式升级为"先想后做、错了能改"的推理增强模式。其核心价值在于跨任务的策略学习能力，使 Agent 在面对同类问题时能够避免重复犯错。但 Reflexion 并非银弹——Token 成本增加、评估器可靠性、记忆时效性是需要权衡的关键因素。在实际落地中，建议根据任务复杂度动态选择推理模式：简单任务用 ReAct，复杂任务启用 Reflexion，并通过评估器准确率和策略记忆命中率持续优化反思质量。

# 后端 # golang
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区