基于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 模式的核心组件与数据流

这个模式通常包含以下几个核心组件，它们通过特定的数据流连接起来，形成一个闭环：

1. 主执行节点：负责处理核心任务，例如回答用户问题、编写代码、分析数据等。它接收用户输入和当前Agent的“知识状态”，并产生输出。
2. 评估与反思节点：这是学习闭环的“触发器”和“诊断器”。它负责评估主执行节点的输出质量。评估标准可以是预设的规则（如代码是否有语法错误）、外部反馈（如用户的“踩”或评分）、或是通过另一个LLM进行质量评审。如果评估结果不达标（例如，置信度低、用户反馈负面），该节点会生成一个“反思信号”，并触发反思流程。
3. 反思与根因分析节点：一旦被触发，这个节点会深入分析失败或表现不佳的原因。它可能会审查完整的交互历史、内部思考过程，并试图定位问题所在：是知识不足？是推理逻辑有误？还是对用户意图理解偏差？这个节点的输出是一个具体的“改进建议”或“学习目标”。
4. 学习与知识更新节点：根据反思节点输出的改进建议，执行具体的学习动作。例如，调用搜索引擎查询缺失信息、让LLM基于新信息生成一段解释、向向量数据库插入新的知识片段、或者微调一个提示模板。这个节点的核心作用是生产出能够弥补Agent能力短板的“新知识”。
5. 状态管理与记忆：这是整个闭环的“粘合剂”。LangGraph的State对象是关键。它需要持久化记录：当前任务上下文、历史交互记录、评估结果、反思内容、以及最重要的——Agent不断演进的知识库或策略参数。状态在节点间传递，并随着循环迭代而更新。

数据流遵循“执行 -> （条件评估） -> 反思 -> 学习 -> 更新状态 -> 再次执行”的路径。LangGraph的StateGraph允许我们轻松定义这种带条件边（conditional edges）的循环。

2.2 与经典Agent架构的对比

为了更清晰地理解，我们将其与经典的ReAct（Reasoning and Acting）或Plan-and-Execute架构进行对比：

可以看到，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 state

3.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_improvement=False, critique=“”, learning_goal=“”, new_knowledge=“”, knowledge_base=[“Python是一种编程语言。”], # 初始知识很薄弱 improvement_cycles=0 ) # 运行图 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应用从静态的、部署即定型的工具，开始向动态的、能够伴随使用而成长的伙伴演进。