LangGraph赋能:构建会“自我反思”的AI智能体

用LangGraph构建自我反思型AI Agent,通过规划、执行、反思和重试,显著提升大模型输出的准确性和可靠性,减少幻觉,满足专业领域对AI的高标准要求。

原文标题:用 LangGraph 搭建会自我反思的 AI Agent

原文作者:图灵编辑部

冷月清谈:

随着大模型能力日益增强,业界对其期望已从单纯给出答案,转变为提供有理有据、可验证的推理过程。尤其在工程、法律、科研等专业领域,模型输出的准确性和可靠性成为基本要求。面对当前大多数基于提示词(prompt-based)模型缺乏验证循环和自我评估机制,易产生幻觉、逻辑不一致等问题,本文提出并详细介绍了如何使用 LangGraph 构建一个“Reflection Agent”。

Reflection Agent 并非一种新的模型,而是一种结构化的循环架构,旨在让推理过程更具系统性。其核心机制是一个由规划(Plan)、执行(Act)、反思(Reflect)和条件性重试(Retry)组成的有限状态循环。简单来说,它模仿了人类“先规划、再输出、最后批判性审查并改进”的思维模式。其最大优势在于将答案生成与评估明确拆分为两个独立连续的步骤,从而主动减少推理错误。

文章进一步通过 LangGraph 与 LangChain 调用 OpenAI GPT 模型,详细演示了从零开始构建 Reflection Agent 的过程。这包括定义 Agent 状态类、初始化模型接口、以及分别实现 Plan、Act、Reflect 和 Retry 各个节点的功能逻辑。其中,Plan 节点负责制定解答策略;Act 节点根据计划生成初步答案;Reflect 节点则对答案进行批判性审查,识别潜在的遗漏、过度简化或错误;最后,Retry 节点会根据反思结果对答案进行迭代改进。通过路由逻辑,Agent 会根据反思结果决定是结束输出还是进入重试循环(最多两次),直到达到可接受的质量标准。

这套架构不仅提升了模型输出的质量,还将其思想多样性、同行评审机制和条件检查点编码进了一个完整的思考闭环。它代表了提示工程与完全自主智能体之间的实用中点,在现实部署中能显著提升AI系统的性能,尤其适用于重视正确性、清晰度和渐进式输出的场景,同时有效减少幻觉输出并保持相对可控。

怜星夜思:

1、这种P-A-R-R循环听起来很酷,但实际跑起来会不会很慢,或者调用API成本翻倍?有啥办法可以优化吗?
2、Reflection Agent 主要解决内部逻辑和输出质量问题,那它能解决所有 AI 安全问题(如偏见、滥用)吗?和 RAG、RLHF 等技术如何协同?
3、除了文章提到的工程、法律分析、科研这些专业领域,这种自我反思机制还能在哪些日常场景或者更复杂的 AI 系统中发挥作用?它的通用性如何?

原文内容

随着大模型能力的不断提升,各行各业对它们的期待也在变化。人们不再满足于模型只给出答案,而是希望它能提供有理有据、可验证的推理过程。

在工程、法律分析、科研、产品开发等专业领域,差不多就行早已不行。一个基于大模型的智能体,能像人类一样对自己的输出进行复查、思考、修订,这已经不是锦上添花,而是变成了基本要求。

问题在于:

  • 大多数基于提示词(prompt-based)的输出是一次性的

  • 它们缺少一个验证循环

  • 没有自我评估机制,就容易出现幻觉、逻辑不一致或语境不匹配的问题

为了解决这些问题,我们将使用 LangGraph 来构建一个基于图结构的智能体,让它能够通过有方向的推理过程实现自我反思、改进并最终完善输出。

什么是 Reflection Agent

Reflection Agent 并不是一种新的模型,而是一种让推理过程更有结构化循环的架构。你可以把它想象成同时进行的三场对话:

  1. 先规划——思考该如何解决问题;

  2. 再输出——生成回答;

  3. 最后反思——对输出进行批判性审查,并在需要时进行改进。

