反思与自我批评循环
能够评估自身输出并生成改进建议的智能体
反思与自我批评循环 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
什么是代理自我反思？
自我反思是指让代理在将刚完成的输出返回给用户before，或紧接着返回后，评估自己的输出。代理会充当自己的批评者。
这类似于专家审阅自己工作的方式：起草 → 批评 → 修订。将这一循环加入代理后，通常可以在不更改底层模型的情况下提高输出质量。
反思提示模式
代理生成响应后，将该响应与结构化反思提示一起再次输入模型。然后，模型会识别不足之处并提出改进建议。
import anthropic
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
def reflect_on_response(task: str, response: str) -> str:
reflection_prompt = (
'You just completed the following task:\n\n'
f'TASK: {task}\n\n'
f'YOUR RESPONSE:\n{response}\n\n'
'Please reflect on your performance by answering:\n'
'1. What did you do well?\n'
'2. What could be improved?\n'
'3. What would you do differently if you had to redo this?\n'
'Be specific and honest.'
)
result = client.messages.create(
model='claude-opus-4-5',
max_tokens=512,
messages=[{'role': 'user', 'content': reflection_prompt}]
)
return result.content[0].text结构化反思输出
非结构化的反思文字很难通过程序处理。请要求模型生成结构化的反思结果，以便可靠地提取评分和行动项。
STRUCTURED_REFLECTION_PROMPT = '''
Reflect on the task and response above. Return ONLY valid JSON:
{
"strengths": ["..."],
"weaknesses": ["..."],
"alternative_approach": "...",
"quality_score": 0.0,
"retry_recommended": false
}
quality_score: 0.0 (terrible) to 1.0 (excellent).
retry_recommended: true if quality_score < 0.6.
'''
import json
def structured_reflect(task: str, response: str, client) -> dict:
prompt = f'TASK: {task}\n\nRESPONSE: {response}\n\n{STRUCTURED_REFLECTION_PROMPT}'
result = client.messages.create(
model='claude-opus-4-5',
max_tokens=512,
messages=[{'role': 'user', 'content': prompt}]
)
text = result.content[0].text.strip()
# strip markdown code fences if present
if text.startswith('```'):
text = text.split('```')[1].lstrip('json').strip()
return json.loads(text)
if __name__ == '__main__':
class FakeContent:
def __init__(self, text):
self.text = text
class FakeResponse:
def __init__(self, text):
self.content = [FakeContent(text)]
class FakeMessages:
def create(self, **kwargs):
return FakeResponse(
'{"strengths": ["clear"], "weaknesses": ["too long"], '
'"alternative_approach": "be more concise", '
'"quality_score": 0.7, "retry_recommended": false}'
)
class FakeClient:
def __init__(self):
self.messages = FakeMessages()
result = structured_reflect('Summarize the article', 'A very long response...', FakeClient())
print('quality_score:', result['quality_score'])
print('weaknesses:', result['weaknesses'])
自我批评循环：低分时重试
当反思评分低于阈值时，请自动重试任务，并将反思中的不足之处和替代方案作为额外上下文。这样便可在单次代理运行中形成由反馈驱动的改进循环。
def agent_with_self_critique(task: str, client, max_retries: int = 2) -> str:
response = run_agent(task, client)
for attempt in range(max_retries):
reflection = structured_reflect(task, response, client)
print(f'Attempt {attempt+1} quality: {reflection["quality_score"]:.2f}')
if not reflection['retry_recommended']:
break
# Enrich the task with reflection insights
improved_task = (
f'{task}\n\n'
'Previous attempt weaknesses:\n'
+ '\n'.join(f'- {w}' for w in reflection['weaknesses'])
+ f'\n\nSuggested approach: {reflection["alternative_approach"]}'
)
response = run_agent(improved_task, client)
return response
def run_agent(task: str, client) -> str:
result = client.messages.create(
model='claude-opus-4-5',
max_tokens=1024,
messages=[{'role': 'user', 'content': task}]
)
return result.content[0].text用于反思的情景记忆
单次反思只能发挥一次作用；存储下来的反思会成为情景记忆，帮助代理跨会话学习。每条反思都是一个情景：任务上下文 + 发生的事情 + 代理学到的内容。
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional
@dataclass
class ReflectionEpisode:
episode_id: str
task_type: str # e.g. 'summarize', 'code_review', 'translate'
task_summary: str # short description (not full text)
quality_score: float
strengths: list
weaknesses: list
alternative_approach: str
timestamp: str = ''
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.utcnow().isoformat()
def to_dict(self) -> dict:
return asdict(self)
# Example
episode = ReflectionEpisode(
episode_id='ep_001',
task_type='summarize',
task_summary='Summarize a 5-page financial report',
quality_score=0.55,
strengths=['Identified key figures'],
weaknesses=['Too verbose', 'Missed conclusion'],
alternative_approach='Lead with the executive summary first'
)
print(episode.to_dict())持久化存储反思
将反思情景持久化到 JSON 文件或数据库中。启动时，针对相同任务类型 load 最近的情景，并将其注入上下文——代理会从过去的表现中学习。
import json
import os
MEMORY_FILE = 'agent_episodic_memory.json'
def save_episode(episode: ReflectionEpisode):
episodes = load_all_episodes()
episodes.append(episode.to_dict())
with open(MEMORY_FILE, 'w') as f:
json.dump(episodes, f, indent=2)
def load_all_episodes() -> list:
if not os.path.exists(MEMORY_FILE):
return []
with open(MEMORY_FILE) as f:
return json.load(f)
def load_recent_episodes(task_type: str, n: int = 3) -> list:
all_ep = load_all_episodes()
matching = [e for e in all_ep if e['task_type'] == task_type]
# Sort by timestamp descending, take most recent n
matching.sort(key=lambda e: e['timestamp'], reverse=True)
return matching[:n]将过去的反思注入上下文
开始任务前，获取该任务类型最近的情景反思，并将其加入系统提示。代理 now 知道上次犯了哪些错误，因此可以主动避免这些错误。
def build_system_prompt_with_memory(task_type: str) -> str:
base = 'You are a helpful AI assistant. Complete the task carefully.'
episodes = load_recent_episodes(task_type, n=3)
if not episodes:
return base
memory_block = '\n\nYour recent performance on similar tasks:\n'
for ep in episodes:
memory_block += (
f'- Score {ep["quality_score"]:.2f}: '
f'Weaknesses: {ep["weaknesses"]}. '
f'Better approach: {ep["alternative_approach"]}\n'
)
memory_block += '\nApply these lessons to your current response.'
return base + memory_block
# Before each task:
system_prompt = build_system_prompt_with_memory('summarize')
print(system_prompt[:300])对工具使用进行反思
对于使用工具的代理，反思更有价值，因为代理可以反思自己的工具调用策略：是否使用了正确的工具、正确的顺序和正确的参数？
TOOL_REFLECTION_PROMPT = '''
You completed a multi-step task using tools. Reflect on your tool usage:
Tool call log:
{tool_log}
Final result: {result}
Answer:
1. Were all tool calls necessary?
2. Were there redundant or incorrect calls?
3. What is the optimal tool sequence for this task type?
Return JSON:
{{
"redundant_calls": [],
"incorrect_calls": [],
"optimal_sequence": [],
"efficiency_score": 0.0
}}
'''
def reflect_on_tool_use(tool_log: list, result: str, client) -> dict:
import json
log_str = json.dumps(tool_log, indent=2)
prompt = TOOL_REFLECTION_PROMPT.format(
tool_log=log_str, result=result
)
response = client.messages.create(
model='claude-opus-4-5',
max_tokens=512,
messages=[{'role': 'user', 'content': prompt}]
)
return json.loads(response.content[0].text)
if __name__ == '__main__':
class FakeContent:
def __init__(self, text):
self.text = text
class FakeResponse:
def __init__(self, text):
self.content = [FakeContent(text)]
class FakeMessages:
def create(self, **kwargs):
return FakeResponse(
'{"redundant_calls": ["search(x)"], "incorrect_calls": [], '
'"optimal_sequence": ["search", "summarize"], "efficiency_score": 0.8}'
)
class FakeClient:
def __init__(self):
self.messages = FakeMessages()
tool_log = [{'tool': 'search', 'args': {'q': 'x'}}, {'tool': 'search', 'args': {'q': 'x'}}]
reflection = reflect_on_tool_use(tool_log, 'Found the answer', FakeClient())
print('Efficiency score:', reflection['efficiency_score'])
print('Redundant calls:', reflection['redundant_calls'])
情景记忆的衰减与修剪
如果现实发生变化或模型得到更新，旧反思就会过时。请实现衰减机制：提高近期情景的权重，并修剪早于某个阈值的情景或质量评分很低的情景（它们可能是异常值）。
from datetime import datetime, timedelta
def prune_old_episodes(
episodes: list,
max_age_days: int = 30,
min_quality: float = 0.0
) -> list:
cutoff = datetime.utcnow() - timedelta(days=max_age_days)
kept = []
for ep in episodes:
ep_time = datetime.fromisoformat(ep['timestamp'])
if ep_time >= cutoff and ep['quality_score'] >= min_quality:
kept.append(ep)
return kept
def weighted_episodes(episodes: list) -> list:
now = datetime.utcnow()
for ep in episodes:
age_days = (now - datetime.fromisoformat(ep['timestamp'])).days
# Recency weight: 1.0 today, halves every 7 days
ep['weight'] = 0.5 ** (age_days / 7)
return sorted(episodes, key=lambda e: e['weight'], reverse=True)
if __name__ == '__main__':
now = datetime.utcnow()
episodes = [
{'timestamp': (now - timedelta(days=2)).isoformat(), 'quality_score': 0.9, 'content': 'recent good episode'},
{'timestamp': (now - timedelta(days=45)).isoformat(), 'quality_score': 0.8, 'content': 'old episode'},
{'timestamp': (now - timedelta(days=10)).isoformat(), 'quality_score': 0.3, 'content': 'low quality episode'},
]
kept = prune_old_episodes(episodes, max_age_days=30, min_quality=0.5)
print(f'Kept {len(kept)} of {len(episodes)} episodes after pruning')
for ep in weighted_episodes(kept):
print(f" weight={ep['weight']:.3f} content={ep['content']}")
衡量反思的有效性
通过比较首次尝试与最终（反思后的）尝试的质量评分，跟踪自我批评是否确实改善了结果。如果改进幅度很小或变差，可能需要调整反思提示。
def measure_reflection_gain(run_log: list) -> dict:
"""
run_log: list of dicts with keys 'attempt', 'quality_score'
e.g. [{'attempt': 1, 'quality_score': 0.55}, {'attempt': 2, 'quality_score': 0.78}]
"""
if not run_log:
return {}
first_score = run_log[0]['quality_score']
best_score = max(r['quality_score'] for r in run_log)
final_score = run_log[-1]['quality_score']
return {
'first_attempt_score': first_score,
'final_score': final_score,
'best_score': best_score,
'absolute_gain': final_score - first_score,
'relative_gain_pct': ((final_score - first_score) / max(first_score, 0.001)) * 100,
'retries': len(run_log) - 1
}
log = [
{'attempt': 1, 'quality_score': 0.55},
{'attempt': 2, 'quality_score': 0.78}
]
print(measure_reflection_gain(log))反思循环的安全措施
如果没有限制，反思循环可能会无限运行。请始终强制执行：最大重试次数、用于提前退出的最低评分阈值，以及 time 预算。记录所有反思，以便审计循环行为。
import time
def safe_reflection_loop(
task: str,
client,
max_retries: int = 3,
quality_target: float = 0.75,
time_budget_seconds: float = 30.0
) -> dict:
start = time.time()
response = run_agent(task, client)
run_log = []
for attempt in range(max_retries + 1):
if time.time() - start > time_budget_seconds:
print('Time budget exceeded, returning best result')
break
reflection = structured_reflect(task, response, client)
run_log.append({'attempt': attempt + 1,
'quality_score': reflection['quality_score']})
if reflection['quality_score'] >= quality_target:
print(f'Quality target reached at attempt {attempt + 1}')
break
if attempt < max_retries:
response = run_agent(task + '\n' + reflection['alternative_approach'], client)
return {'response': response, 'run_log': run_log,
'gain': measure_reflection_gain(run_log)}知识检查
将反思情景存储为情景记忆的主要好处是什么？
回顾：反思与自我批评循环
非常出色！以下是您在本课中学习的内容：
- 反思提示：包含优势、不足、质量评分和重试标志的结构化数据
- 自我批评循环：低分时重试，并使用反思洞见丰富任务
- 情景记忆：按任务类型将反思存储为带时间戳的情景
- 记忆注入：每次运行前将最近的情景加载到系统提示中
- 安全措施：最大重试次数、时间预算，以及达到质量目标时提前退出
下一步：了解如何利用成功和失败的轨迹进行更深入的自我改进。
常见问题解答
「反思与自我批评循环」课时是免费的吗？
是的 — 「反思与自我批评循环」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「反思与自我批评循环」这节课中我会学到什么？
能够评估自身输出并生成改进建议的智能体 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「反思与自我批评循环」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。