一、前言本篇约1.2万字阅读时长约30分钟。随着LLM应用越来越复杂单轮对话已经很难满足高质量输出的需求。一个常见的痛点是LLM第一次生成的答案往往不够好——可能遗漏关键信息可能逻辑不够严密可能缺乏引用支撑。那么问题来了怎么让LLM回头看看自己写的答案想想哪里不对然后改一改这就是Reflection模式要解决的事。而Reflexion模式更进一步——不光自己想还去外面查。本文就来拆解这两种模式在LangGraph上的完整实现。一、前言二、Reflection与Reflexion两种反思范式三、Reflection模式源码剖析四、Reflexion模式源码剖析五、两种模式的工程对比六、要点总结二、Reflection与Reflexion两种反思范式Reflection的核心思想很简单让LLM自己批评自己。具体来说它引入了两个角色生成器Generator负责写内容反思器Reflector负责挑毛病生成器写完反思器来批批完生成器再改如此循环。说白了就是让LLM瞎折腾自己的输出——但折腾完确实变好了。Reflexion在Reflection的基础上多了一层外部工具调用。它不光靠LLM自己想还会真正去搜索外部信息来验证和补充。那么Reflexion相比Reflection多了什么结构化输出通过bind_tools强制LLM输出结构化的AnswerQuestion对象工具调用引入搜索工具执行search_queries拿回真实信息情景记忆Episodic Memory每一轮的反思和搜索结果都累积到消息列表中后续轮次能看到前面的教训引用机制修订后的答案必须带数字引用[1] https://...简单说Reflection是闭卷自我修改Reflexion是开卷查证修改。两者最核心的差异有三个输出方式Reflection是纯文本Reflexion是结构化bind_tools Pydantic外部信息Reflection纯靠LLM脑子里的知识Reflexion会调TavilySearchResults去外面查终止条件Reflection靠哨兵标记Reflexion用双闸门更详细的对比放在第五章。先把两个模式的源码逐行拆开。三、Reflection模式源码剖析先看整体拓扑再逐层拆解。Reflection拓扑自我精炼的闭环整条链路只有三个节点generate、reflect、加一个条件路由should_continue。简单到不能再简单。1、State设计一条消息链承载全部记忆State只维护一个字段——messages列表class State(TypedDict): messages: Annotated[list, add_messages]add_messages是LangGraph内置的reducer作用是每轮generate和reflect产出的新消息会追加到列表尾部而不是覆盖。这意味着整个对话历史——请求、初稿、批评、修订稿、再批评——全部累积在一条messages链里。这就是情景记忆的载体。2、Prompt设计生成器与反思器各司其职生成器和反思器的Prompt分别长这样generate_prompt ChatPromptTemplate.from_messages([ (system, You are an essay assistant tasked with writing excellent 5-paragraph essays. Generate the best essay possible for the users request. If the user provides critique, respond with a revised version.), MessagesPlaceholder(variable_namemessages),])reflection_prompt ChatPromptTemplate.from_messages([ (system, You are a teacher grading an essay submission. Generate critique and recommendations for the users submission. If the submission needs NO further revision, append the exact token NO_REVISIONS_NEEDED at the very end of your critique.), MessagesPlaceholder(variable_namemessages),])注意一个关键设计反思器的Prompt里埋了一个哨兵标记NO_REVISIONS_NEEDED。当反思器认为不用再改了它会在批评末尾附带这个标记。这是一种确定性的终止信号——不靠关键词猜测靠精确匹配。3、generate节点朴实无华的链调用generate节点就是一个prompt | model的链调用包了一层工厂函数def make_generation_node(generate) - Callable[[State], dict]: def generation_node(state: State) - dict: return {messages: [generate.invoke(state[messages])]} return generation_node工厂函数的目的是让generate这个Runnable可注入——测试时可以传一个假的Runnable进来不需要真正调用LLM。这是有必要的否则单测没法跑。4、reflect节点角色翻转是精髓reflect节点干了一件看起来很诡异的事——翻转消息角色def make_reflection_node(reflect) - Callable[[State], dict]: def reflection_node(state: State) - dict: cls_map {ai: HumanMessage, human: AIMessage} # 保留原始请求翻转之后的所有消息角色 translated [state[messages][0]] [ cls_map[m.type](contentm.content) for m in state[messages][1:] ] res reflect.invoke(translated) return {messages: [HumanMessage(contentres.content)]} return reflection_node为什么要翻转因为反思器的Prompt设定是你是老师在批改学生提交的作文。而实际上之前的作文是AI写的AIMessage。所以需要把AI写的作文伪装成HumanMessage学生提交让反思器以为自己在批改学生的作业。反过来反思器输出的批评又被包装成HumanMessage返回给生成器。生成器看到的是用户给了反馈让我改。角色翻转一场精心编排的双簧 生成器AI写作文 → 翻转为 HumanMessage 反思器AI批改 → 输出包装为 HumanMessage 生成器收到反馈 → 以为用户要求修改5、should_continue什么时候停终止逻辑只有两条规则干净利落_SATISFIED_MARKER NO_REVISIONS_NEEDEDdef should_continue(state: State) - str: messages state[messages] if len(messages) 6: # 兜底消息数超限 return END last messages[-1] if isinstance(last, HumanMessage) and _SATISFIED_MARKER in last.content: return END # 满意了停 return generate # 不满意回去继续改个人认为把检查放在reflect之后而不是generate之后是一个很务实的改进。因为这样可以直接读到最新的批评不需要去messages[-2]偷看前一条消息。6、build_app把零件拼起来最后把所有节点和边组装成一张完整的图def build_app(*, modelNone, generateNone, reflectNone, checkpointerNone): if generate is None or reflect is None: if model is None: model make_default_model() if generate is None: generate generate_prompt | model if reflect is None: reflect reflection_prompt | model workflow StateGraph(State) workflow.add_node(generate, make_generation_node(generate)) workflow.add_node(reflect, make_reflection_node(reflect)) workflow.add_edge(START, generate) workflow.add_edge(generate, reflect) workflow.add_conditional_edges(reflect, should_continue, [generate, END]) return workflow.compile(checkpointercheckpointer or InMemorySaver())generate和reflect都是可选注入的。传了就用你传的没传就从model推导。这种设计让整个图的每个角色都可替换——这是工程上靠谱的做法。四、Reflexion模式源码剖析Reflexion的拓扑明显更复杂多了execute_tools节点Reflexion拓扑带工具调用的反思闭环关键区别在于draft和revise是两个不同的角色绑定不同的tool schema中间还插了一个execute_tools来执行搜索。1、Schema设计用Pydantic锁死输出格式Reflexion用Pydantic模型定义了三种结构化输出强制LLM按schema吐数据class Reflection(BaseModel): The actors self-critique. missing: str Field(descriptionCritique of what is missing.) superfluous: str Field(descriptionCritique of what is superfluous.)class AnswerQuestion(BaseModel): Answer the question: an answer, a reflection, and follow-up queries. answer: str Field(description~250 word detailed answer to the question.) reflection: Reflection Field(descriptionYour reflection on the initial answer.) search_queries: list[str] Field( description1-3 search queries for researching improvements. )class ReviseAnswer(AnswerQuestion): Revise the original answer, with citations motivating the changes. references: list[str] Field(descriptionCitations motivating your updated answer.)关键点在于ReviseAnswer继承了AnswerQuestion并多加一个references字段。这意味着修订版必须带引用而初始版不需要。这些schema通过bind_tools绑定到LLM上强制LLM以function call的形式输出结构化数据而不是自由文本。2、Prompt设计一个模板复用两个角色Reflexion的Prompt同时服务于初始回答和修订回答actor_prompt_template ChatPromptTemplate.from_messages([ (system, You are expert researcher.\n Current time: {time}\n\n 1. {first_instruction}\n 2. Reflect and critique your answer. Be severe to maximize improvement.\n 3. Recommend search queries to research information and improve your answer.), MessagesPlaceholder(variable_namemessages), (user, \n\nsystemReflect on the users original question and the actions taken thus far. Respond using the {function_name} function./reminder),]).partial(timelambda: datetime.datetime.now().isoformat())first_instruction和function_name是可变参数初始回答时first_instructionProvide a detailed ~250 word answer.function_nameAnswerQuestion修订回答时first_instructionrevise_instructions要求加引用function_nameReviseAnswer一个模板复用两个角色靠参数区分这是务实的做法。3、ResponderWithRetries带着质检员的流水线未经审视的人生不值得过。——苏格拉底这是Reflexion模式里最值得品的设计。它不是简单地调用LLM而是调用-验证-失败重试三步走就像流水线末端站了一个质检员不合格就打回去返工class ResponderWithRetries: def __init__(self, runnable, validator): self.runnable runnable self.validator validator def respond(self, state: State) - dict: messages state[messages] response None for attempt in range(3): # 最多重试3次 response self.runnable.invoke( {messages: messages}, {tags: [fattempt:{attempt}]} ) try: self.validator.invoke(response) # PydanticToolsParser 验证 return {messages: [response]} except ValidationError as e: # 把错误信息塞回去当ToolMessage让LLM看着错误改 tool_call_id ( response.tool_calls[0][id] if response.tool_calls else str(attempt) ) messages messages [ response, ToolMessage( content( f{repr(e)}\n\nPay close attention to the function schema.\n\nRespond by fixing all validation errors. ), tool_call_idtool_call_id, ), ] return {messages: [response]}执行流程拆解调用runnable拿到LLM响应用PydanticToolsParser验证输出是否符合schema验证通过直接返回验证失败把错误信息塞回去当ToolMessage让LLM看着错误自己改最多重试3次3次都不行就返回最后一次的结果个人认为这种把验证错误反馈给模型的设计比直接抛异常要靠谱得多。LLM看到ValidationError的具体描述后往往能在下一轮修正格式问题。注意一个原版notebook的Bug修复原代码写的是state [response, ToolMessage(...)]这是dict list操作直接抛TypeError。正确写法是messages [response, ToolMessage(...)]——追加到列表而不是用state做加法。4、execute_tools跑腿小弟execute_tools负责把LLM给出的搜索词拿去跑工具节点负责执行LLM给出的search_queriesdef run_queries(search_queries: list[str], **kwargs): if os.getenv(TAVILY_API_KEY): try: from langchain_community.tools.tavily_search import TavilySearchResults tavily TavilySearchResults(max_results5) return tavily.batch([{query: q} for q in search_queries]) except ImportError: pass # 可选依赖缺失走离线 return [_offline_lookup(q) for q in search_queries]def build_tool_node() - ToolNode: return ToolNode([ StructuredTool.from_function(run_queries, nameAnswerQuestion.__name__), StructuredTool.from_function(run_queries, nameReviseAnswer.__name__), ])设计上有两个细节值得注意run_queries同时注册了两个名字——AnswerQuestion和ReviseAnswer。因为draft和revise两个节点的LLM输出的tool call名称不同但实际执行的是同一个搜索函数没有配置TAVILY_API_KEY时返回离线占位文本。这让整个loop不需要网络也能跑起来方便演示和调试5、event_loop两道水坝谁先满谁开闸流水不腐户枢不蠹。但流太多了也得拦住。Reflexion的终止条件是最复杂也最精妙的部分。它用了两道独立的闸门就像两道水坝——哪道先溢出就触发泄洪终止MAX_QUERIES 12 # 所有轮次search_queries总和上限MAX_REVISIONS 5 # 修订次数上限不含初始draftdef event_loop(state: State) - str: if ( _get_total_search_queries(state) MAX_QUERIES or _get_revision_count(state) MAX_REVISIONS ): return END return execute_tools两道闸门是OR关系谁先超就谁触发。为什么需要两道因为LLM的行为不可预测如果LLM每轮都吐出很长的search_queries列表查询总数会先爆——MAX_QUERIES闸门触发如果LLM每轮只吐一两个查询修订次数会先爆——MAX_REVISIONS闸门触发无论哪种情况循环都不会无限跑下去以下是一个假设场景假设LLM每轮吐3条查询展示闸门触发过程双闸门终止示意假设LLM每轮吐3条查询轮次 累计查询数 累计修订数 触发闸门 1 3 0 - 2 6 1 - 3 9 2 - 4 12 3 - 5 15 4 MAX_QUERIES 12 → END如果只有一道闸门要么拦不住查得多的要么拦不住改得多的。两道齐上整得挺靠谱。_get_revision_count的计算逻辑也需要留意它统计的是带tool_calls的AIMessage数量减1。因为第一条AIMessage是初始draft不计入修订次数。6、build_app组装带工具的反思图def build_app(*, modelNone, first_responderNone, revisorNone, tool_nodeNone, checkpointerNone): if first_responder is None or revisor is None: if model is None: model make_default_model() if first_responder is None: initial_chain actor_prompt_template.partial( first_instructionProvide a detailed ~250 word answer., function_nameAnswerQuestion.__name__, ) | model.bind_tools(tools[AnswerQuestion]) first_responder ResponderWithRetries( runnableinitial_chain, validatorPydanticToolsParser(tools[AnswerQuestion]), ) if revisor is None: revision_chain actor_prompt_template.partial( first_instructionrevise_instructions, function_nameReviseAnswer.__name__, ) | model.bind_tools(tools[ReviseAnswer]) revisor ResponderWithRetries( runnablerevision_chain, validatorPydanticToolsParser(tools[ReviseAnswer]), ) tool_node tool_node or build_tool_node() workflow StateGraph(State) workflow.add_node(draft, first_responder.respond) workflow.add_node(execute_tools, tool_node) workflow.add_node(revise, revisor.respond) workflow.add_edge(START, draft) workflow.add_edge(draft, execute_tools) workflow.add_edge(execute_tools, revise) workflow.add_conditional_edges(revise, event_loop, [execute_tools, END]) return workflow.compile(checkpointercheckpointer or InMemorySaver())和Reflection的build_app对比区别一目了然多了execute_tools节点draft和revise绑定了不同的tool schema两者都包在ResponderWithRetries里带验证重试条件路由从should_continue换成了event_loop双闸门五、两种模式的工程对比把两种模式放在一起从工程视角做一次全面对比对比维度ReflectionReflexion核心思路自我批评循环自我批评 外部搜索验证LLM输出纯文本str结构化bind_tools Pydantic外部依赖仅需LLMLLM 搜索工具Tavily消息角色翻转AI→Human保持原始角色终止策略哨兵标记 硬上限双闸门MAX_QUERIESORMAX_REVISIONS容错无重试ValidationError重试3轮输出质量依赖LLM自身能力有引用、有搜索验证Provider兼容性高纯文本即可中需支持function calling适用场景创意写作、文本润色知识问答、研究分析选型建议很简单如果你的场景不需要查外部资料Reflection就够了还省token如果需要事实核查和引用支撑上Reflexion。六、要点总结Reflection的核心是生成-批评-修订的纯文本循环通过角色翻转让LLM扮演师生关系用哨兵标记NO_REVISIONS_NEEDED控制终止。Reflexion在Reflection基础上增加了结构化输出bind_tools、工具调用Tavily搜索和情景记忆实现开卷反思。ResponderWithRetries的验证重试机制通过把ValidationError反馈给LLM来修复格式问题最多重试3次。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】