从设计上看,这个 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 节点

Agent 为输入问题制定一个简短的解答策略:
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 实战营】都能系统学到。我们不依赖任何框架,帮你从理论到实战都搞懂,让你的智能体真正可靠、高效地运行起来。

从项目管理和成本控制角度看,P-A-R-R模式的成本提升是显而易见的。每次额外的LLM调用都是实打实的token消耗和时间损失。对于企业级应用,我们通常会这么平衡:首先,通过 A/B 测试来量化收益,看多反思几轮带来的质量提升是否值得投入。如果质量是核心指标,比如法律文件分析或医疗诊断辅助,那这种投入是值得的。其次,可以引入阈值管理和智能调度。例如,如果首次输出的置信度很高,可以跳过Reflect;或者仅在特定类型的复杂问题上触发完整的反思流程。此外,并行化某些可以独立运行的检查任务,或者利用专有小模型(SLM)负责特定的反思子任务,也能有效降低整体延迟。最重要的是,需要明确业务场景对‘足够好’的定义,而不是追求无限的反思与优化。

哎呀,这不就是让 AI 变得像我们人类一样,做事之前先想清楚,做完之后再反省一下哪里做得不够好的嘛!太通用了!:partying_face::partying_face::partying_face: 日常生活里,我觉得它简直就是懒人福音啊!比如,你让 AI 写个旅行计划,它先给你个初稿,然后自己反思一下“这个景点安排是不是太赶了?”“是不是忘了考虑交通时长?”要是能像这样自动改进,那我的拖延症都能被治好一半!再比如,游戏里的 NPC AI,如果能自我反思它的战术“刚才偷袭没成功,是不是应该换个角度?”那游戏体验岂不是更上一层楼?还有咱们程序员写代码,如果 AI 能帮我先写一段,然后自己反思一下“这变量名起得好low”“这里是不是有安全漏洞”,那加班时间估计能少一半!总之,只要需要思考和改进的地方,这货就能派上用场,简直就是 AI 界的“万能胶”啊!

问得好!Reflection Agent 就像是个勤奋的“学习委员”,它能帮你检查作业(输出)有没有低级错误,逻辑对不对。但它解决不了你班级同学(训练数据)本身‘偏科’(偏见)或者‘耍小聪明’(被滥用)的问题。所以,要让 AI 真正‘品学兼优’,光有学习委员还不够。RAG 就像是给 AI 配了个学霸助教,遇到不会的可以去查资料,补充外部知识;RLHF 更像是请来了班主任和家长,他们不断地告诉 AI 啥是对的,啥是不该做的,教它做人(啊不,做 AI)的道理。所以,这些技术不是非此即彼,而是各司其职,组合起来才能构建一个更‘乖’更‘聪明’的 AI 嘛!就像拼乐高,单个模块再厉害,也得搭配起来才能搭出大别墅!

关于您提出的P-A-R-R循环在效率和成本上的担忧,确实是这种多阶段推理架构需要面对的核心挑战。从计算复杂性角度看,引入Plan、Reflect甚至Retry步骤必然会增加推理的延迟(latency)和计算资源消耗,因为每次LLM调用都需要额外的时间和token。优化策略可以从几个方面着手:一是优化每个子步骤的Prompt设计,使其更精炼高效地引导模型;二是考虑使用不同大小或能力的模型来处理不同阶段的任务,例如,Plan阶段可以用更轻量的模型快速生成粗略策略,而Act和Reflect阶段再使用更强大的模型;三是引入缓存机制,对于重复性高或有确定性输出的中间步骤进行缓存。此外,结合人类在环(Human-in-the-Loop)可以有选择性地触发反思或重试,在某些非关键场景下跳过部分验证环节,以平衡性能与准确性。

