当前位置：首页 > news >正文

Agentic Design Patterns-模式4：反思（Reflection）的代码实现

news 2026/5/27 11:12:42

1. 概述

2. 价值

3. 关键技术

4. 代码实现

5. 核心组件说明

5.1 ReflectionState（状态管理类）

5.2 ProducerAgent（生产者智能体）

5.3 CriticAgent（批评者智能体）

5.4 ReflectionOrchestrator（编排器）

6. 使用场景

7. 优势与局限

优势：

局限：

8. 最佳实践

9. 拓展方向

10. 总结

1. 概述

反思模式指的是智能体对自身的工作、输出或内部状态进行评估，并利用评估结果来提升性能或优化响应。它是一种自我纠错或自我改进机制，使智能体能够根据反馈、内部批判或与目标标准的对比，持续优化输出或调整策略。反思有时也可以由专门负责分析初始智能体输出的独立智能体来实现。

与简单的链式传递或路径选择不同，反思引入了反馈循环。智能体不只生成输出，还会审视该输出（或生成过程），识别潜在问题或改进空间，并据此生成优化版本或调整后续行为。一种高效的反思实现方式是将流程分为两个逻辑角色，生产者（Producer）和批评者（Critic），其典型流程包括执行——评估/批判——反思/优化——迭代。

图1：反思设计模式（自我反思）

图2：反思设计模式（批判者Agent）

2. 价值

反思为智能体系统增加了元认知层，使其能从自身输出和过程学习，带来更智能、可靠、高质量的结果。反思模式的核心优势是能迭代自我纠错和优化输出，显著提升质量、准确性和复杂指令的遵循度。

3. 关键技术

反思模式为智能体工作流提供了关键的自我纠错机制，实现了超越单次执行的迭代优化。其核心是建立一个循环：系统生成输出，按特定标准评估，再利用评估结果生成优化版本。评估可由智能体自评，也可由独立批评者智能体完成。

完整的多步反思过程需要健壮的状态管理架构，其核心原理可通过单次生成‑批判优化循环高效演示，作为控制结构，反思可与其他基础模式结合，构建更健壮、功能更复杂的智能体系统。

4. 代码实现

以下是一个完整的反思（Reflection）模式Python程序示例，包含生产者（Producer）和批评者（Critic）两个逻辑角色，实现了执行-评估-优化-迭代的循环机制：

import json from typing import Dict, Any, List, Optional from dataclasses import dataclass, asdict # 导入模型客户端 from model_factory import get_model_client client = get_model_client() MODEL = "gpt-4o" @dataclass class ReflectionState: """反思过程的状态管理""" iteration: int = 1 current_output: str = "" critique: str = "" is_satisfactory: bool = False history: List[Dict[str, str]] = None def __post_init__(self): if self.history is None: self.history = [] def add_to_history(self, output: str, critique: str = ""): """将当前步骤加入历史记录""" self.history.append({ "iteration": self.iteration, "output": output, "critique": critique }) def to_dict(self) -> Dict[str, Any]: return asdict(self) class ProducerAgent: """生产者智能体 - 负责生成初始输出""" @staticmethod def generate_response(task_description: str, previous_critique: str = "") -> str: """生成任务响应""" system_prompt = """你是一个专业的内容生产者。根据用户的任务描述生成高质量的输出。 如果提供了之前的批评意见，请基于这些意见改进你的输出。""" if previous_critique: user_prompt = f"""任务：{task_description} 之前的批评意见：{previous_critique} 请根据上述批评意见，生成改进后的响应。确保解决所有指出的问题。""" else: user_prompt = f"""任务：{task_description} 请为上述任务生成高质量的响应。""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] try: response = client.chat.completions.create( model=MODEL, messages=messages, stream=False ) return response.choices[0].message.content except Exception as e: print(f"生产者生成失败: {e}") return "生成失败，请重试。" class CriticAgent: """批评者智能体 - 负责评估和提出改进意见""" def __init__(self, evaluation_criteria: Optional[List[str]] = None): self.evaluation_criteria = evaluation_criteria or [ "准确性和事实正确性", "逻辑连贯性和结构", "完整性和深度", "清晰度和可读性", "与任务要求的符合度" ] def critique_output(self, task_description: str, output: str) -> Dict[str, Any]: """批评输出并提供改进建议""" criteria_str = "\n".join([f"{i+1}. {criterion}" for i, criterion in enumerate(self.evaluation_criteria)]) system_prompt = """你是一个严格的批评者。请仔细评估提供的输出，并根据评估标准提出具体的改进建议。 你的批评应该具体、建设性，并提供明确的改进方向。""" user_prompt = f"""任务描述：{task_description} 待评估的输出： {output} 评估标准： {criteria_str} 请按以下格式提供反馈： 1. 总体评分（1-10分） 2. 主要优点 3. 存在的问题和不足 4. 具体的改进建议 5. 是否需要进一步优化（是/否）""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] try: response = client.chat.completions.create( model=MODEL, messages=messages, stream=False ) critique_text = response.choices[0].message.content # 解析是否需要进一步优化 needs_improvement = self._check_needs_improvement(critique_text) return { "critique": critique_text, "needs_improvement": needs_improvement, "raw_output": output } except Exception as e: print(f"批评者评估失败: {e}") return { "critique": "评估失败", "needs_improvement": False, "raw_output": output } def _check_needs_improvement(self, critique_text: str) -> bool: """检查是否需要进一步改进""" if "是" in critique_text and "是否需要进一步优化" in critique_text: return True return False class ReflectionOrchestrator: """反思流程编排器 - 协调生产者和批评者""" def __init__(self, max_iterations: int = 3): self.producer = ProducerAgent() self.critic = CriticAgent() self.max_iterations = max_iterations def execute(self, task_description: str) -> Dict[str, Any]: """执行反思流程""" state = ReflectionState() for iteration in range(1, self.max_iterations + 1): state.iteration = iteration # 生产者生成输出 output = self.producer.generate_response( task_description, state.critique if iteration > 1 else "" ) state.current_output = output # 批评者评估输出 critique_result = self.critic.critique_output(task_description, output) state.critique = critique_result["critique"] # 记录历史 state.add_to_history(output, state.critique) # 检查是否满足要求 if not critique_result["needs_improvement"]: state.is_satisfactory = True break return state.to_dict() # 使用示例 if __name__ == "__main__": orchestrator = ReflectionOrchestrator(max_iterations=3) result = orchestrator.execute("写一篇关于人工智能伦理的文章") print(f"最终输出: {result['current_output']}") print(f"迭代次数: {result['iteration']}") print(f"是否满意: {result['is_satisfactory']}")