基于LangGraph构建自我改进AI Agent：从执行者到学习者的范式跃迁

1. 项目概述从“执行者”到“学习者”的Agent范式跃迁最近在AI Agent的圈子里一个概念被反复提及Self-Improving Agent自我改进智能体。这不再是那种只会机械执行预设流程的“脚本小子”而是一个能通过与环境交互、分析自身表现、并主动优化策略的“学习者”。Hermes Agent作为这个领域一个颇具代表性的开源项目其核心思想就是构建这样一个学习闭环。而LangGraph这个由LangChain推出的新框架以其强大的有状态、多步骤工作流编排能力为我们实现这种复杂的设计模式提供了绝佳的“脚手架”。今天我们就来彻底拆解这个闭环看看如何用LangGraph将“自我改进”从一个美好的愿景落地为可运行、可复现的代码设计模式。简单来说这个模式的目标是让Agent具备“吃一堑长一智”的能力。想象一下你部署了一个客服Agent来处理用户咨询。传统的Agent可能按照固定的知识库匹配答案遇到未知问题就卡壳。而一个具备Self-Improving能力的Agent在首次回答失败或收到用户负面反馈后能够自动触发一个分析流程回顾对话历史、定位知识缺口、生成新的知识条目或优化回答策略并将这个新知识整合到自己的“大脑”中确保下次遇到同类问题能回答得更好。这背后就是一个由“执行”、“评估”、“反思”、“学习”、“更新”等节点构成的有向图这正是LangGraph所擅长的。2. 核心设计模式剖析Self-Improving Agent的循环架构要实现自我改进关键在于设计一个能够持续运转的闭环系统。这个闭环不是简单的“if-else”而是一个有状态、可分支、能循环的智能工作流。我们可以将其抽象为一个经典的设计模式我称之为“执行-评估-优化”循环模式。这个模式是构建Self-Improving Agent的蓝图。2.1 模式的核心组件与数据流这个模式通常包含以下几个核心组件它们通过特定的数据流连接起来形成一个闭环主执行节点负责处理核心任务例如回答用户问题、编写代码、分析数据等。它接收用户输入和当前Agent的“知识状态”并产生输出。评估与反思节点这是学习闭环的“触发器”和“诊断器”。它负责评估主执行节点的输出质量。评估标准可以是预设的规则如代码是否有语法错误、外部反馈如用户的“踩”或评分、或是通过另一个LLM进行质量评审。如果评估结果不达标例如置信度低、用户反馈负面该节点会生成一个“反思信号”并触发反思流程。反思与根因分析节点一旦被触发这个节点会深入分析失败或表现不佳的原因。它可能会审查完整的交互历史、内部思考过程并试图定位问题所在是知识不足是推理逻辑有误还是对用户意图理解偏差这个节点的输出是一个具体的“改进建议”或“学习目标”。学习与知识更新节点根据反思节点输出的改进建议执行具体的学习动作。例如调用搜索引擎查询缺失信息、让LLM基于新信息生成一段解释、向向量数据库插入新的知识片段、或者微调一个提示模板。这个节点的核心作用是生产出能够弥补Agent能力短板的“新知识”。状态管理与记忆这是整个闭环的“粘合剂”。LangGraph的State对象是关键。它需要持久化记录当前任务上下文、历史交互记录、评估结果、反思内容、以及最重要的——Agent不断演进的知识库或策略参数。状态在节点间传递并随着循环迭代而更新。数据流遵循“执行 - 条件评估 - 反思 - 学习 - 更新状态 - 再次执行”的路径。LangGraph的StateGraph允许我们轻松定义这种带条件边conditional edges的循环。2.2 与经典Agent架构的对比为了更清晰地理解我们将其与经典的ReActReasoning and Acting或Plan-and-Execute架构进行对比特性经典Agent (如 ReAct)Self-Improving Agent (本模式)核心目标完成单次任务在多次任务中持续提升性能工作流线性或带工具调用的链式带反馈循环的图状工作流状态变化任务完成后状态重置状态在循环中累积和演化知识增长触发机制用户输入或计划步骤基于对自身输出的评估结果输出产物任务答案任务答案 内部知识/策略的增量更新可以看到Self-Improving Agent引入了一个“元认知”层使其不仅能处理任务还能处理“如何处理任务更好”这个元任务。注意设计这个循环时必须设置合理的终止条件防止陷入“无限反思-学习”的死循环。例如限制单次对话中的最大改进循环次数或当评估分数超过某个阈值时直接退出循环返回最终答案。3. 基于LangGraph的实现拆解构建可运行的循环图理论说再多不如一行代码。我们现在就用LangGraph来搭建这个“执行-评估-优化”循环。我将以构建一个“能越用越聪明的技术问答助手”为例分步拆解实现细节。3.1 定义状态State闭环的记忆核心在LangGraph中状态是一个贯穿始终的可变对象。我们需要精心设计它以承载闭环中的所有必要信息。from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 用户输入与对话历史 messages: Annotated[List, add_messages] # LangGraph内置的消息历史管理 user_query: str # 当前用户问题 # 主执行节点的输出 initial_answer: str # Agent的初次回答 final_answer: str # 经过改进循环后的最终答案 # 评估与反思相关 needs_improvement: bool # 评估结果是否需要改进 critique: str # 具体的批评意见或反思内容 learning_goal: str # 本次需要学习的具体目标 # 学习与知识更新相关 new_knowledge: str # 学习到的新知识片段 knowledge_base: List[str] # 模拟的Agent知识库实际可能是向量数据库 # 控制流 improvement_cycles: int # 记录已进行的改进循环次数用于防止无限循环这里的关键是needs_improvement和improvement_cycles它们将作为控制图流程走向的条件判断依据。knowledge_base作为状态的一部分使得学习成果能够被持久化并在后续交互中使用。3.2 实现节点Node函数闭环的各个功能模块接下来我们实现闭环中的每一个功能节点。每个节点都是一个接收并更新AgentState的函数。节点1主执行节点这个节点根据当前知识库和用户问题生成初步答案。def primary_agent_node(state: AgentState) - AgentState: 基于现有知识回答用户问题。 query state[“user_query”] kb_context “\n”.join(state[“knowledge_base”][-5:]) # 使用最近的一些知识 # 构建提示词让LLM结合知识库回答 prompt f 你是一个技术问答助手。请根据以下已知信息回答问题。 如果已知信息不足以回答问题请诚实地说你不知道不要编造答案。 已知信息 {kb_context} 用户问题{query} 请给出专业、准确的回答 # 这里调用LLM (例如通过ChatOpenAI) # 为简化示例我们模拟一个可能不完善的回答 messages state[“messages”] # ... 调用LLM将prompt加入messages并获取响应 ... # simulated_response llm.invoke(...) simulated_response “在Python中你可以使用asyncio.sleep()来实现异步等待。这是标准库的一部分。” # 假设回答 state[“initial_answer”] simulated_response state[“final_answer”] simulated_response # 初始时最终答案即初次答案 return state节点2评估与反思触发节点这个节点评估答案质量并决定是否进入改进循环。def evaluation_node(state: AgentState) - AgentState: 评估答案质量并生成反思。 answer state[“initial_answer”] query state[“user_query”] # 评估策略1基于规则的简单检查例如是否包含“我不知道” if “我不知道” in answer or “无法回答” in answer: state[“needs_improvement”] True state[“critique”] “Agent的回答表明它缺乏回答此问题的知识。” return state # 评估策略2调用另一个LLM进行质量评审更强大 critique_prompt f 请扮演一个严格的评审员评审以下问答对的质量。 问题{query} 回答{answer} 请从准确性、完整性、清晰度三个方面评审。 如果回答存在事实错误、信息缺失或表述模糊请指出具体问题。 如果回答基本合格请说“合格”。 评审意见 # simulated_critique llm.invoke(critique_prompt) simulated_critique “回答基本正确但不够深入。例如没有提到在异步函数中必须使用await asyncio.sleep()也没有提及与time.sleep()的区别。这对于初学者可能造成困惑。” if “合格” not in simulated_critique: state[“needs_improvement”] True state[“critique”] simulated_critique else: state[“needs_improvement”] False return state节点3根因分析与学习目标生成节点如果评估需要改进这个节点会分析具体原因并制定学习目标。def reflection_node(state: AgentState) - AgentState: 根据评审意见生成具体的学习目标。 critique state[“critique”] query state[“user_query”] reflection_prompt f 基于以下评审意见请分析助理回答不足的根本原因并提炼出一个具体、可执行的学习目标。 原问题{query} 评审意见{critique} 请用一句话概括助理需要学习或补充的知识点是什么 # simulated_goal llm.invoke(reflection_prompt) simulated_goal “需要深入学习asyncio.sleep()的具体用法、与time.sleep()的对比、以及在异步函数中await关键字的重要性。” state[“learning_goal”] simulated_goal return state节点4主动学习与知识获取节点根据学习目标主动获取新知识。def learning_node(state: AgentState) - AgentState: 根据学习目标获取新知识。 goal state[“learning_goal”] # 学习策略可以调用搜索引擎API、查询本地文档、或让LLM自行生成解释 learning_prompt f 你是一个知识整理助手。请针对以下学习目标生成一段简明、准确、易于理解的知识摘要。 学习目标{goal} 知识摘要 # simulated_knowledge llm.invoke(learning_prompt) simulated_knowledge “asyncio.sleep(delay)是Python asyncio库中用于挂起当前协程的函数参数delay是以秒为单位的浮点数。关键点1. 必须在异步函数(async def)内使用。2. 必须配合await关键字(await asyncio.sleep(1))。3. 它只会挂起当前协程而不阻塞整个线程这是与同步time.sleep()最本质的区别后者会阻塞整个线程。4. 常用于模拟I/O等待或控制协程执行节奏。” state[“new_knowledge”] simulated_knowledge return state节点5知识整合与状态更新节点将学到的新知识整合到Agent的长期记忆中并准备重新回答。def update_knowledge_node(state: AgentState) - AgentState: 将新知识整合到知识库并更新状态以准备重新执行。 new_knowledge state[“new_knowledge”] # 将新知识加入知识库 if new_knowledge and new_knowledge not in state[“knowledge_base”]: state[“knowledge_base”].append(new_knowledge) print(f“[知识更新] 新知识已入库{new_knowledge[:50]}...) # 增加改进循环计数 state[“improvement_cycles”] 1 # 清空临时字段为下一次主执行做准备 state[“initial_answer”] “” state[“critique”] “” state[“learning_goal”] “” state[“new_knowledge”] “” # 注意此时不直接设置 needs_improvement由下一次评估决定 return state3.3 编排图Graph与条件边让循环转起来现在我们用LangGraph的StateGraph把这些节点组装起来并定义它们之间的流转逻辑。from langgraph.graph import StateGraph, END # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“primary_agent”, primary_agent_node) workflow.add_node(“evaluate”, evaluation_node) workflow.add_node(“reflect”, reflection_node) workflow.add_node(“learn”, learning_node) workflow.add_node(“update”, update_knowledge_node) # 设置入口点 workflow.set_entry_point(“primary_agent”) # 定义边流程 workflow.add_edge(“primary_agent”, “evaluate”) # 执行后必然评估 # 关键条件边。根据评估结果决定下一步 def decide_after_evaluation(state: AgentState) - str: if state[“needs_improvement”] and state[“improvement_cycles”] 3: # 防止无限循环 return “reflect” # 需要改进进入反思学习流程 else: return END # 无需改进或循环次数太多结束 workflow.add_conditional_edges( “evaluate”, decide_after_evaluation, { “reflect”: “reflect”, END: END } ) # 定义改进循环内的线性流程 workflow.add_edge(“reflect”, “learn”) workflow.add_edge(“learn”, “update”) workflow.add_edge(“update”, “primary_agent”) # 关键学习后回到主执行节点重新回答 # 编译图 app workflow.compile()这个图结构清晰地定义了我们的闭环主执行 - 评估 - (如果需要改进) - 反思 - 学习 - 更新知识 - 回到主执行。add_conditional_edges是实现分支逻辑的核心而add_edge(“update”, “primary_agent”)则构成了循环的回路。3.4 运行与迭代现在我们可以初始化一个状态并运行这个自我改进的Agent了。# 初始化状态 initial_state AgentState( messages[], user_query“Python里怎么让程序等待几秒”, initial_answer“”, final_answer“”, needs_improvementFalse, critique“”, learning_goal“”, new_knowledge“”, knowledge_base[“Python是一种编程语言。”], # 初始知识很薄弱 improvement_cycles0 ) # 运行图 final_state app.invoke(initial_state) print(f“最终答案{final_state[‘final_answer’]}”) print(f“改进循环次数{final_state[‘improvement_cycles’]}”) print(f“当前知识库条目数{len(final_state[‘knowledge_base’])}”)在这个模拟中由于初始知识库薄弱Agent的第一次回答很可能触发评估不通过。于是它会进入反思-学习循环生成关于asyncio.sleep的详细知识并存入知识库然后重新执行主Agent节点。第二次执行时Agent就能利用刚学到的知识给出更优质的回答。最终评估节点可能会认为答案合格流程结束。4. 关键问题与实战调优让学习闭环真正高效可靠实现一个能跑通的闭环只是第一步。要让Self-Improving Agent在实际中可靠、高效地工作会遇到一系列挑战。下面是我在实践和复现类似项目时总结的关键问题和调优经验。4.1 评估机制的可靠性与成本控制评估节点是整个闭环的“守门人”它的质量直接决定了学习循环是否被正确触发。问题1评估不准导致“瞎学”或“不学”表现LLM作为评估者可能不稳定同样的答案在不同时间评估结果可能不同规则评估又过于死板。解决策略多维度投票不要只依赖一次LLM调用做判断。可以让评估节点提出多个评估问题如“答案准确吗”“答案完整吗”“表述清晰吗”综合多个结果做决策。置信度过滤让LLM在评估时输出一个置信度分数。只有低置信度或明确指出的错误才触发学习。可以设置一个阈值如0.7。关键信息验证对于事实性回答可以集成一个简单的“事实核查”工具比如从答案中提取实体或陈述快速搜索知识库或可信源进行验证。问题2评估成本过高表现每次主Agent回答后都要调用一次甚至多次LLM进行评估Token消耗翻倍。解决策略抽样评估并非每次交互都触发完整评估。可以按一定概率如20%进行或者当用户提供明确反馈点赞/点踩时才触发。轻量级评估模型使用小尺寸、低成本的模型如小型开源模型进行初步评估只有小模型不确定时才动用大模型进行深度反思。缓存评估结果对相似的问题和答案可以缓存评估结果避免重复计算。4.2 学习内容的质量与知识管理学什么、怎么学、学了怎么存是决定Agent能力增长质量的关键。问题3学习内容冗余或低质表现Agent反复学习相似内容或学到的知识碎片化、难以利用。解决策略学习目标去重在reflection_node中将生成的learning_goal与知识库中已有主题进行相似度匹配可用嵌入向量如果高度相似则合并学习目标或跳过本次学习。知识结构化不要让new_knowledge只是一段文本。设计一个结构化的知识模式例如主题、要点、示例代码、参考链接。learning_node可以按照这个模式来生成知识卡片。知识摘要与压缩定期例如每积累10条新知识运行一个“知识整理”后台任务让LLM对相关主题的知识进行去重、合并和摘要形成更系统化的知识条目。问题4知识检索与利用效率低表现知识库越来越大但主Agent节点在回答时无法快速准确地检索到相关知识。解决策略向量化检索这是标配。将知识库条目和用户查询都转化为向量使用向量数据库如Chroma, Weaviate, Pinecone进行相似度检索。在primary_agent_node中用user_query去检索最相关的N条知识作为上下文。元数据过滤为每条知识添加元数据如来源、创建时间、置信度、适用领域。检索时结合语义相似度和元数据过滤提升精度。检索后重排序初步检索出多条知识后可以用一个轻量级交叉编码器模型或让LLM快速判断哪几条与当前问题最相关进行重排序只将Top-1或Top-2喂给主Agent。4.3 循环控制与稳定性保障一个不受控的自我改进循环是危险的可能导致资源耗尽或行为失控。问题5无限循环或振荡表现Agent在“评估-改进”循环中出不来或者在两个都不完美的答案间来回切换。解决策略硬性次数限制如我们代码中的improvement_cycles设置一个绝对上限如3次。改进收敛判断比较本次改进后的答案与上一次答案的差异。如果差异小于某个阈值或者评估分数不再显著提升则主动终止循环。人工审核介入当循环达到一定次数或触发了某些高风险关键词时将学习目标和生成的新知识挂起等待人工审核批准后再入库。问题6状态管理与错误恢复表现长对话中状态复杂某个节点出错可能导致整个图状态混乱。解决策略状态快照与回滚在进入可能出错的节点尤其是调用外部工具的学习节点前可以对关键状态进行快照。如果节点执行失败能够回滚到上一个稳定状态并记录错误日志而不是让错误状态污染后续流程。子图Subgraph封装将“反思-学习-更新”这个改进循环封装成一个独立的子图。这样主图结构更清晰而且子图内部的错误可以被隔离和处理不影响主对话流的进行。LangGraph对子图有很好的支持。完善的日志每个节点的输入、输出、关键决策如needs_improvement的值都应被详细记录。这对于调试复杂的工作流至关重要。5. 进阶模式与扩展思考超越单一任务的学习基础的“执行-评估-优化”循环主要针对单任务表现的提升。我们可以在此基础上探索更高级的Self-Improving模式。5.1 多技能协同与元技能学习一个复杂的任务往往需要多个技能或工具的协同。Agent可以学习如何更好地组合和调用这些技能。模式在reflection_node中不仅分析“答案对不对”还分析“解决路径是否最优”。例如Agent在解决一个数据可视化问题时先尝试了用A库但效果不好。反思节点可以分析出“对于时间序列数据使用B库的C函数比A库更合适”并将“问题类型 - 推荐工具/技能”的映射关系作为元知识存入知识库。下次遇到类似问题Agent能直接选择更优的路径。5.2 从交互历史中进行批量离线学习实时在线学习虽然及时但可能受到单次交互噪音的影响。可以引入一个离线学习管道。模式定期例如每天将一段时间内的所有交互历史包括用户查询、Agent回答、用户反馈、最终修正后的答案导出。用一个独立的、更强大的学习流程可能涉及微调小模型、生成高质量的提示模板、提炼新的工具使用规范对这些数据进行分析和学习。学习成果再以“模型参数更新”或“高质量知识包”的形式批量注入到在线Agent的知识库或配置中。这相当于Agent的“定期进修”。5.3 基于人类反馈的强化学习集成将人类反馈如评分、排序、修正直接作为强化学习的奖励信号是让Agent对齐人类偏好的高级方式。模式将整个LangGraph工作流视为一个策略。用户的正面反馈如“有帮助”的点击作为正奖励负面反馈作为负奖励。我们可以记录下产生这些反馈的完整状态和动作序列即Agent的思考过程和工具调用。虽然完整的RL训练负载很重但可以简化例如当收到强烈负面反馈时不仅触发当前对话的反思学习还可以主动在知识库中标记或降权与导致错误答案相关的知识条目甚至触发一个更广泛的、针对相似知识点的审查和学习任务。实现Self-Improving Agent是一个系统工程LangGraph提供了优雅的编排框架但核心的挑战在于评估、学习和控制逻辑的设计。从一个小而精的闭环开始比如先让Agent学会在回答“我不知道”后去主动搜索一次再逐步增加评估维度和学习深度是更稳妥的实践路径。这个模式的价值在于它让AI应用从静态的、部署即定型的工具开始向动态的、能够伴随使用而成长的伙伴演进。