AI Agent Reflection 模式：让智能体学会自我反思
引言
在构建 AI Agent 的过程中，我们常常会遇到这样的问题：模型生成的初始回答可能存在事实错误、逻辑漏洞或不符合预期的格式。传统的单次调用模式（Single-shot）难以解决这些问题。Reflection 模式应运而生——它通过让 Agent 对自己的输出进行自我评估和迭代改进，显著提升输出质量。
Reflection 模式的核心思想借鉴了人类的反思过程：当我们完成一项任务后，会回顾自己的表现，发现不足，然后加以改进。将这种能力赋予 AI Agent，使其成为一个能够持续学习和优化的智能系统。
一、Reflection 模式的核心概念
1.1 什么是 Reflection 模式？
Reflection 模式是一种元认知（Meta-cognitive）架构，它允许 AI Agent：
- 生成初始响应（Generation）
- 评估响应质量（Evaluation）
- 基于反馈进行改进（Refinement）
- 迭代直到满足条件（Iteration）
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Generation │────▶│ Evaluation │────▶│ Refinement │
│ (生成答案) │ │ (评估质量) │ │ (改进优化) │
└─────────────┘ └──────┬──────┘ └──────┬──────┘
│ │
└──────────┬─────────┘
│
┌────────────▼────────────┐
│ Pass? → Output │
│ Fail? → Loop back │
└─────────────────────────┘1.2 Reflection 与其他模式的对比
| 特性 | Single-shot | ReAct | Plan-and-Execute | Reflection |
|---|---|---|---|---|
| 迭代次数 | 1次 | 多轮（工具调用） | 多轮（任务分解） | 多轮（自我改进） |
| 核心机制 | 直接生成 | Thought-Action-Observation | Plan → Execute | Generate → Evaluate → Refine |
| 自我修正 | ❌ | 有限 | 有限 | ✅ 强 |
| 适用场景 | 简单问答 | 工具使用 | 复杂任务规划 | 高质量内容生成 |
| 延迟 | 低 | 中 | 高 | 中-高 |
二、Self-Reflection：单智能体自我反思
2.1 基础架构
Self-Reflection 是最简单的 Reflection 形式，单个 Agent 扮演两个角色：生成者（Generator）和评估者（Critic）。
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import operator
# 定义状态
class SelfReflectionState(TypedDict):
"""
自我反思 Agent 的状态定义
"""
original_query: str # 原始用户查询
current_draft: str # 当前生成的草稿
reflection_notes: List[str] # 反思笔记
revision_count: int # 修订次数
max_revisions: int # 最大修订次数
is_satisfied: bool # 是否满足质量标准
# 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
# ============ 节点 1: 初始生成 ============
def generate_initial_response(state: SelfReflectionState) -> SelfReflectionState:
"""生成初始响应"""
query = state["original_query"]
prompt = f"""请回答以下问题。提供详细且准确的回答：
问题：{query}
请直接给出你的最佳回答。"""
response = llm.invoke([HumanMessage(content=prompt)])
return {
**state,
"current_draft": response.content,
"revision_count": 0
}
# ============ 节点 2: 自我反思 ============
def reflect_on_response(state: SelfReflectionState) -> SelfReflectionState:
"""对当前草稿进行自我反思和评估"""
query = state["original_query"]
draft = state["current_draft"]
reflection_prompt = f"""你是一位严格的评审专家。请仔细评估以下回答的质量。
原始问题：{query}
当前回答：
{draft}
请从以下维度进行评估：
1. **准确性**：信息是否准确？有无事实错误？
2. **完整性**：是否全面回答了问题？有无遗漏？
3. **逻辑性**：论证是否清晰？逻辑是否连贯？
4. **实用性**：回答是否实用？能否真正帮助用户？
请提供具体的改进建议。如果回答已经很好，请明确说明"无需改进"。
反思笔记："""
reflection = llm.invoke([HumanMessage(content=reflection_prompt)])
reflection_content = reflection.content
# 判断是否满足条件
is_satisfied = "无需改进" in reflection_content or "无需修改" in reflection_content
# 提取反思要点
notes = [line.strip() for line in reflection_content.split('\n')
if line.strip() and not line.strip().startswith('反思笔记')]
return {
**state,
"reflection_notes": notes,
"is_satisfied": is_satisfied
}
# ============ 节点 3: 改进优化 ============
def refine_response(state: SelfReflectionState) -> SelfReflectionState:
"""基于反思结果改进回答"""
query = state["original_query"]
draft = state["current_draft"]
notes = state["reflection_notes"]
notes_text = '\n'.join([f"- {note}" for note in notes])
refinement_prompt = f"""请基于以下反馈改进你的回答。
原始问题：{query}
当前回答：
{draft}
改进建议：
{notes_text}
请提供一个改进后的回答，确保解决上述所有问题。保持回答的完整性和准确性。"""
refined = llm.invoke([HumanMessage(content=refinement_prompt)])
return {
**state,
"current_draft": refined.content,
"revision_count": state["revision_count"] + 1
}
# ============ 条件判断 ============
def should_continue(state: SelfReflectionState) -> str:
"""决定是否继续迭代"""
if state["is_satisfied"]:
return "end"
if state["revision_count"] >= state["max_revisions"]:
return "end"
return "continue"2.2 构建 LangGraph 工作流
# 创建工作流
workflow = StateGraph(SelfReflectionState)
# 添加节点
workflow.add_node("generate", generate_initial_response)
workflow.add_node("reflect", reflect_on_response)
workflow.add_node("refine", refine_response)
# 设置入口
workflow.set_entry_point("generate")
# 添加边
workflow.add_edge("generate", "reflect")
workflow.add_conditional_edges(
"reflect",
should_continue,
{
"continue": "refine",
"end": END
}
)
workflow.add_edge("refine", "reflect")
# 编译
self_reflection_agent = workflow.compile()
# ============ 使用示例 ============
def run_self_reflection_agent(query: str, max_revisions: int = 3):
"""运行自我反思 Agent"""
initial_state = {
"original_query": query,
"current_draft": "",
"reflection_notes": [],
"revision_count": 0,
"max_revisions": max_revisions,
"is_satisfied": False
}
result = self_reflection_agent.invoke(initial_state)
print(f"=== 最终回答 (经过 {result['revision_count']} 轮修订) ===")
print(result["current_draft"])
print("\n=== 反思过程 ===")
for i, note in enumerate(result["reflection_notes"], 1):
print(f"{i}. {note}")
return result
# 运行示例
if __name__ == "__main__":
query = "解释量子计算的基本原理，以及它与经典计算的主要区别"
run_self_reflection_agent(query, max_revisions=2)2.3 进阶：结构化反思
使用 Pydantic 模型让反思过程更加结构化：
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
class ReflectionResult(BaseModel):
"""结构化反思结果"""
accuracy_score: int = Field(1, ge=1, le=10, description="准确性评分 1-10")
completeness_score: int = Field(1, ge=1, le=10, description="完整性评分 1-10")
clarity_score: int = Field(1, ge=1, le=10, description="清晰度评分 1-10")
issues: List[str] = Field(default_factory=list, description="发现的问题列表")
suggestions: List[str] = Field(default_factory=list, description="改进建议")
needs_revision: bool = Field(False, description="是否需要修订")
def structured_reflect(state: SelfReflectionState) -> SelfReflectionState:
"""使用结构化输出的反思"""
parser = PydanticOutputParser(pydantic_object=ReflectionResult)
query = state["original_query"]
draft = state["current_draft"]
prompt = f"""评估以下回答的质量。
原始问题：{query}
当前回答：{draft}
{parser.get_format_instructions()}
"""
response = llm.invoke([HumanMessage(content=prompt)])
reflection = parser.parse(response.content)
# 计算综合得分
avg_score = (reflection.accuracy_score +
reflection.completeness_score +
reflection.clarity_score) / 3
is_satisfied = avg_score >= 8.0 and not reflection.needs_revision
notes = reflection.issues + reflection.suggestions
return {
**state,
"reflection_notes": notes,
"is_satisfied": is_satisfied
}三、Multi-Reflection：多智能体协作反思
3.1 架构设计
Multi-Reflection 引入多个专门的评估 Agent，每个负责不同的评估维度：
┌─────────────────────────────────────────────────────────────┐
│ User Query │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Generator Agent │
│ (生成初始回答) │
└───────────────────────┬─────────────────────────────────────┘
│
┌───────────────┼───────────────┐
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Fact Checker │ │ Logic Review │ │ Style Editor │
│ (事实核查) │ │ (逻辑审查) │ │ (风格编辑) │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
└────────────────┼────────────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ Revision Aggregator │
│ (整合反馈并改进) │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────┐
│ Final Output │
└─────────────────┘3.2 完整实现
from typing import TypedDict, List, Dict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 初始化多个 LLM 实例（可以使用不同模型或不同温度）
generator_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
critic_llm = ChatOpenAI(model="gpt-4o", temperature=0.3) # 评估用较低温度
class MultiReflectionState(TypedDict):
"""多智能体反思状态"""
query: str
current_draft: str
revision_count: int
max_revisions: int
# 各评审员的反馈
fact_check_feedback: Dict
logic_review_feedback: Dict
style_feedback: Dict
# 聚合结果
all_issues: List[str]
should_continue: bool
# ============ 生成节点 ============
def generate_response(state: MultiReflectionState) -> MultiReflectionState:
"""生成初始回答"""
query = state["query"]
system_prompt = """你是一位专业的内容创作者。请提供高质量、准确的回答。"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"请回答：{query}")
]
response = generator_llm.invoke(messages)
return {
**state,
"current_draft": response.content,
"revision_count": 0
}
# ============ 事实核查节点 ============
def fact_checker(state: MultiReflectionState) -> MultiReflectionState:
"""事实核查 Agent"""
query = state["query"]
draft = state["current_draft"]
system_prompt = """你是一位严谨的事实核查专家。你的任务是：
1. 识别回答中的事实性陈述
2. 标记可能存在的错误或不确定信息
3. 提供具体的事实修正建议
4. 评分：1-10分
输出格式：
- 发现的问题：[具体问题]
- 建议修正：[如何修正]
- 置信度评分：[1-10]"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"问题：{query}\n\n回答内容：\n{draft}")
]
response = critic_llm.invoke(messages)
content = response.content
# 解析评分
score = 10
for line in content.split('\n'):
if '评分' in line or 'score' in line.lower():
try:
score = int([s for s in line if s.isdigit()][0])
except:
pass
return {
**state,
"fact_check_feedback": {
"content": content,
"score": score,
"passed": score >= 8
}
}
# ============ 逻辑审查节点 ============
def logic_reviewer(state: MultiReflectionState) -> MultiReflectionState:
"""逻辑审查 Agent"""
query = state["query"]
draft = state["current_draft"]
system_prompt = """你是一位逻辑分析专家。你的任务是：
1. 检查论证的逻辑连贯性
2. 识别逻辑漏洞或跳跃
3. 评估推理过程的合理性
4. 评分：1-10分
输出格式：
- 逻辑问题：[具体问题]
- 改进建议：[如何改进]
- 逻辑评分：[1-10]"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"问题：{query}\n\n回答内容：\n{draft}")
]
response = critic_llm.invoke(messages)
content = response.content
score = 10
for line in content.split('\n'):
if '评分' in line or 'score' in line.lower():
try:
score = int([s for s in line if s.isdigit()][0])
except:
pass
return {
**state,
"logic_review_feedback": {
"content": content,
"score": score,
"passed": score >= 8
}
}
# ============ 风格编辑节点 ============
def style_editor(state: MultiReflectionState) -> MultiReflectionState:
"""风格编辑 Agent"""
query = state["query"]
draft = state["current_draft"]
system_prompt = """你是一位专业的文字编辑。你的任务是：
1. 评估语言表达的清晰度
2. 检查格式和结构
3. 确保适合目标读者
4. 评分：1-10分
输出格式：
- 风格问题：[具体问题]
- 润色建议：[如何改进]
- 风格评分：[1-10]"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=f"问题：{query}\n\n回答内容：\n{draft}")
]
response = critic_llm.invoke(messages)
content = response.content
score = 10
for line in content.split('\n'):
if '评分' in line or 'score' in line.lower():
try:
score = int([s for s in line if s.isdigit()][0])
except:
pass
return {
**state,
"style_feedback": {
"content": content,
"score": score,
"passed": score >= 8
}
}
# ============ 反馈聚合节点 ============
def aggregate_feedback(state: MultiReflectionState) -> MultiReflectionState:
"""聚合所有反馈并决定下一步"""
fact = state["fact_check_feedback"]
logic = state["logic_review_feedback"]
style = state["style_feedback"]
all_issues = []
if not fact.get("passed", True):
all_issues.append(f"[事实] {fact['content'][:200]}...")
if not logic.get("passed", True):
all_issues.append(f"[逻辑] {logic['content'][:200]}...")
if not style.get("passed", True):
all_issues.append(f"[风格] {style['content'][:200]}...")
# 判断是否继续
all_passed = (fact.get("passed", True) and
logic.get("passed", True) and
style.get("passed", True))
should_continue = not all_passed and state["revision_count"] < state["max_revisions"]
return {
**state,
"all_issues": all_issues,
"should_continue": should_continue
}
# ============ 修订节点 ============
def revise_response(state: MultiReflectionState) -> MultiReflectionState:
"""基于所有反馈修订回答"""
query = state["query"]
draft = state["current_draft"]
issues = state["all_issues"]
issues_text = '\n'.join([f"{i+1}. {issue}" for i, issue in enumerate(issues)])
system_prompt = """你是一位专业的内容修订专家。请基于反馈改进回答，同时保持回答的完整性和准确性。"""
prompt = f"""请基于以下反馈改进你的回答：
原始问题：{query}
当前回答：
{draft}
需要改进的问题：
{issues_text}
请提供改进后的完整回答。"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=prompt)
]
response = generator_llm.invoke(messages)
return {
**state,
"current_draft": response.content,
"revision_count": state["revision_count"] + 1
}
# ============ 条件判断 ============
def check_continue(state: MultiReflectionState) -> str:
"""检查是否继续迭代"""
if state["should_continue"]:
return "revise"
return "end"3.3 构建并行评审工作流
# 创建并行评审的工作流
workflow = StateGraph(MultiReflectionState)
# 添加节点
workflow.add_node("generate", generate_response)
workflow.add_node("fact_check", fact_checker)
workflow.add_node("logic_review", logic_reviewer)
workflow.add_node("style_edit", style_editor)
workflow.add_node("aggregate", aggregate_feedback)
workflow.add_node("revise", revise_response)
# 设置入口
workflow.set_entry_point("generate")
# 生成后并行执行三个评审
workflow.add_edge("generate", "fact_check")
workflow.add_edge("generate", "logic_review")
workflow.add_edge("generate", "style_edit")
# 三个评审都完成后聚合
workflow.add_edge("fact_check", "aggregate")
workflow.add_edge("logic_review", "aggregate")
workflow.add_edge("style_edit", "aggregate")
# 聚合后决定路径
workflow.add_conditional_edges(
"aggregate",
check_continue,
{
"revise": "revise",
"end": END
}
)
# 修订后重新评审
workflow.add_edge("revise", "fact_check")
workflow.add_edge("revise", "logic_review")
workflow.add_edge("revise", "style_edit")
# 编译
multi_reflection_agent = workflow.compile()
# ============ 使用示例 ============
def run_multi_reflection(query: str, max_revisions: int = 2):
"""运行多智能体反思"""
initial_state = {
"query": query,
"current_draft": "",
"revision_count": 0,
"max_revisions": max_revisions,
"fact_check_feedback": {},
"logic_review_feedback": {},
"style_feedback": {},
"all_issues": [],
"should_continue": True
}
result = multi_reflection_agent.invoke(initial_state)
print(f"=== 最终回答 (经过 {result['revision_count']} 轮修订) ===")
print(result["current_draft"])
print("\n=== 评审反馈 ===")
print(f"事实核查：{result['fact_check_feedback'].get('score', 'N/A')}/10")
print(f"逻辑审查：{result['logic_review_feedback'].get('score', 'N/A')}/10")
print(f"风格编辑：{result['style_feedback'].get('score', 'N/A')}/10")
return result
if __name__ == "__main__":
query = "撰写一篇关于人工智能伦理的短文，讨论隐私、偏见和透明度问题"
run_multi_reflection(query)四、最佳实践
4.1 反思提示词设计
生成提示
GENERATION_PROMPT = """你是一位{role}。请回答用户的问题。
要求：
1. 提供准确、全面的信息
2. 使用清晰的结构和格式
3. 适当使用示例和解释
用户问题：{query}"""反思提示
REFLECTION_PROMPT = """你是一位严格的评审专家。请评估以下回答。
评估维度：
1. 准确性（Accuracy）：事实是否正确
2. 完整性（Completeness）：是否回答完整
3. 清晰度（Clarity）：表达是否清晰
4. 实用性（Utility）：是否有实际帮助
评分标准：
- 8-10分：优秀，几乎无需改进
- 5-7分：良好，有改进空间
- 1-4分：需要重大改进
请提供：
1. 各维度评分
2. 具体问题
3. 具体改进建议
回答内容：
{draft}"""修订提示
REVISION_PROMPT = """请基于以下反馈改进你的回答。
原始回答：
{draft}
改进建议：
{feedback}
要求：
1. 解决所有指出的问题
2. 保持回答的完整性
3. 不要引入新的错误
4. 保持原有的语气和风格"""4.2 迭代控制策略
class IterationController:
"""迭代控制器"""
def __init__(self, max_iterations: int = 3, quality_threshold: float = 0.8):
self.max_iterations = max_iterations
self.quality_threshold = quality_threshold
self.history = []
def should_continue(self, current_score: float, iteration: int) -> bool:
"""决定是否继续迭代"""
# 达到最大迭代次数
if iteration >= self.max_iterations:
return False
# 质量已达标
if current_score >= self.quality_threshold:
return False
# 检查是否收敛（连续两次得分提升小于阈值）
if len(self.history) >= 2:
recent_improvement = self.history[-1] - self.history[-2]
if recent_improvement < 0.05: # 提升小于5%
return False
self.history.append(current_score)
return True
def adaptive_max_iterations(self, complexity: str) -> int:
"""根据任务复杂度自适应调整最大迭代次数"""
complexity_map = {
"simple": 1,
"medium": 3,
"complex": 5,
"research": 7
}
return complexity_map.get(complexity, 3)4.3 成本与延迟优化
from functools import lru_cache
class OptimizedReflection:
"""优化的 Reflection 实现"""
def __init__(self):
# 轻量级模型用于简单评估
self.light_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 强模型用于生成和复杂评估
self.strong_llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
def quick_check(self, draft: str) -> bool:
"""快速检查是否需要详细反思"""
prompt = f"""这个回答看起来质量如何？简单回答：好/需要改进
回答：{draft[:500]}..."""
response = self.light_llm.invoke([HumanMessage(content=prompt)])
return "好" in response.content
def tiered_reflection(self, query: str, draft: str) -> Dict:
"""分层反思策略"""
# 第一层：快速检查
if self.quick_check(draft):
return {"needs_revision": False, "score": 0.85}
# 第二层：详细反思（使用强模型）
return self.detailed_reflect(query, draft)
@lru_cache(maxsize=100)
def cached_reflect(self, query_hash: str, draft_hash: str) -> Dict:
"""带缓存的反思（适用于重复查询）"""
# 实现缓存逻辑
pass五、应用场景
5.1 代码生成与审查
class CodeReflectionAgent:
"""代码生成与反思 Agent"""
def __init__(self):
self.generator = ChatOpenAI(model="gpt-4o", temperature=0.7)
self.code_reviewer = ChatOpenAI(model="gpt-4o", temperature=0.2)
def generate_code(self, requirements: str, language: str = "python") -> Dict:
"""生成并优化代码"""
# 生成初始代码
gen_prompt = f"""根据以下需求编写{language}代码：
需求：
{requirements}
要求：
1. 代码应清晰、高效
2. 包含必要的注释
3. 处理边界情况
4. 遵循最佳实践
请只输出代码，不要解释。"""
code_response = self.generator.invoke([HumanMessage(content=gen_prompt)])
initial_code = code_response.content
# 代码审查
review_prompt = f"""请审查以下代码：
```{language}
{initial_code}请检查：
- 语法正确性
- 逻辑正确性
- 性能问题
- 安全问题
- 可读性
如有问题，请指出并提供修正后的代码。"""
review_response = self.code_reviewer.invoke([HumanMessage(content=review_prompt)])
返回代码审查结果（简化展示）
return result
### 5.2 内容创作与润色
```python
class ContentReflectionAgent:
"""内容创作反思 Agent"""
def create_article(self, topic: str, style: str = "professional",
word_count: int = 1000) -> str:
"""创作并优化文章"""
# 生成初稿
draft = self.generate_draft(topic, style, word_count)
# 多维度反思
evaluations = {
"structure": self.evaluate_structure(draft),
"tone": self.evaluate_tone(draft, style),
"engagement": self.evaluate_engagement(draft),
"seo": self.evaluate_seo(draft, topic)
}
# 整合改进
final_draft = self.revise_article(draft, evaluations)
return final_draft
def evaluate_structure(self, draft: str) -> Dict:
"""评估文章结构"""
prompt = f"""评估以下文章的结构：
{draft}
请分析：
1. 开头是否吸引人
2. 段落组织是否清晰
3. 过渡是否自然
4. 结尾是否有力
给出结构评分（1-10）和具体改进建议。"""
response = self.code_reviewer.invoke([HumanMessage(content=prompt)])
return self.parse_evaluation(response.content)5.3 数据分析报告生成
class DataAnalysisReflectionAgent:
"""数据分析报告反思 Agent"""
def generate_report(self, data_summary: str, analysis_goal: str) -> Dict:
"""生成并优化数据分析报告"""
# 生成报告
report = self.create_initial_report(data_summary, analysis_goal)
# 专业评估
checks = [
self.check_statistical_accuracy(report),
self.check_visualization_suggestions(report),
self.check_business_insights(report),
self.check_actionability(report)
]
# 迭代改进
for check in checks:
if not check["passed"]:
report = self.improve_report(report, check["feedback"])
return {"report": report, "quality_checks": checks}六、与其他模式的对比
6.1 Reflection vs ReAct
# ReAct 模式：思考-行动-观察循环
"""
ReAct Pattern:
Thought → Action → Observation → Thought → Action → ... → Answer
适用场景：
- 需要与外部工具交互
- 多步骤推理任务
- 信息收集类任务
特点：
- 强调与环境的交互
- 每一步都有明确的行动
- 观察结果驱动下一步
"""
# Reflection 模式：生成-评估-改进循环
"""
Reflection Pattern:
Generate → Evaluate → Refine → Evaluate → ... → Final Output
适用场景：
- 内容生成任务
- 需要高质量输出
- 自我修正能力重要
特点：
- 强调自我评估
- 不需要外部工具
- 内部质量驱动改进
"""
# 组合使用：ReAct + Reflection
class ReActWithReflection:
"""结合 ReAct 和 Reflection 的 Agent"""
def execute(self, query: str) -> str:
# 使用 ReAct 收集信息和推理
react_result = self.react_agent.run(query)
# 使用 Reflection 优化最终输出
refined_result = self.reflection_agent.refine(react_result)
return refined_result6.2 Reflection vs Plan-and-Execute
| 维度 | Plan-and-Execute | Reflection |
|---|---|---|
| 核心思想 | 先规划，后执行 | 生成后评估改进 |
| 适用任务 | 复杂多步骤任务 | 内容生成类任务 |
| 迭代触发 | 步骤失败时重新规划 | 质量不达标时改进 |
| 优势 | 结构清晰、可追踪 | 输出质量高 |
| 劣势 | 规划可能僵化 | 可能过度迭代 |
| 最佳组合 | Plan → Execute → Reflect |
6.3 选择决策树
任务类型
│
┌───────────┼───────────┐
│ │ │
工具使用 内容生成 复杂规划
│ │ │
▼ ▼ ▼
┌─────┐ ┌─────┐ ┌─────┐
│ReAct│ │Reflect│ │Plan │
│ │ │ion │ │&Exec│
└──┬──┘ └──┬──┘ └──┬──┘
│ │ │
└──────────┴──────────┘
│
需要高质量输出？
│
┌──────────┴──────────┐
│ │
是 否
│ │
▼ ▼
添加 Reflection 保持原模式
层进行优化七、高级技巧
7.1 反思链（Chain of Reflection）
class ChainOfReflection:
"""反思链：多层次的深度反思"""
def deep_reflect(self, content: str, depth: int = 3) -> str:
"""多层次反思"""
current = content
for level in range(depth):
# 每一层关注不同粒度
if level == 0:
# 第一层：宏观结构
current = self.reflect_structure(current)
elif level == 1:
# 第二层：段落逻辑
current = self.reflect_paragraphs(current)
else:
# 第三层：句子表达
current = self.reflect_sentences(current)
return current
def reflect_structure(self, content: str) -> str:
"""反思整体结构"""
prompt = f"""从宏观结构角度评估以下内容：
{content}
关注：整体框架、章节安排、逻辑流
提供改进后的完整版本。"""
response = llm.invoke([HumanMessage(content=prompt)])
return response.content7.2 反思记忆
class ReflectionMemory:
"""反思记忆系统"""
def __init__(self):
self.reflection_history = []
self.pattern_library = {}
def store_reflection(self, original: str, reflection: str, improved: str):
"""存储反思经验"""
self.reflection_history.append({
"original": original,
"reflection": reflection,
"improved": improved,
"timestamp": datetime.now()
})
# 提取常见模式
self.extract_patterns(reflection)
def extract_patterns(self, reflection: str):
"""从反思中提取常见改进模式"""
# 识别常见问题类型
common_issues = [
"逻辑不清晰", "事实错误", "表达冗余",
"结构混乱", "缺乏例证"
]
for issue in common_issues:
if issue in reflection:
self.pattern_library[issue] = self.pattern_library.get(issue, 0) + 1
def get_relevant_patterns(self, content: str) -> List[str]:
"""获取与当前内容相关的改进模式"""
relevant = []
for pattern, count in self.pattern_library.items():
if count > 2: # 出现超过2次的模式
relevant.append(pattern)
return relevant7.3 对抗性反思
class AdversarialReflection:
"""对抗性反思：多个评审员辩论"""
def __init__(self):
self.proponent = ChatOpenAI(model="gpt-4o", temperature=0.7) # 支持方
self.opponent = ChatOpenAI(model="gpt-4o", temperature=0.7) # 反对方
self.judge = ChatOpenAI(model="gpt-4o", temperature=0.2) # 裁判
def debate_reflection(self, draft: str, rounds: int = 2) -> Dict:
"""通过辩论进行反思"""
debate_history = []
for round_num in range(rounds):
# 支持方辩护
pro_args = self.proponent.invoke([
HumanMessage(content=f"为以下回答辩护，指出其优点：\n{draft}")
])
# 反方批评
con_args = self.opponent.invoke([
HumanMessage(content=f"批评以下回答，指出其问题：\n{draft}\n\n辩护方观点：{pro_args.content}")
])
debate_history.append({
"round": round_num + 1,
"pro": pro_args.content,
"con": con_args.content
})
# 裁判裁决
judge_prompt = f"""基于以下辩论，给出最终评估和改进建议：
{json.dumps(debate_history, ensure_ascii=False, indent=2)}
请提供：
1. 最终评分
2. 关键改进点
3. 修订建议"""
verdict = self.judge.invoke([HumanMessage(content=judge_prompt)])
return {
"debate_history": debate_history,
"verdict": verdict.content
}八、常见问题与解决方案
8.1 问题：反思过度导致输出质量下降
原因：迭代次数过多，模型开始过度修正，引入新问题。
解决方案：
def smart_termination(current_draft: str, previous_draft: str,
iteration: int) -> bool:
"""智能终止条件"""
# 1. 计算与上一版本的相似度
similarity = calculate_similarity(current_draft, previous_draft)
# 2. 如果变化很小，说明已经收敛
if similarity > 0.95:
return True
# 3. 检查质量是否开始下降
current_score = evaluate_quality(current_draft)
previous_score = evaluate_quality(previous_draft)
if current_score < previous_score * 0.9: # 质量下降超过10%
return True # 回退到上一版本
return False8.2 问题：反思过于严苛，永远无法满足
原因：评估标准过高或评估提示词设计不当。
解决方案：
class BalancedReflection:
"""平衡的反思机制"""
def __init__(self):
self.threshold_decay = 0.95 # 阈值衰减
self.base_threshold = 8.0
def get_adaptive_threshold(self, iteration: int) -> float:
"""自适应调整阈值"""
# 随着迭代进行，适当降低阈值
return self.base_threshold * (self.threshold_decay ** iteration)
def reflect_with_balance(self, draft: str, iteration: int) -> Dict:
"""平衡的反思"""
threshold = self.get_adaptive_threshold(iteration)
# 同时评估优点和缺点
prompt = f"""请全面评估以下内容，既要指出问题，也要认可优点：
{draft}
要求：
1. 列出至少2个优点
2. 列出需要改进的地方
3. 只有当存在严重问题时才建议修订
4. 评分标准：{threshold}/10 为通过线"""
response = llm.invoke([HumanMessage(content=prompt)])
return self.parse_balanced_feedback(response.content)8.3 问题：延迟过高
解决方案：
import asyncio
from concurrent.futures import ThreadPoolExecutor
class AsyncReflection:
"""异步并行反思"""
def __init__(self):
self.executor = ThreadPoolExecutor(max_workers=3)
async def parallel_reflect(self, draft: str) -> Dict:
"""并行执行多个反思任务"""
tasks = [
self.check_accuracy(draft),
self.check_logic(draft),
self.check_style(draft)
]
results = await asyncio.gather(*tasks)
return {
"accuracy": results[0],
"logic": results[1],
"style": results[2]
}九、总结
Reflection 模式是提升 AI Agent 输出质量的有力工具。通过引入自我评估和迭代改进机制，它能够：
- 显著提升输出质量：通过多轮迭代，逐步消除错误和不足
- 增强可靠性：自我纠错能力减少了对人工审核的依赖
- 适应性强：可应用于内容生成、代码编写、数据分析等多种场景
关键要点回顾
| 要点 | 说明 |
|---|---|
| 核心机制 | Generate → Evaluate → Refine 的迭代循环 |
| 两种形式 | Self-Reflection（单 Agent）和 Multi-Reflection（多 Agent） |
| 关键设计 | 评估提示词设计、迭代终止条件、成本控制 |
| 最佳实践 | 分层反思、并行评估、智能终止、缓存优化 |
| 组合策略 | 可与 ReAct、Plan-and-Execute 等模式结合使用 |
下一步学习
- 探索 LangGraph 的高级特性，构建更复杂的反思工作流
- 研究 多模态 Reflection，扩展到图像、代码等领域
- 实践 反思与工具使用 的结合，构建更强大的 Agent
参考资源
- LangGraph Documentation
- Reflexion: Self-Reflective Agents
- CRITIC: Large Language Models Can Self-Correct
- Tree of Thoughts
本文代码示例基于 LangGraph 0.1.x 和 LangChain 0.2.x 版本编写，建议读者使用最新版本以获得最佳体验。