Agent 反思机制
大家好,我是程序员小策。
Agent 的反思机制是什么?
为什么要让 Agent 反思,不反思会出什么问题?
具体怎么实现,从 LangGraph 到 AutoGen 到 CrewAI 都有哪些姿势?
一、为什么「能跑」不等于「能跑对」
去年帮一个朋友看他做的研究型 Agent,模型用的 Claude 4.5,工具链也接得整整齐齐:能搜网页、能读 PDF、能跑代码。看着挺唬人,但跑出来的报告他自己都不敢用。
我问他:「为啥不敢用?」
他苦笑:「每次跑出来的结论都不一样,第一段明明说『该药对 X 病症有效』,第五段突然又说『需要更多临床证据』。让它重跑一次,又改口了。Agent 不知道自己前后矛盾,也不知道自己哪句话是猜的、哪句话是证据支撑的。」
我当时脑子里蹦出来的词就是——这个 Agent 不会反思。
它就像一个刚拿到驾照的新手司机,只知道按导航走,不知道回头看一眼刚才那个弯道是不是拐早了、那个变道是不是没看后视镜。不反思的 Agent,本质上是一个「永远只走第一遍」的司机。你让它开 100 公里,它能跑完全程,但每公里都可能是错的。
更扎心的事实:哪怕是 GPT-5 / Claude 4.5 这种级别的模型,单次推理的「一次通过率」在复杂任务上也只有 50%-70%。剩下那 30%-50%,不是模型不够聪明,而是它没有机会看到自己的错误。
这就是「反思机制」要解决的问题。
二、什么是 Agent 反思机制
先把这几个最容易混的词儿掰扯清楚——Self-Reflection、Self-Critique、Reflexion、Self-Refine,很多文章里把它们当同义词用,其实是四套不同的思路。
Agent 反思机制:让 Agent 在完成任务后(或每一步推理后),显式生成对自己输出的评估或批评,并把这份反思作为下一轮决策的输入,不更新模型参数。
四个常被混用的概念:
- Self-Reflection(自我反思):最宽泛的概念。Agent 用自然语言写一段「我刚才哪儿做错了、下次该怎么改」的总结。本质是 self-generated feedback。
- Self-Critique(自我批评):更聚焦,Agent 扮演「批评者」角色,对自己刚才的输出打分、挑刺,给出可执行的修改建议。常见做法是同一段 prompt 让 LLM 既当「答题者」又当「阅卷人」。
- Reflexion(NeurIPS 2023,noahshinn/reflexion ):把反思结果持久化到 episodic memory,下一次同任务重试时,Actor 能「想起」自己上次怎么错的。论文里 HumanEval 一次通过率从 80% 提到 91%,AlfWorld 决策任务 134/134 全部解决。
- Self-Refine(NeurIPS 2023):迭代式自我改进,强调「同一份输出反复 refinement」直到质量收敛。每一轮都用上轮的 critique 去修。
类比一下:医院里每周的 M&M 病例讨论会(Morbidity and Mortality Conference)。
医生做完一台手术,不直接进下一台。先去会议室,团队一起对刚才这台手术做复盘:哪一刀切深了、哪一步缝合漏了、下次该怎么避免。然后把这份复盘写进病历里,下一次同类型手术前先翻一翻。这就是 Agent 反思机制在做的事——把失败经验沉淀成下次决策的上下文。
而 Self-Reflection 是「医生自己写复盘笔记」,Self-Critique 是「主刀医生自己当评委给自己打分」,Reflexion 是「复盘笔记归档到病历系统、下次开术前必查」,Self-Refine 是「写错了就重写、直到主任医师签字放行」。
三、Reflexion 的核心循环:4 步闭环
不管是哪个框架的反思实现,底层的循环结构都长这样:
四个角色:
- Actor:负责产生动作的 LLM(通常是 ReAct 风格)。
- Environment / Tool:执行动作、返回 observation。
- Evaluator:判定 Actor 这一轮是否成功。可以是另一个 LLM,也可以是确定性规则——比如代码任务看测试是否通过、SQL 任务看查询是否报错。
- Self-Reflection:失败时,单独调一次 LLM,让它基于「任务 + Actor 的轨迹 + 失败原因」生成一段自然语言反思,写进 Memory。
下一轮 Actor 调用时,Memory 里的反思文本会和原任务一起塞进 prompt,相当于「这次你重做一遍,但记住上次的教训」。
注意:整个过程不更新 LLM 权重,完全是 prompt 层面的「以人为鉴」。
四、三大主流框架的反思实现对比
4.1 LangGraph:把反思做成图上的一个节点
LangGraph 1.x 的核心是 StateGraph,反思就是加一个独立的 reflection node,通过条件边决定是否回环重试。
# langgraph_reflection.py
# 完整可运行:Agent 答错就反思、重答,最多 3 轮
import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class State(TypedDict):
messages: Annotated[list, add_messages]
reflections: list[str] # 反思历史
attempt: int # 当前是第几轮
def actor_node(state: State):
"""根据历史反思重新回答"""
history = "\n".join(
f"第 {i+1} 轮反思:{r}" for i, r in enumerate(state["reflections"])
)
feedback = history or "这是第一轮,没有历史反馈。"
prompt = f"""请回答用户的问题。
过去几轮你犯过这些错,请避免重蹈覆辙:
{feedback}
用户问题:{state['messages'][-1].content}
"""
resp = llm.invoke(prompt)
return {"messages": [resp], "attempt": state.get("attempt", 0) + 1}
def evaluator_node(state: State) -> Literal["reflect", "end"]:
"""判定答案是否合格——这里用 LLM 当评委"""
last = state["messages"][-1].content
grade_prompt = f"""你是质检员。判断下面这段回答是否合格。
- 如果有事实错误、自相矛盾、答非所问,回答 NO
- 如果逻辑清晰、事实正确、回答完整,回答 YES
回答内容:
{last}
只回答 YES 或 NO。"""
verdict = llm.invoke(grade_prompt).content.strip().upper()
if "YES" in verdict or state["attempt"] >= 3:
return "end"
return "reflect"
def reflect_node(state: State):
"""对刚才的回答做反思"""
last = state["messages"][-1].content
reflect_prompt = f"""你刚才的回答是:
{last}
请用 3 句话反思:
1. 哪里可能错了 / 不够好
2. 下次该怎么避免
3. 下次回答时需要特别注意什么
只输出反思文本。"""
reflection = llm.invoke(reflect_prompt).content
return {"reflections": state["reflections"] + [reflection]}
# --- 组装状态图 ---
builder = StateGraph(State)
builder.add_node("actor", actor_node)
builder.add_node("evaluator", evaluator_node)
builder.add_node("reflect", reflect_node)
builder.add_edge(START, "actor")
builder.add_edge("actor", "evaluator")
builder.add_conditional_edges("evaluator", evaluator_node, {
"end": END,
"reflect": "reflect",
})
builder.add_edge("reflect", "actor") # 反思完回到 actor 重答
graph = builder.compile()
# --- 跑一遍 ---
if __name__ == "__main__":
result = graph.invoke({
"messages": [("user", "请用 100 字介绍 LangGraph 的核心思想")],
"reflections": [],
"attempt": 0,
})
print("最终回答:", result["messages"][-1].content)
print("反思历史:", result["reflections"])
关键点:
evaluator是一个 conditional edge 函数,返回"end"或"reflect",决定是否回环。attempt >= 3是硬上限兜底,防止反思死循环烧光 token。reflections用list[str]累积,全部塞进下一轮 actor 的 prompt,让 LLM 看到「自己以前反思过什么」。
4.2 AutoGen:双 Agent 互怼式反思
AutoGen 0.4+ 的核心是 BaseAgent + on_message_impl,做反思最自然的姿势就是让两个 Agent 互相打分——一个负责生成、一个负责挑刺。
# autogen_reflection.py
# 完整可运行:AssistantAgent 答,CriticAgent 批,循环到满意为止
import asyncio
from autogen_core import AgentId, BaseAgent, MessageContext, SingleThreadedAgentRuntime
from dataclasses import dataclass
@dataclass
class Task:
content: str
@dataclass
class Answer:
content: str
@dataclass
class Critique:
content: str
approved: bool
# --- 主答 Agent ---
class AssistantAgent(BaseAgent):
def __init__(self) -> None:
super().__init__("Assistant")
async def on_message_impl(self, message, ctx: MessageContext) -> Answer:
# 真实场景这里调 LLM
answer = f"[assistant 模拟回答]: {message.content} 的解读是 A、B、C 三点。"
print(f"[Assistant] 给出回答: {answer}")
return Answer(content=answer)
# --- 批评 Agent ---
class CriticAgent(BaseAgent):
def __init__(self) -> None:
super().__init__("Critic")
async def on_message_impl(self, message: Answer, ctx: MessageContext) -> Critique:
# 真实场景调 LLM 当评委
text = message.content
approved = "A、B、C" in text and len(text) > 20
critique = Critique(content="回答覆盖了 A/B/C,但缺少 D 的分析" if not approved else "OK",
approved=approved)
print(f"[Critic] 判定 {'通过' if approved else '不通过'}: {critique.content}")
return critique
# --- 反思协调器:把两个 Agent 串起来 ---
class ReflectionCoordinator(BaseAgent):
def __init__(self, max_rounds: int = 3) -> None:
super().__init__("Coordinator")
self.max_rounds = max_rounds
async def on_message_impl(self, message: Task, ctx: MessageContext) -> str:
assistant = AgentId("Assistant", "default")
critic = AgentId("Critic", "default")
answer = await self.send_message(message, assistant)
for i in range(self.max_rounds):
critique = await self.send_message(answer, critic)
if critique.approved:
return answer.content
# 反思:把 critique 加进下一轮 prompt
print(f"[Coordinator] 第 {i+1} 轮不通过,让 Assistant 基于 critique 重答")
answer = await self.send_message(
Task(content=f"原问题:{message.content}\n上轮批评:{critique.content}"),
assistant,
)
return answer.content
async def main():
runtime = SingleThreadedAgentRuntime()
await AssistantAgent.register(runtime, "Assistant", lambda: AssistantAgent())
await CriticAgent.register(runtime, "Critic", lambda: CriticAgent())
await ReflectionCoordinator.register(runtime, "Coordinator",
lambda: ReflectionCoordinator(max_rounds=3))
runtime.start()
result = await runtime.send_message(
Task(content="请介绍 Agent 反思机制"),
AgentId("Coordinator", "default"),
)
print("\n最终结果:", result)
await runtime.stop_when_idle()
asyncio.run(main())
关键点:
ReflectionCoordinator是个第三方胶水 Agent,负责把「生成→批评→重生成」的循环包起来。max_rounds=3是硬兜底,避免死循环。- 反思 = 把 critic 的 critique 当 context 塞回 Assistant 的下一轮 prompt,和 LangGraph 是同一个套路。
4.3 CrewAI:把反思做成一个 Task
CrewAI 1.x 的抽象是 Agent + Task + Crew。反思不需要写循环代码,直接定义一个「反思 Task」交给一个 Reviewer Agent 干就行。
# crewai_reflection.py
# 完整可运行:Writer 写稿,Reviewer 反思,循环
from crewai import Agent, Task, Crew, Process
writer = Agent(
role="技术博主",
goal="写一段 200 字的技术介绍",
backstory="你擅长用大白话讲清楚复杂概念",
allow_delegation=False,
verbose=True,
)
reviewer = Agent(
role="资深审稿人",
goal="找出文章的事实错误、逻辑漏洞、表达不清",
backstory="你是一个 10 年经验的程序员,对错误零容忍",
allow_delegation=False,
verbose=True,
)
write_task = Task(
description="写 200 字介绍 Agent 反思机制(Self-Reflection)",
expected_output="一段 200 字左右、通顺、有技术含量、不夸张的介绍",
agent=writer,
)
# 关键:把反思做成一个独立 Task,让 Reviewer 跑 reflection
reflect_task = Task(
description="""对上一步 Task 的输出做反思:
1. 找出 3 个最明显的质量问题(事实 / 逻辑 / 表达)
2. 给出具体的修改建议
3. 如果质量问题太多,返回 'NEEDS_REWRITE' 让 writer 重写
4. 如果质量可以接受,返回 'OK' 并附上修改后的版本""",
expected_output="要么 'NEEDS_REWRITE',要么 'OK' + 改后版本",
agent=reviewer,
context=[write_task], # 关键:拿上一步 writer 的输出当 context
)
crew = Crew(
agents=[writer, reviewer],
tasks=[write_task, reflect_task],
process=Process.sequential,
verbose=True,
)
if __name__ == "__main__":
result = crew.kickoff()
print("\n===== 最终输出 =====")
print(result.raw)
关键点:
Task.context=[write_task]是 CrewAI 反思的精髓——前一个 Task 的输出自动作为后一个 Task 的 context。- Reviewer Agent 的
goal直接定义成「找问题、给建议」,让 LLM 天然进入批评者角色。 - 通过
Process.sequential+ 多个反思 Task 可以自然形成多轮反思链(写→批→再批→再写)。
五、看起来很完美,但这 5 个坑你可能踩过
坑 1:反思循环死循环,token 烧光
现象:Agent 反思 20 轮还没收敛,一次跑了几十块。
根因:没设 max_rounds 硬上限,evaluator 又用 LLM 判定,LLM 自己都可能前后矛盾。
解法:硬编码 attempt >= 3 或 max_rounds=3 兜底;evaluator 优先用确定性规则(测试用例、断言、SQL 报错信息)。
坑 2:反思把自己反思崩了(Reflection Collapse)
现象:第二轮、第三轮反思把正确的内容「反思」成错的,越反思越差。
根因:反思 prompt 没说「保留原文中正确的部分,只改错的」。
解法:反思 prompt 显式约束「只指出问题,不要改写正确部分」;或者只在第一轮反思时改,后续轮只追加补充。
坑 3:反思用同一份 LLM,永远挑不出自己的 bug
现象:让 GPT-4 反思 GPT-4 的输出,10 次有 8 次通过质检,但人一看还是有错。
根因:同一模型的「生成-批评」分布是高度相关的,模型抓不到自己没意识到的错误。
解法:用更强的模型当 evaluator(用 GPT-5 评 GPT-4-mini),或者多模型交叉评审(GPT-4 评 + Claude 评 + 规则评,三者多数通过)。
坑 4:反思 prompt 太啰嗦,把 context 撑爆
现象:每轮反思写 500 字,第三轮 actor 的 prompt 已经 2k 字。
根因:没控制反思文本长度,没做历史反思的压缩/裁剪。
解法:反思 prompt 限字数(3 句话以内);历史反思超过 3 条做滑动窗口 + 摘要压缩
坑 5:反思结果没落盘,下次同任务还是从零开始
现象:用户问同样的问题,Agent 还是犯同样的错。
根因:反思只活在这次 session 的 memory 里,没存到长期记忆。
解法:把反思结果持久化到 vector store 或关系数据库,下次同任务前先 similarity_search 拉历史反思塞进 prompt。
六、生产环境的 4 个考量
1. 反思的"成本/收益"控制
不是所有任务都值得反思。简单分类任务(“这段文本是中文还是英文”)上反思 = 浪费 token。建议:
- 简单任务(一次通过率 > 95%):不开反思
- 中等任务(一次通过率 60%-90%):开 1-2 轮反思
- 复杂任务(一次通过率 < 60%):开 3 轮反思 + 多模型交叉评审
2. 反思的可观测性
生产里要 trace 每轮反思的 prompt、response、判定理由。LangGraph 用 LangSmith,AutoGen 用 OpenTelemetry,CrewAI 用内置的 verbose=True + 自定义 callback。没有 trace 的反思,出问题你连定位都定位不了。
3. 反思的"非对称成本"
反思 = 至少多调 1 次 LLM = token ×2。生产里要做批量化反思(多个任务一起反思、共享 evaluator prompt)和缓存化反思(相同问题、相同失败模式的反思结果可以缓存复用)。
4. 反思的"冷启动"
新任务、新领域,反思记忆是空的,第一轮几乎等于无反思。可以用 few-shot prompt:「历史上类似任务的反思长这样……」来 warm up。
七、3 大框架反思能力对比
| 维度 | LangGraph | AutoGen | CrewAI |
|---|---|---|---|
| 反思抽象方式 | StateGraph 节点 + conditional edge | 多 Agent 互发消息 | Task 链 + context 传递 |
| 循环控制 | 显式 add_conditional_edges + 状态字段 |
协调器 Agent 控制 rounds | 多个反思 Task 串联 |
| 死循环兜底 | recursion_limit + 状态 attempt 计数 |
协调器 max_rounds |
Process 类型决定(sequential 不会死循环) |
| 反思落盘 | 配合 Checkpointer(MemorySaver / Postgres) | 自定义 Store 接数据库 | 配合 Memory 工具或外部存储 |
| 学习曲线 | 中(要懂图状态机) | 高(异步消息、Agent 生命周期) | 低(贴近人类分工直觉) |
| 多模型交叉评审 | 需要自己写多 evaluator 节点 | 原生支持多 Agent 互评 | 需要定义多个 Reviewer Agent |
| 生产成熟度 | ⭐⭐⭐⭐⭐(LangSmith 全套) | ⭐⭐⭐⭐(v0.4 异步架构稳) | ⭐⭐⭐(CrewAI AMP 商业版成熟) |
一句话总结:要精细控制反思循环 → 选 LangGraph;要多 Agent 互怼 / 学术场景 → 选 AutoGen;要快速搭多 Agent 团队、靠角色分工 → 选 CrewAI。
八、面试官还会追问什么
追问 1:Reflexion 和 Self-Refine 的本质区别是什么?
→ 回答方向:Reflexion 是「跨 episode 的反思」(同一个任务重做 N 轮,每轮之间用反思衔接),核心是记忆;Self-Refine 是「同 episode 内 的迭代」(同一份输出反复 refinement N 轮直到收敛),核心是同输入多次改稿。Reflexion 适合「做错了要重做」的任务,Self-Refine 适合「做对了但要更好」的任务。
追问 2:反思机制会不会让 Agent 越来越慢?怎么权衡?
→ 回答方向:会。反思 = 额外 LLM 调用,token 翻倍甚至 N 倍。生产里要做「任务分级 + 反思分级」:简单任务关反思,复杂任务开反思;同时把反思结果缓存化(同类失败模式复用反思文本)。另外反思 prompt 要强约束字数(3 句话以内),别让反思把 context 撑爆。
追问 3:怎么判断 Agent 的反思真的有效,而不是「自我感动」?
→ 回答方向:跑消融实验。同一批任务,跑「无反思」「反思 N 轮」「用更强模型反思」三组,对比:(1) 一次通过率,(2) token 成本,(3) 人工抽检质量。如果反思组通过率显著高于无反思组、且 token 成本可以接受,才是真的有效。只看通过率不看成本 = 自我感动。
追问 4:反思机制能替代 fine-tuning 吗?
→ 回答方向:不能完全替代。反思是「prompt 层面的临时记忆」,每次 session 都要重新生成;fine-tuning 是「权重层面的永久记忆」,一次训练永久生效。反思解决的是「这一次怎么改」,fine-tuning 解决的是「从根本上变聪明」。生产里两者配合用:反思处理任务级失败,fine-tuning 处理领域级能力差距。
追问 5:你们生产里反思机制踩过最深的坑是什么?
(这个问题留给你——我相信每个做过 Agent 反思的人都有一个「反思把自己反思崩了」或者「反思一晚上 token 烧了几千块」的故事,欢迎评论区分享。)
九、总结:反思是 Agent 的「错题本」
Agent 反思机制的本质,就是给模型一个机会看到自己错了。
人之所以能从新手变老手,不是因为脑子突然变好了,是因为我们有一本错题本。每次翻看错题,都相当于给大脑做一次「以人为鉴」。Agent 反思机制在做的事一模一样——让 LLM 在不更新权重的情况下,靠「看自己的失败记录」来变聪明。
如果你读完这篇,应该能:
- 区分 Self-Reflection / Self-Critique / Reflexion / Self-Refine 这 4 个概念
- 画出 Reflexion 4 步闭环(Actor → Environment → Evaluator → Self-Reflection → Memory)
- 在 LangGraph / AutoGen / CrewAI 三个框架里挑一个把反思跑起来
- 避开 反思死循环 / 反思崩塌 / 反思撑爆 context 这 3 个最常见的坑
如果这篇文章让你少烧几千块 token,欢迎点赞 + 在看。
最后一个问题留给你:你团队里的 Agent,现在是靠「prompt 调优」硬扛质量,还是已经在用反思机制了?如果还没有,下周可以先挑一个高频失败任务试一周,看 token 成本下降多少。
Agent 反思机制
大家好,我是程序员小策。
Agent 的反思机制是什么?
为什么要让 Agent 反思,不反思会出什么问题?
具体怎么实现,从 LangGraph 到 AutoGen 到 CrewAI 都有哪些姿势?
一、为什么「能跑」不等于「能跑对」
去年帮一个朋友看他做的研究型 Agent,模型用的 Claude 4.5,工具链也接得整整齐齐:能搜网页、能读 PDF、能跑代码。看着挺唬人,但跑出来的报告他自己都不敢用。
我问他:「为啥不敢用?」
他苦笑:「每次跑出来的结论都不一样,第一段明明说『该药对 X 病症有效』,第五段突然又说『需要更多临床证据』。让它重跑一次,又改口了。Agent 不知道自己前后矛盾,也不知道自己哪句话是猜的、哪句话是证据支撑的。」
我当时脑子里蹦出来的词就是——这个 Agent 不会反思。
它就像一个刚拿到驾照的新手司机,只知道按导航走,不知道回头看一眼刚才那个弯道是不是拐早了、那个变道是不是没看后视镜。不反思的 Agent,本质上是一个「永远只走第一遍」的司机。你让它开 100 公里,它能跑完全程,但每公里都可能是错的。
更扎心的事实:哪怕是 GPT-5 / Claude 4.5 这种级别的模型,单次推理的「一次通过率」在复杂任务上也只有 50%-70%。剩下那 30%-50%,不是模型不够聪明,而是它没有机会看到自己的错误。
这就是「反思机制」要解决的问题。
二、什么是 Agent 反思机制
先把这几个最容易混的词儿掰扯清楚——Self-Reflection、Self-Critique、Reflexion、Self-Refine,很多文章里把它们当同义词用,其实是四套不同的思路。
Agent 反思机制:让 Agent 在完成任务后(或每一步推理后),显式生成对自己输出的评估或批评,并把这份反思作为下一轮决策的输入,不更新模型参数。
四个常被混用的概念:
- Self-Reflection(自我反思):最宽泛的概念。Agent 用自然语言写一段「我刚才哪儿做错了、下次该怎么改」的总结。本质是 self-generated feedback。
- Self-Critique(自我批评):更聚焦,Agent 扮演「批评者」角色,对自己刚才的输出打分、挑刺,给出可执行的修改建议。常见做法是同一段 prompt 让 LLM 既当「答题者」又当「阅卷人」。
- Reflexion(NeurIPS 2023,noahshinn/reflexion ):把反思结果持久化到 episodic memory,下一次同任务重试时,Actor 能「想起」自己上次怎么错的。论文里 HumanEval 一次通过率从 80% 提到 91%,AlfWorld 决策任务 134/134 全部解决。
- Self-Refine(NeurIPS 2023):迭代式自我改进,强调「同一份输出反复 refinement」直到质量收敛。每一轮都用上轮的 critique 去修。
类比一下:医院里每周的 M&M 病例讨论会(Morbidity and Mortality Conference)。
医生做完一台手术,不直接进下一台。先去会议室,团队一起对刚才这台手术做复盘:哪一刀切深了、哪一步缝合漏了、下次该怎么避免。然后把这份复盘写进病历里,下一次同类型手术前先翻一翻。这就是 Agent 反思机制在做的事——把失败经验沉淀成下次决策的上下文。
而 Self-Reflection 是「医生自己写复盘笔记」,Self-Critique 是「主刀医生自己当评委给自己打分」,Reflexion 是「复盘笔记归档到病历系统、下次开术前必查」,Self-Refine 是「写错了就重写、直到主任医师签字放行」。
三、Reflexion 的核心循环:4 步闭环
不管是哪个框架的反思实现,底层的循环结构都长这样:
四个角色:
- Actor:负责产生动作的 LLM(通常是 ReAct 风格)。
- Environment / Tool:执行动作、返回 observation。
- Evaluator:判定 Actor 这一轮是否成功。可以是另一个 LLM,也可以是确定性规则——比如代码任务看测试是否通过、SQL 任务看查询是否报错。
- Self-Reflection:失败时,单独调一次 LLM,让它基于「任务 + Actor 的轨迹 + 失败原因」生成一段自然语言反思,写进 Memory。
下一轮 Actor 调用时,Memory 里的反思文本会和原任务一起塞进 prompt,相当于「这次你重做一遍,但记住上次的教训」。
注意:整个过程不更新 LLM 权重,完全是 prompt 层面的「以人为鉴」。
四、三大主流框架的反思实现对比
4.1 LangGraph:把反思做成图上的一个节点
LangGraph 1.x 的核心是 StateGraph,反思就是加一个独立的 reflection node,通过条件边决定是否回环重试。
# langgraph_reflection.py
# 完整可运行:Agent 答错就反思、重答,最多 3 轮
import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
class State(TypedDict):
messages: Annotated[list, add_messages]
reflections: list[str] # 反思历史
attempt: int # 当前是第几轮
def actor_node(state: State):
"""根据历史反思重新回答"""
history = "\n".join(
f"第 {i+1} 轮反思:{r}" for i, r in enumerate(state["reflections"])
)
feedback = history or "这是第一轮,没有历史反馈。"
prompt = f"""请回答用户的问题。
过去几轮你犯过这些错,请避免重蹈覆辙:
{feedback}
用户问题:{state['messages'][-1].content}
"""
resp = llm.invoke(prompt)
return {"messages": [resp], "attempt": state.get("attempt", 0) + 1}
def evaluator_node(state: State) -> Literal["reflect", "end"]:
"""判定答案是否合格——这里用 LLM 当评委"""
last = state["messages"][-1].content
grade_prompt = f"""你是质检员。判断下面这段回答是否合格。
- 如果有事实错误、自相矛盾、答非所问,回答 NO
- 如果逻辑清晰、事实正确、回答完整,回答 YES
回答内容:
{last}
只回答 YES 或 NO。"""
verdict = llm.invoke(grade_prompt).content.strip().upper()
if "YES" in verdict or state["attempt"] >= 3:
return "end"
return "reflect"
def reflect_node(state: State):
"""对刚才的回答做反思"""
last = state["messages"][-1].content
reflect_prompt = f"""你刚才的回答是:
{last}
请用 3 句话反思:
1. 哪里可能错了 / 不够好
2. 下次该怎么避免
3. 下次回答时需要特别注意什么
只输出反思文本。"""
reflection = llm.invoke(reflect_prompt).content
return {"reflections": state["reflections"] + [reflection]}
# --- 组装状态图 ---
builder = StateGraph(State)
builder.add_node("actor", actor_node)
builder.add_node("evaluator", evaluator_node)
builder.add_node("reflect", reflect_node)
builder.add_edge(START, "actor")
builder.add_edge("actor", "evaluator")
builder.add_conditional_edges("evaluator", evaluator_node, {
"end": END,
"reflect": "reflect",
})
builder.add_edge("reflect", "actor") # 反思完回到 actor 重答
graph = builder.compile()
# --- 跑一遍 ---
if __name__ == "__main__":
result = graph.invoke({
"messages": [("user", "请用 100 字介绍 LangGraph 的核心思想")],
"reflections": [],
"attempt": 0,
})
print("最终回答:", result["messages"][-1].content)
print("反思历史:", result["reflections"])
关键点:
evaluator是一个 conditional edge 函数,返回"end"或"reflect",决定是否回环。attempt >= 3是硬上限兜底,防止反思死循环烧光 token。reflections用list[str]累积,全部塞进下一轮 actor 的 prompt,让 LLM 看到「自己以前反思过什么」。
4.2 AutoGen:双 Agent 互怼式反思
AutoGen 0.4+ 的核心是 BaseAgent + on_message_impl,做反思最自然的姿势就是让两个 Agent 互相打分——一个负责生成、一个负责挑刺。
# autogen_reflection.py
# 完整可运行:AssistantAgent 答,CriticAgent 批,循环到满意为止
import asyncio
from autogen_core import AgentId, BaseAgent, MessageContext, SingleThreadedAgentRuntime
from dataclasses import dataclass
@dataclass
class Task:
content: str
@dataclass
class Answer:
content: str
@dataclass
class Critique:
content: str
approved: bool
# --- 主答 Agent ---
class AssistantAgent(BaseAgent):
def __init__(self) -> None:
super().__init__("Assistant")
async def on_message_impl(self, message, ctx: MessageContext) -> Answer:
# 真实场景这里调 LLM
answer = f"[assistant 模拟回答]: {message.content} 的解读是 A、B、C 三点。"
print(f"[Assistant] 给出回答: {answer}")
return Answer(content=answer)
# --- 批评 Agent ---
class CriticAgent(BaseAgent):
def __init__(self) -> None:
super().__init__("Critic")
async def on_message_impl(self, message: Answer, ctx: MessageContext) -> Critique:
# 真实场景调 LLM 当评委
text = message.content
approved = "A、B、C" in text and len(text) > 20
critique = Critique(content="回答覆盖了 A/B/C,但缺少 D 的分析" if not approved else "OK",
approved=approved)
print(f"[Critic] 判定 {'通过' if approved else '不通过'}: {critique.content}")
return critique
# --- 反思协调器:把两个 Agent 串起来 ---
class ReflectionCoordinator(BaseAgent):
def __init__(self, max_rounds: int = 3) -> None:
super().__init__("Coordinator")
self.max_rounds = max_rounds
async def on_message_impl(self, message: Task, ctx: MessageContext) -> str:
assistant = AgentId("Assistant", "default")
critic = AgentId("Critic", "default")
answer = await self.send_message(message, assistant)
for i in range(self.max_rounds):
critique = await self.send_message(answer, critic)
if critique.approved:
return answer.content
# 反思:把 critique 加进下一轮 prompt
print(f"[Coordinator] 第 {i+1} 轮不通过,让 Assistant 基于 critique 重答")
answer = await self.send_message(
Task(content=f"原问题:{message.content}\n上轮批评:{critique.content}"),
assistant,
)
return answer.content
async def main():
runtime = SingleThreadedAgentRuntime()
await AssistantAgent.register(runtime, "Assistant", lambda: AssistantAgent())
await CriticAgent.register(runtime, "Critic", lambda: CriticAgent())
await ReflectionCoordinator.register(runtime, "Coordinator",
lambda: ReflectionCoordinator(max_rounds=3))
runtime.start()
result = await runtime.send_message(
Task(content="请介绍 Agent 反思机制"),
AgentId("Coordinator", "default"),
)
print("\n最终结果:", result)
await runtime.stop_when_idle()
asyncio.run(main())
关键点:
ReflectionCoordinator是个第三方胶水 Agent,负责把「生成→批评→重生成」的循环包起来。max_rounds=3是硬兜底,避免死循环。- 反思 = 把 critic 的 critique 当 context 塞回 Assistant 的下一轮 prompt,和 LangGraph 是同一个套路。
4.3 CrewAI:把反思做成一个 Task
CrewAI 1.x 的抽象是 Agent + Task + Crew。反思不需要写循环代码,直接定义一个「反思 Task」交给一个 Reviewer Agent 干就行。
# crewai_reflection.py
# 完整可运行:Writer 写稿,Reviewer 反思,循环
from crewai import Agent, Task, Crew, Process
writer = Agent(
role="技术博主",
goal="写一段 200 字的技术介绍",
backstory="你擅长用大白话讲清楚复杂概念",
allow_delegation=False,
verbose=True,
)
reviewer = Agent(
role="资深审稿人",
goal="找出文章的事实错误、逻辑漏洞、表达不清",
backstory="你是一个 10 年经验的程序员,对错误零容忍",
allow_delegation=False,
verbose=True,
)
write_task = Task(
description="写 200 字介绍 Agent 反思机制(Self-Reflection)",
expected_output="一段 200 字左右、通顺、有技术含量、不夸张的介绍",
agent=writer,
)
# 关键:把反思做成一个独立 Task,让 Reviewer 跑 reflection
reflect_task = Task(
description="""对上一步 Task 的输出做反思:
1. 找出 3 个最明显的质量问题(事实 / 逻辑 / 表达)
2. 给出具体的修改建议
3. 如果质量问题太多,返回 'NEEDS_REWRITE' 让 writer 重写
4. 如果质量可以接受,返回 'OK' 并附上修改后的版本""",
expected_output="要么 'NEEDS_REWRITE',要么 'OK' + 改后版本",
agent=reviewer,
context=[write_task], # 关键:拿上一步 writer 的输出当 context
)
crew = Crew(
agents=[writer, reviewer],
tasks=[write_task, reflect_task],
process=Process.sequential,
verbose=True,
)
if __name__ == "__main__":
result = crew.kickoff()
print("\n===== 最终输出 =====")
print(result.raw)
关键点:
Task.context=[write_task]是 CrewAI 反思的精髓——前一个 Task 的输出自动作为后一个 Task 的 context。- Reviewer Agent 的
goal直接定义成「找问题、给建议」,让 LLM 天然进入批评者角色。 - 通过
Process.sequential+ 多个反思 Task 可以自然形成多轮反思链(写→批→再批→再写)。
五、看起来很完美,但这 5 个坑你可能踩过
坑 1:反思循环死循环,token 烧光
现象:Agent 反思 20 轮还没收敛,一次跑了几十块。
根因:没设 max_rounds 硬上限,evaluator 又用 LLM 判定,LLM 自己都可能前后矛盾。
解法:硬编码 attempt >= 3 或 max_rounds=3 兜底;evaluator 优先用确定性规则(测试用例、断言、SQL 报错信息)。
坑 2:反思把自己反思崩了(Reflection Collapse)
现象:第二轮、第三轮反思把正确的内容「反思」成错的,越反思越差。
根因:反思 prompt 没说「保留原文中正确的部分,只改错的」。
解法:反思 prompt 显式约束「只指出问题,不要改写正确部分」;或者只在第一轮反思时改,后续轮只追加补充。
坑 3:反思用同一份 LLM,永远挑不出自己的 bug
现象:让 GPT-4 反思 GPT-4 的输出,10 次有 8 次通过质检,但人一看还是有错。
根因:同一模型的「生成-批评」分布是高度相关的,模型抓不到自己没意识到的错误。
解法:用更强的模型当 evaluator(用 GPT-5 评 GPT-4-mini),或者多模型交叉评审(GPT-4 评 + Claude 评 + 规则评,三者多数通过)。
坑 4:反思 prompt 太啰嗦,把 context 撑爆
现象:每轮反思写 500 字,第三轮 actor 的 prompt 已经 2k 字。
根因:没控制反思文本长度,没做历史反思的压缩/裁剪。
解法:反思 prompt 限字数(3 句话以内);历史反思超过 3 条做滑动窗口 + 摘要压缩,具体压缩策略可以参考我之前写的 Context 压缩策略(偷个懒引用一下)。
坑 5:反思结果没落盘,下次同任务还是从零开始
现象:用户问同样的问题,Agent 还是犯同样的错。
根因:反思只活在这次 session 的 memory 里,没存到长期记忆。
解法:把反思结果持久化到 vector store 或关系数据库,下次同任务前先 similarity_search 拉历史反思塞进 prompt。
六、生产环境的 4 个考量
1. 反思的"成本/收益"控制
不是所有任务都值得反思。简单分类任务(“这段文本是中文还是英文”)上反思 = 浪费 token。建议:
- 简单任务(一次通过率 > 95%):不开反思
- 中等任务(一次通过率 60%-90%):开 1-2 轮反思
- 复杂任务(一次通过率 < 60%):开 3 轮反思 + 多模型交叉评审
2. 反思的可观测性
生产里要 trace 每轮反思的 prompt、response、判定理由。LangGraph 用 LangSmith,AutoGen 用 OpenTelemetry,CrewAI 用内置的 verbose=True + 自定义 callback。没有 trace 的反思,出问题你连定位都定位不了。
3. 反思的"非对称成本"
反思 = 至少多调 1 次 LLM = token ×2。生产里要做批量化反思(多个任务一起反思、共享 evaluator prompt)和缓存化反思(相同问题、相同失败模式的反思结果可以缓存复用)。
4. 反思的"冷启动"
新任务、新领域,反思记忆是空的,第一轮几乎等于无反思。可以用 few-shot prompt:「历史上类似任务的反思长这样……」来 warm up。
七、3 大框架反思能力对比
| 维度 | LangGraph | AutoGen | CrewAI |
|---|---|---|---|
| 反思抽象方式 | StateGraph 节点 + conditional edge | 多 Agent 互发消息 | Task 链 + context 传递 |
| 循环控制 | 显式 add_conditional_edges + 状态字段 |
协调器 Agent 控制 rounds | 多个反思 Task 串联 |
| 死循环兜底 | recursion_limit + 状态 attempt 计数 |
协调器 max_rounds |
Process 类型决定(sequential 不会死循环) |
| 反思落盘 | 配合 Checkpointer(MemorySaver / Postgres) | 自定义 Store 接数据库 | 配合 Memory 工具或外部存储 |
| 学习曲线 | 中(要懂图状态机) | 高(异步消息、Agent 生命周期) | 低(贴近人类分工直觉) |
| 多模型交叉评审 | 需要自己写多 evaluator 节点 | 原生支持多 Agent 互评 | 需要定义多个 Reviewer Agent |
| 生产成熟度 | ⭐⭐⭐⭐⭐(LangSmith 全套) | ⭐⭐⭐⭐(v0.4 异步架构稳) | ⭐⭐⭐(CrewAI AMP 商业版成熟) |
一句话总结:要精细控制反思循环 → 选 LangGraph;要多 Agent 互怼 / 学术场景 → 选 AutoGen;要快速搭多 Agent 团队、靠角色分工 → 选 CrewAI。
八、面试官还会追问什么
追问 1:Reflexion 和 Self-Refine 的本质区别是什么?
→ 回答方向:Reflexion 是「跨 episode 的反思」(同一个任务重做 N 轮,每轮之间用反思衔接),核心是记忆;Self-Refine 是「同 episode 内 的迭代」(同一份输出反复 refinement N 轮直到收敛),核心是同输入多次改稿。Reflexion 适合「做错了要重做」的任务,Self-Refine 适合「做对了但要更好」的任务。
追问 2:反思机制会不会让 Agent 越来越慢?怎么权衡?
→ 回答方向:会。反思 = 额外 LLM 调用,token 翻倍甚至 N 倍。生产里要做「任务分级 + 反思分级」:简单任务关反思,复杂任务开反思;同时把反思结果缓存化(同类失败模式复用反思文本)。另外反思 prompt 要强约束字数(3 句话以内),别让反思把 context 撑爆。
追问 3:怎么判断 Agent 的反思真的有效,而不是「自我感动」?
→ 回答方向:跑消融实验。同一批任务,跑「无反思」「反思 N 轮」「用更强模型反思」三组,对比:(1) 一次通过率,(2) token 成本,(3) 人工抽检质量。如果反思组通过率显著高于无反思组、且 token 成本可以接受,才是真的有效。只看通过率不看成本 = 自我感动。
追问 4:反思机制能替代 fine-tuning 吗?
→ 回答方向:不能完全替代。反思是「prompt 层面的临时记忆」,每次 session 都要重新生成;fine-tuning 是「权重层面的永久记忆」,一次训练永久生效。反思解决的是「这一次怎么改」,fine-tuning 解决的是「从根本上变聪明」。生产里两者配合用:反思处理任务级失败,fine-tuning 处理领域级能力差距。
追问 5:你们生产里反思机制踩过最深的坑是什么?
(这个问题留给你——我相信每个做过 Agent 反思的人都有一个「反思把自己反思崩了」或者「反思一晚上 token 烧了几千块」的故事,欢迎评论区分享。)
九、总结:反思是 Agent 的「错题本」
Agent 反思机制的本质,就是给模型一个机会看到自己错了。
人之所以能从新手变老手,不是因为脑子突然变好了,是因为我们有一本错题本。每次翻看错题,都相当于给大脑做一次「以人为鉴」。Agent 反思机制在做的事一模一样——让 LLM 在不更新权重的情况下,靠「看自己的失败记录」来变聪明。
如果你读完这篇,应该能:
- 区分 Self-Reflection / Self-Critique / Reflexion / Self-Refine 这 4 个概念
- 画出 Reflexion 4 步闭环(Actor → Environment → Evaluator → Self-Reflection → Memory)
- 在 LangGraph / AutoGen / CrewAI 三个框架里挑一个把反思跑起来
- 避开 反思死循环 / 反思崩塌 / 反思撑爆 context 这 3 个最常见的坑
如果这篇文章让你少烧几千块 token,欢迎点赞 + 在看。
最后一个问题留给你:你团队里的 Agent,现在是靠「prompt 调优」硬扛质量,还是已经在用反思机制了?如果还没有,下周可以先挑一个高频失败任务试一周,看 token 成本下降多少。
更多推荐


所有评论(0)