基于 LangGraph 的 Agent 循环图编排:条件边与记忆持久化
2026/9/17 2:29:10 网站建设 项目流程

基于 LangGraph 的 Agent 循环图编排:条件边与记忆持久化

在构建能够处理复杂长任务的大模型 Agent 系统时,传统的链式调用模式(如早期的 LangChain Sequential Chain)很快就会撞上能力的“天花板”:
真实世界的任务流绝不是线性的,而是充斥着循环重试(Cyclic Loops)、动态分支路由(Conditional Branching)以及跨轮次状态记忆更新(State Persistence)

  • 比如:Agent 生成了一段 Python 代码后,必须调用解释器执行;如果执行报错,Agent 必须带着错误堆栈循环回退到代码生成节点重新修复,直到单测 100% 通过才能流转至下一步。

LangChain 团队推出的LangGraph,将 Agent 工作流抽象为一个强类型的有向循环图(Directed Cyclic Graph, DCG)

通过将图节点(Nodes)绑定为纯函数式 Agent 推理单元、将条件边(Conditional Edges)绑定为确定性路由守卫、并结合持久化检查点(Checkpointer),我们能够构建出具备企业级循环控制与时间旅行调试能力的复杂 Agent 编排系统。

LangGraph 循环状态图架构拓扑

[用户输入 Task Prompt] │ ▼ ┌──────────────────┐ │ node: plan │ ◄────────────────────────┐ └────────┬─────────┘ │ │ │ ▼ │ ┌──────────────────┐ │ │ node: generate │ ◄──────────┐ │ └────────┬─────────┘ │ │ │ │ (重试修复) │ (重试规划) ▼ │ │ ┌──────────────────┐ │ │ │ node: execute │ ───────────┴─────────────┤ └────────┬─────────┘ │ │ (条件边路由判定) │ ┌────────────────────┼────────────────────┐ │ ▼ (执行报错 && <3次) ▼ (严重架构错误) ▼ (单测全部通过) │ [路由回 generate] [路由回 plan] └───────┬───────────┘ ▼ ┌──────────────────┐ │ node: summarize │ ──> [最终交付] └──────────────────┘

核心实现:基于 TypeScript 的 LangGraph 状态图与条件边

import { StateGraph, END, START, MemorySaver } from '@langchain/langgraph'; import { BaseMessage, HumanMessage, AIMessage } from '@langchain/core/messages'; // 1. 定义全图强类型共享状态接口 (State Annotation) export interface AgentGraphState { messages: BaseMessage[]; codeDraft: string; testError: string | null; retryCount: number; isCompleted: boolean; } // 2. 初始化 LangGraph 状态图定义 const workflow = new StateGraph<AgentGraphState>({ channels: { messages: { value: (prev: BaseMessage[], next: BaseMessage[]) => prev.concat(next), default: () => [], }, codeDraft: { value: (prev: string, next: string) => next ?? prev, default: () => '', }, testError: { value: (prev: string | null, next: string | null) => next, default: () => null, }, retryCount: { value: (prev: number, next: number) => next, default: () => 0, }, isCompleted: { value: (prev: boolean, next: boolean) => next, default: () => false, }, }, }); // 3. 定义图节点处理函数 (Nodes) async function generateCodeNode(state: AgentGraphState) { console.log('🤖 [Node: generateCode] 正在生成/修复代码...'); const prompt = state.testError ? `上次运行报错: ${state.testError},请修复代码。` : `请编写代码实现目标。`; // 模拟大模型调用生成代码 const generatedCode = `function calculateTotal() { /* fixed logic */ return 100; }`; return { codeDraft: generatedCode, messages: [new AIMessage(generatedCode)], }; } async function executeTestNode(state: AgentGraphState) { console.log('⚡ [Node: executeTest] 正在沙箱中运行单元测试...'); // 模拟第一次执行失败、第二次执行成功的场景 if (state.retryCount === 0) { return { testError: 'AssertionError: Expected 100 but received 0', retryCount: state.retryCount + 1, isCompleted: false, }; } return { testError: null, retryCount: state.retryCount + 1, isCompleted: true, }; } // 4. 定义条件边动态路由函数 (Conditional Edge Router) function routeAfterTest(state: AgentGraphState): 'generate' | typeof END { if (state.isCompleted) { console.log('✔ 测试全部通过,流转至结束节点!'); return END; } if (state.retryCount >= 3) { console.warn('🚨 达到最大重试次数上限,强制中断!'); return END; } console.log('🔄 检测到单测失败,条件边触发循环回退至 [generate] 节点!'); return 'generate'; } // 5. 组装图节点与边拓扑 workflow.addNode('generate', generateCodeNode); workflow.addNode('execute', executeTestNode); // 连线拓扑 workflow.addEdge(START, 'generate'); workflow.addEdge('generate', 'execute'); // 核心:添加条件分支循环边 workflow.addConditionalEdges('execute', routeAfterTest, { generate: 'generate', [END]: END, }); // 6. 挂载持久化检查点存储器 (支持状态序列化与多轮断点恢复) const checkpointer = new MemorySaver(); export const appGraph = workflow.compile({ checkpointer });

前端调用与状态流式监听

在前端 Web 客户端中,调用编译后的 LangGraph 工作流并实时监听状态图节点的跃迁事件:

export async function runAgentWorkflow(sessionId: string, userPrompt: string) { const config = { configurable: { thread_id: sessionId } }; // 触发有向图流式执行 const stream = await appGraph.stream( { messages: [new HumanMessage(userPrompt)], retryCount: 0, isCompleted: false, }, config ); for await (const stepOutput of stream) { const [nodeName, output] = Object.entries(stepOutput)[0]; console.log(`📍 状态跃迁至节点: [${nodeName}]`, output); } }

工程落地收益

  1. 确定性的循环与分支控制:彻底摆脱了传统自然语言 Prompt 容易陷入死循环的失控问题,重试上限与路由分支在图结构上被代码强行锁死。
  2. 原生支持断点持久化与时间旅行(Time Travel):通过checkpointer,系统自动保存每一个 Node 执行前后的完整状态快照,支持开发者在前端控制台秒级回滚到任意历史步骤重新调试。
  3. 天然契合可视化编排:LangGraph 的图结构可以直接导出为标准 Mermaid 或 JSON 拓扑,直接在前端画布上渲染出动态发光的执行节点,极大提升了 Agent 系统的可解释性。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询