Agent 反思机制 Reflexion基于自评结果重新触发检索在传统的单向流水线 RAG 系统Naive RAG中执行流是一个**“开弓没有回头箭”的单向死胡同**用户提问 - 检索一次 - 拼入 Prompt - 大模型生成答案 - 交付用户。如果第一步检索发生偏离或者召回的文档根本不足以回答当前问题大模型在没有后路的情况下只能硬着头皮使用有限且不相关的上下文开始“强行脑补”最终向用户交付一段充满事实硬伤的幻觉内容。真正成熟的人类专家在面对复杂问题时绝不会盲目作答当我们翻开参考书发现书里的资料完全没有解答核心疑问时我们会立刻停下来进行自我反思Self-Reflection“我查的关键词不对我应该换个更专业的术语重新查另一章”在智能体Agent领域将这一人类元认知能力工程化落地的核心架构正是Reflexion反思与自我纠错机制 / Self-RAG。如何基于 LangGraph 设计一个具备答案自评、证据充分性校验、自动触发二次检索与多轮迭代自愈的高可靠 AgentReflexion 架构的自适应闭环流转模型[ 用户复杂提问 ] | v [ 节点 1: 初始证据检索 (Initial Retrieve) ] | v [ 节点 2: 草稿生成 (Draft Generation) ] | v ----------------------- 节点 3: 自评反思评估器 (Self-Reflection Evaluator) ----------------------- | 使用轻量评估 Prompt 进行三维打分: | | 1. 证据充分性 (Is_Grounded): 答案中的论据是否完全被检索文档支持(True/False) | | 2. 答案有用性 (Is_Helpful): 是否直接、完整地回答了用户的核心诉求(True/False) | | 3. 改进反思日志 (Reflection_Feedback): 缺少关于 2026 年新版费率的说明需针对费率关键词重搜| ---------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------- | (自评通过: 证据充分且有用) | (自评未通过: 存在缺陷且重试 3次) v v [ 节点 4: 正式交付最终答案 (END) ] [ 节点 5: 基于反思日志重写 Query (Query Rewrite) ] | v [ 返回节点 1: 重新触发针对性二次检索! ]Python LangGraph 生产级 Reflexion 智能体实现from typing import TypedDict, List, Optional from pydantic import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END # 1. 结构化反思评估契约 class ReflectionGrade(BaseModel): is_grounded: bool Field(description生成的答案是否 100% 被检索到的参考文档所支持无凭空捏造) is_helpful: bool Field(description答案是否完整解答了用户的核心提问) critique: str Field(description详细的反思批评意见指出缺失的事实或需要补充的盲区) improved_search_query: Optional[str] Field(description如果需要重新检索给出更精确优化的新查询语句) # 2. Agent 全局状态 class ReflexionAgentState(TypedDict): question: str current_search_query: str retrieved_docs: List[str] draft_answer: str reflection: Optional[ReflectionGrade] retry_count: int # 3. 节点逻辑 async def retrieve_node(state: ReflexionAgentState): query state.get(current_search_query) or state[question] print(f [Retrieve Node] 正在根据 Query 执行检索: {query}) # 模拟检索第一次检索可能只搜到粗略信息第二次精准搜到核心数据 if 费率 in query: docs [【权威资料】2026 年新版企业支付接口费率为单笔 0.28%无月度保底消费。] else: docs [【基础资料】企业支付接口提供高并发结算能力接入流程请参考开发文档。] return {retrieved_docs: docs} async def generate_draft_node(state: ReflexionAgentState): docs_text \n.join(state[retrieved_docs]) prompt ChatPromptTemplate.from_template(基于以下资料回答问题\n{docs}\n\n问题{question}\n回答) llm ChatOpenAI(modelgpt-4o-mini, temperature0) chain prompt | llm resp await chain.ainvoke({docs: docs_text, question: state[question]}) print(f [Generate Node] 生成草稿: {resp.content}) return {draft_answer: resp.content} async def self_reflect_node(state: ReflexionAgentState): 核心反思节点扮演严苛的审核法官 docs_text \n.join(state[retrieved_docs]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个极其严苛的事实核查员。请对比用户问题、参考资料与生成的回答草稿输出客观的反思评估结果。), (user, 【用户问题】{question}\n【参考资料】{docs}\n【待审回答】{draft}) ]) llm ChatOpenAI(modelgpt-4o-mini, temperature0).with_structured_output(ReflectionGrade) chain prompt | llm grade: ReflectionGrade await chain.ainvoke({ question: state[question], docs: docs_text, draft: state[draft_answer] }) print(f [Reflect Node] 自评结果: 有据{grade.is_grounded}, 有用{grade.is_helpful} | 批评: {grade.critique}) return { reflection: grade, retry_count: state.get(retry_count, 0) 1, current_search_query: grade.improved_search_query or state[question] } # 4. 组装具备自愈能力的状态图 def build_reflexion_graph(): workflow StateGraph(ReflexionAgentState) workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_draft_node) workflow.add_node(reflect, self_reflect_node) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, generate) workflow.add_edge(generate, reflect) # 核心定义条件反思路由 def evaluate_loop_condition(state: ReflexionAgentState) - str: reflection: ReflectionGrade state[reflection] retries state[retry_count] # 如果通过自评或者重试达到上限防死循环硬熔断则结束交付 if (reflection.is_grounded and reflection.is_helpful) or retries 2: return end # 否则触发回路带着反思出的新 Query 重新检索 print(f 触发自愈回路准备发起第 {retries 1} 轮针对性检索...) return re_retrieve workflow.add_conditional_edges( reflect, evaluate_loop_condition, { end: END, re_retrieve: retrieve } ) return workflow.compile()生产治理核心总结Reflexion 机制是大模型摆脱“盲目自信与幻觉”的最高阶范式。自评提示词必须极度苛刻明确要求大模型在没有完全覆盖用户疑问时坚决打出False重试上限硬熔断Max Retries必须显式设置retries 2硬熔断防止大模型陷入自我怀疑的无限死循环小模型做初审大模型做终审反思打分节点可选用耗时低的小模型将单轮自评耗时压缩在 150ms 以内兼顾高准确率与低延迟。