我觉得 Reflection Agent 的通用性确实很高,因为它并非针对某个特定领域的模型,而是一种通用的“思维框架”。在日常应用中,这种框架可以显著提升用户体验。比如在智能客服中,它不仅能给出答案,还能反思答案是否完全解决了用户问题,甚至预测用户可能追问的下一个问题并提前给出补充信息。在内容创作领域,如新闻稿、营销文案生成,Agent 可以自我评判内容的吸引力、语调是否符合目标受众,甚至检查是否存在事实性错误或敏感词汇。对于更复杂的 AI 系统,比如涉及多模态输入的机器人,它可以在接收到视觉、听觉等信息后,对自己的感知和决策进行反思,从而提高交互的准确性和自主性。本质上,任何需要从错误中学习、从经验中成长、并追求更高质量输出的 AI 应用,都能够通过引入 Reflection Agent 的架构来获得显著的效能提升,真正从‘工具’向‘智能体’迈进。

Reflection Agent 的自我反思机制具有很强的通用性,其“规划-执行-反思-重试”循环可以抽象为任何需要高可靠性、逻辑严谨性和迭代改进的智能系统。在专业领域外,它在日常场景中也有巨大潜力:例如,在智能写作助手中,可以用于优化邮件、报告的措辞、逻辑结构或观点支撑;在教育领域,个性化辅导系统可以反思学生答案,提供更具针对性的反馈,甚至自我修正教学策略。在更复杂的 AI 系统中,例如:AI 医生诊断辅助系统,反思机制可以对诊断结论、治疗方案进行多维度交叉验证;在自动驾驶决策系统中,对驾驶策略的即时反思和修正能提高安全性;在金融风险评估中,能对复杂模型输出的风险报告进行逻辑校验和潜在漏洞检查。这种通用性在于,所有包含“决策-行动-反馈-调整”循环的智能任务,都能受益于这种内化的评估与改进机制,从而提升系统的鲁棒性和决策的准确性。

哈哈,你说的太对了!我一看到这个循环就想到,这不就是让我家的GPT干三份活甚至更多嘛!以前一次就出结果,现在得‘思考人生’好几趟,这不是妥妥地给我API账单double甚至triple的机会吗?!:money_with_wings::money_with_wings::money_with_wings: 优化嘛,我觉得是不是可以搞个‘经验库’,模型反思过的问题就存起来,下次遇到类似的,直接‘抄作业’不就好了?再不然咱们设置个‘心情值’,模型心情好就少跑几圈反思,心情不好就多反思几次(开玩笑哈哈哈)。更实际点,就是得看任务紧急程度,不那么重要的就少反思几次,急着要的就让它好好反思,不然出了岔子更麻烦。反正,钱和效率,总得做取舍嘛!

要回答“Reflection Agent 能否解决所有 AI 安全问题”这个问题,答案肯定是不能。Reflection Agent 主要的贡献在于通过自我批判和迭代过程来提升输出的逻辑一致性、准确性与完善性,这确实能有效减少如“幻觉”这类与模型内部推理相关的错误。然而,AI 安全问题是一个更宏大的范畴,包括但不限于数据偏见、伦理道德、隐私保护以及恶意滥用等。这些问题往往根植于训练数据、模型设计理念甚至是应用环境。Reflection Agent 自身并不能直接解决数据偏见(如果模型训练数据本身就带有偏见,它反思的结果可能依然带有偏见)或恶意攻击。在这种情况下,它需要与检索增强生成(RAG)、强化学习与人类反馈(RLHF)、红队测试(Red Teaming)以及数据脱敏等多种 AI 安全技术协同工作,形成一个多层防护体系。例如,RAG 可以为 Reflection Agent 提供更权威、实时的外部知识,提升其反思的依据;RLHF 则可以从人类的价值观和偏好层面,校准模型的行为和输出,帮助其理解并避免生成带有偏见或不当的内容。