用LangGraph构建自我反思型AI Agent,通过规划、执行、反思和重试,显著提升大模型输出的准确性和可靠性,减少幻觉,满足专业领域对AI的高标准要求。
原文标题:用 LangGraph 搭建会自我反思的 AI Agent
原文作者:图灵编辑部
冷月清谈:
Reflection Agent 并非一种新的模型,而是一种结构化的循环架构,旨在让推理过程更具系统性。其核心机制是一个由规划(Plan)、执行(Act)、反思(Reflect)和条件性重试(Retry)组成的有限状态循环。简单来说,它模仿了人类“先规划、再输出、最后批判性审查并改进”的思维模式。其最大优势在于将答案生成与评估明确拆分为两个独立连续的步骤,从而主动减少推理错误。
文章进一步通过 LangGraph 与 LangChain 调用 OpenAI GPT 模型,详细演示了从零开始构建 Reflection Agent 的过程。这包括定义 Agent 状态类、初始化模型接口、以及分别实现 Plan、Act、Reflect 和 Retry 各个节点的功能逻辑。其中,Plan 节点负责制定解答策略;Act 节点根据计划生成初步答案;Reflect 节点则对答案进行批判性审查,识别潜在的遗漏、过度简化或错误;最后,Retry 节点会根据反思结果对答案进行迭代改进。通过路由逻辑,Agent 会根据反思结果决定是结束输出还是进入重试循环(最多两次),直到达到可接受的质量标准。
这套架构不仅提升了模型输出的质量,还将其思想多样性、同行评审机制和条件检查点编码进了一个完整的思考闭环。它代表了提示工程与完全自主智能体之间的实用中点,在现实部署中能显著提升AI系统的性能,尤其适用于重视正确性、清晰度和渐进式输出的场景,同时有效减少幻觉输出并保持相对可控。
怜星夜思:
2、Reflection Agent 主要解决内部逻辑和输出质量问题,那它能解决所有 AI 安全问题(如偏见、滥用)吗?和 RAG、RLHF 等技术如何协同?
3、除了文章提到的工程、法律分析、科研这些专业领域,这种自我反思机制还能在哪些日常场景或者更复杂的 AI 系统中发挥作用?它的通用性如何?
原文内容
随着大模型能力的不断提升,各行各业对它们的期待也在变化。人们不再满足于模型只给出答案,而是希望它能提供有理有据、可验证的推理过程。
在工程、法律分析、科研、产品开发等专业领域,差不多就行早已不行。一个基于大模型的智能体,能像人类一样对自己的输出进行复查、思考、修订,这已经不是锦上添花,而是变成了基本要求。
问题在于:
-
大多数基于提示词(prompt-based)的输出是一次性的
-
它们缺少一个验证循环
-
没有自我评估机制,就容易出现幻觉、逻辑不一致或语境不匹配的问题
为了解决这些问题,我们将使用 LangGraph 来构建一个基于图结构的智能体,让它能够通过有方向的推理过程实现自我反思、改进并最终完善输出。
什么是 Reflection Agent?
Reflection Agent 并不是一种新的模型,而是一种让推理过程更有结构化循环的架构。你可以把它想象成同时进行的三场对话:
-
先规划——思考该如何解决问题;
-
再输出——生成回答;
-
最后反思——对输出进行批判性审查,并在需要时进行改进。
从设计上看,这个 Agent 就像一个由三个主要步骤组成的有限状态循环:
Plan(计划) → Act(执行) → Reflect(反思)如果需要,还会触发一个条件性的 Retry(重试)。
它最大的优势在于:将生成答案和评估答案拆分为两个独立、连续的步骤,从而主动减少推理错误。
就像写论文一样——先写初稿,再交给编辑修改。
环境设置
我们将通过 LangGraph 使用 LangChain 来调用 OpenAI 的 GPT 模型。
首先安装:
pip install --upgrade langgraph langchain langchain-openai
然后设置你的 API Key,可以通过环境变量方式
export OPENAI_API_KEY="your-key"
或者在脚本中手动配置:
import osos.environ["OPENAI_API_KEY"] = "your-key"
代码讲解:从零开始构建
新建一个文件,命名为 reflection_agent.py
下面是我们一步步搭建逻辑的过程。
定义状态类(State Class)
首先,在 LangGraph 的语境下定义 Agent 的状态结构:
from typing import TypedDict, List, Optional
class AgentState(TypedDict):
input_question: str
history: List
reflection: Optional[str]
answer: Optional[str]
attempts: int
这个模型用于在图的各个节点之间保存 Agent 的思考和行为决策数据。
初始化模型接口:
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
然后初始化模型:
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
Plan 节点
def plan_step(state: AgentState) -> AgentState:
question = state['input_question']
history = state.get('history', [])
plan_prompt = f"You're assigned to plan an explanation for:\n\n{question}\n\nBreak your thoughts into steps or bullets."
messages = history + [HumanMessage(content=plan_prompt)]
response = llm.invoke(messages)
updated_history = history + [
HumanMessage(content=plan_prompt),
AIMessage(content=response.content)
]
return {
**state,
"history": updated_history
}
在这个阶段,Agent 还没有开始真正解答问题,它只是在思考应该如何去解答。
Act 节点
根据上一步的计划,Agent 开始生成答案:
def act_step(state: AgentState) -> AgentState:
act_prompt = "Use your planning above to produce the best possible answer."
messages = state["history"] + [HumanMessage(content=act_prompt)]
response = llm.invoke(messages)
updated_history = state["history"] + [
HumanMessage(content=act_prompt),
AIMessage(content=response.content)
]
return {
**state,
"history": updated_history,
"answer": response.content
}
这一步不会对之前的有效性作任何假设,它只是按照计划执行并生成输出。
Reflect 节点
这是最关键的一步:在这里进行反思。答案是否可靠?有没有遗漏?
def reflect_step(state: AgentState) -> AgentState:
question = state["input_question"]
answer = state.get("answer", "")
reflection_prompt = (
f"Review this answer to the following question:\n"
f"Q: {question}\nA: {answer}\n\n"
f"Are there missing considerations, oversimplifications, false claims, or ambiguous logic? If so, describe them."
)
messages = [
SystemMessage(content="You're reviewing for clarity, logic, and completeness."),
HumanMessage(content=reflection_prompt)
]
response = llm.invoke(messages)
return {
**state,
"reflection": response.content
}
反思本身并不会修正答案——它只是系统地发现问题。
重试触发(路由逻辑)
根据反思结果,我们决定是重试还是结束。
def should_react(state: AgentState) -> str:
reflection = state.get("reflection", "")
attempts = state.get("attempts", 0)
if attempts >= 2:
return "end"
if "no flaws" in reflection.lower() or "satisfactory" in reflection.lower():
return "end"
return "retry"
Retry 节点
在需要时,Agent 会根据反思结果重新调用自身,对答案进行改进:
def retry_step(state: AgentState) -> AgentState:
reflection = state["reflection"]
repair_prompt = f"You reflected that:\n{reflection}\n\nUpdate your answer accordingly."
updated_history = state["history"] + [HumanMessage(content=repair_prompt)]
response = llm.invoke(updated_history)
return {
**state,
"history": updated_history + [AIMessage(content=response.content)],
"answer": response.content,
"attempts": state.get("attempts", 0) + 1
}
组装 LangGraph 工作流
from langgraph.graph import StateGraph, END
graph_builder = StateGraph(AgentState)
graph_builder.add_node("plan", plan_step)
graph_builder.add_node("act", act_step)
graph_builder.add_node("reflect", reflect_step)
graph_builder.add_node("retry", retry_step)
graph_builder.set_entry_point("plan")
graph_builder.add_edge("plan", "act")
graph_builder.add_edge("act", "reflect")
graph_builder.add_conditional_edges("reflect", should_react, {
"end": END,
"retry": "retry"
})
graph_builder.add_edge("retry", "reflect")
graph = graph_builder.compile()
运行代理
if __name__ == "__main__":
entry_question = "Explain the second law of thermodynamics in layman's terms."
state = {
"input_question": entry_question,
"history": [],
"reflection": None,
"answer": None,
"attempts": 0
}
result = graph.invoke(state)
print("Final Answer:\n")
print(result["answer"])
print("\nReflection:\n")
print(result["reflection"])
输出示例(带注释)
典型的结果结构如下:
Final Answer:
The second law of thermodynamics says that energy naturally spreads out. For example, if you put a hot spoon in cool water, the heat spreads into the water, and the total disorder (entropy) increases.
Reflection:
The explanation is mostly solid and simplified correctly. Consider clarifying entropy’s role as a tendency toward statistical variance or chaos. Quality acceptable but improves with an analogy.
可以注意到,它提供了:
-
易懂的答案
-
技术性审查
-
可选的重试
总结
我们在这里构建的,并不仅仅是一个简单的 GPT 包装器,而是一种能模拟完整思考闭环的架构:
-
策略规划
-
执行生成
-
编辑式反思
-
答案优化
从技术角度来看,我们将多样性、同行评审和条件检查点都编码进了思考循环中。在任何重视正确性、清晰度或渐进式输出的实际部署环境中,这种架构都能显著提升性能。
结语
Reflection Agent 是提示工程与完全自主智能体之间的实用中点。它在减少幻觉输出的同时,仍保持相对安全和可控。
去实现它,去扩展它——这是当下你可以为 AI 系统引入的最具实用价值的架构模式之一。
想真正从零开始,亲手搭建属于自己的 AI Agent?不仅仅是能读懂代码,而是要理解背后的设计逻辑、掌握调试技巧、学会优化策略——这一切,在图灵的【AI Agent 实战营】都能系统学到。我们不依赖任何框架,帮你从理论到实战都搞懂,让你的智能体真正可靠、高效地运行起来。
