2026/8/22 5:15:39
FEATURED · 精选文章 
1. 从LangChain到LangGraph为什么我们需要“可中断”的AI工作流如果你在过去一两年里折腾过AI应用开发尤其是基于大语言模型LLM构建一些自动化流程那么“LangChain”这个名字你一定不陌生。它像是一套乐高积木把提示词模板、记忆、工具调用这些组件串起来让我们能相对容易地搭建起一个AI驱动的对话机器人或者文档分析工具。我早期很多原型项目都是基于LangChain快速搭起来的它确实极大地降低了入门门槛。但用久了尤其是在尝试构建一些稍微复杂点的、带点“业务流程”味道的应用时痛点就来了状态管理太麻烦流程一旦跑起来就像脱缰的野马想中途干预一下、暂停一下、或者失败了从某个点重试简直是一场噩梦。这就像你写一个简单的脚本A - B - C顺序执行没问题。但现实中的业务逻辑往往是这样的A - (根据结果判断) - 要么走B1要么走B2 - 等待外部输入比如用户确认- 继续执行C - 如果C失败回退到B2重试。用传统的链式Chain思维来硬套代码会迅速变得臃肿不堪各种if-else嵌套在回调函数里状态散落在各个角落。更头疼的是“可恢复性”想象一个处理长篇文档的AI工作流运行到一半服务器重启了你难道要从头再处理一遍吗这就是LangGraph.js切入的战场。它不是一个替代LangChain的全新框架而是一个专门为构建有状态、多步骤、可循环且最重要的是可中断与可恢复的AI智能体Agent或工作流而设计的库。你可以把它理解为在LangChain的“积木块”之上提供了一套可视化、可编程的“流程图”绘制和引擎。它的核心模型是一个有向图节点Node是你的处理函数可以是LLM调用、工具执行、条件判断边Edge定义了节点间的流转逻辑。最关键的是它内置了完整的状态管理和检查点Checkpoint机制这让“暂停-继续”和“失败重试”从概念变成了几行配置就能实现的功能。所以当我们谈论“LangGraph.js可中断可恢复的AI工作流”时我们本质上是在讨论如何将那些脆弱、冗长、不可控的AI自动化脚本升级为健壮、灵活、像传统工作流引擎如Airflow、Camunda一样可靠的应用。这对于构建复杂的AI助手、自动化客服、内容审核流水线、多步骤数据分析Agent等场景是质的飞跃。2. LangGraph.js核心架构拆解图、状态与检查点要玩转可中断的工作流必须吃透LangGraph.js的三个核心概念图Graph、状态State和检查点Checkpoint。这三者构成了其可中断、可恢复能力的基石。2.1 图Graph工作流的骨架LangGraph.js中的“图”就是你的工作流蓝图。它由两种基本元素构成节点Nodes这是实际干活的地方。每个节点是一个异步函数它接收当前整个工作流的“状态”对象执行一些操作比如调用LLM、查询数据库、运行计算然后返回一个对状态的更新。在JavaScript/TypeScript中它通常长这样const myNode async (state: StateType) { // 从state中读取所需数据 const userQuestion state.user_input; // 执行核心逻辑例如调用LLM const llmResponse await chatModel.invoke(请回答${userQuestion}); // 返回一个对象这个对象会用来更新全局state return { assistant_response: llmResponse.content }; };关键点在于节点函数不直接修改传入的state而是返回一个更新“补丁”。这保证了状态变化的可预测性和可序列化。边Edges决定了工作流的走向。分为两种条件边Conditional Edges根据当前状态的值决定下一步去哪个节点。这是实现分支if-else和循环的关键。普通边Regular Edges无条件地从一个节点指向下一个节点。通过组合节点和边你可以定义出非常复杂的流程比如经典的“ReActReasoning Acting代理”循环思考 - 判断是否需要行动 - 是则执行工具 - 合并结果并循环。2.2 状态State工作流的记忆中枢状态是一个中心化的、类型化的对象它随着工作流的执行而演变。LangGraph.js强烈推荐在TypeScript中几乎是必须使用Zod库来定义状态的模式Schema。这不仅是类型安全的需要更是可序列化的前提。import { z } from zod; // 1. 使用Zod定义状态结构 const StateSchema z.object({ // 输入 user_input: z.string(), // 中间过程 ai_thoughts: z.string().optional(), // 代理的“思考”过程 tools_to_call: z.array(z.string()).optional(), // 决定要调用的工具 tool_results: z.array(z.any()).optional(), // 工具执行结果 // 输出 final_answer: z.string().optional(), // 元数据对中断/恢复很有用 current_step: z.string().optional(), // 当前所在节点名 error: z.string().optional(), // 错误信息 }); // 2. 推导出TypeScript类型 type MyWorkflowState z.infertypeof StateSchema;所有节点都读写这个统一的状态对象。当工作流被中断时LangGraph.js需要将整个状态对象可能包含LLM的对话历史、工具调用结果等完整地保存下来。这就要求状态里的所有值都必须是可序列化为JSON的。像函数、DOM元素这类不可序列化的东西绝对不能直接放在状态里。2.3 检查点Checkpoint实现可中断与可恢复的魔法这是LangGraph.js最精髓的部分。检查点机制允许你在工作流执行到任何一个节点后将当前完整的状态State以及工作流所处的上下文比如接下来该执行哪个节点持久化保存起来。它是如何工作的配置检查点存储器你需要提供一个“检查点存储器”CheckpointSaver的实现。LangGraph.js提供了内存存储MemorySaver用于开发测试但对于生产环境你需要将其存储到数据库如Redis、PostgreSQL或文件系统中。import { MemorySaver } from langchain/langgraph; const checkpointSaver new MemorySaver(); // 开发用 // 生产环境可能需要 new RedisSaver(redisClient) 之类的自定义实现在图中启用检查点创建图时将存储器传入。const workflow new StateGraph({ schema: StateSchema, }) .addNode(process_input, processInputNode) .addEdge(start, process_input) .compile({ checkpointer: checkpointSaver, // 关键启用检查点 });执行与中断当你调用workflow.invoke()时可以传入一个config对象其中包含一个configurable字段通常用来指定这次运行的“线程ID”thread_id。const initialInput { user_input: 今天的天气怎么样 }; const config { configurable: { thread_id: user_123_session_1 } }; // 第一次执行可能只执行了几步就被主动暂停或意外中断 const result1 await workflow.invoke(initialInput, config);执行过程中每经过一个节点或你配置的特定节点引擎都会自动调用检查点存储器将当前状态快照保存起来并与这个thread_id关联。恢复执行当需要恢复时你不需要重新构造初始状态。只需使用**相同的thread_id**再次调用invoke甚至可以传入新的输入来更新状态。// 一段时间后恢复执行。注意这里没有传initialInput因为状态已保存。 const result2 await workflow.invoke({}, config); // 从上次中断处继续 // 或者提供新的输入来更新状态后再继续 const result3 await workflow.invoke({ user_input: 那么明天呢 }, config);引擎会根据thread_id从检查点存储器加载最新的状态和进度然后从上次中断的节点之后继续执行。这对于处理长对话、分步任务和错误恢复至关重要。实操心得thread_id的设计非常巧妙。它可以是用户ID、会话ID、或任务ID。这让你能轻松管理同一个工作流的多个并行实例。例如一个客服机器人每个用户对话就是一个独立的、可随时暂停和恢复的工作流线程。3. 构建一个可中断的AI客服工单处理工作流理论说再多不如动手。我们来设计一个稍微贴近实际场景的例子一个AI客服工单自动处理工作流。它的流程是1) 分类用户问题2) 根据分类要么直接回答简单问题要么查询知识库要么在需要人工时暂停并等待3) 最终生成回复。这个流程天然需要“可中断”——因为在“等待人工”节点工作流必须暂停直到客服人员提供了干预信息后才能继续。3.1 定义状态与节点首先定义工作流的状态。我们需要记录用户问题、AI分类结果、查询到的知识、人工干预输入以及最终回复。import { z } from zod; import { StateGraph, Annotation } from langchain/langgraph; import { ChatOpenAI } from langchain/openai; import { MemorySaver } from langchain/langgraph; // 使用Annotation来方便地定义带默认值的状态模式 const StateSchema Annotation.Root({ // 输入 ticketId: z.string().describe(工单唯一ID), customerQuery: z.string().describe(客户原始问题), // 处理过程 classification: z.enum([simple_q, need_kb, need_human]).optional().describe(AI分类结果), kbSearchResult: z.string().optional().describe(知识库查询结果), humanAgentInput: z.string().optional().describe(人工客服的补充输入或指示), // 输出与元数据 aiResponse: z.string().optional().describe(AI生成的回复), finalResponse: z.string().optional().describe(最终发给客户的回复), isResolved: z.boolean().default(false).describe(工单是否已解决), }); type WorkflowState typeof StateSchema.State; // 初始化LLM const llm new ChatOpenAI({ modelName: gpt-4o-mini, temperature: 0, }); // 节点1分类用户问题 const classifyNode async (state: WorkflowState) { console.log([分类节点] 处理工单: ${state.ticketId}); const classifyPrompt 请将以下客户问题分类 - simple_q: 简单问候、感谢或非常基础的问题如“你们上班时间”。 - need_kb: 需要查询产品文档、政策条款才能回答的具体问题。 - need_human: 涉及投诉、退款、复杂技术问题或需要人工判断的情感化问题。 客户问题${state.customerQuery} 只输出分类标签simple_q, need_kb, need_human不要任何其他文字。 ; const classification (await llm.invoke(classifyPrompt)).content.trim() as WorkflowState[classification]; // 返回状态更新补丁 return { classification }; }; // 节点2处理简单问题 const handleSimpleQueryNode async (state: WorkflowState) { console.log([简单问题节点] 直接生成回复); const responsePrompt 客户问了一个简单问题${state.customerQuery}。请以友好、专业的客服口吻直接回答。; const aiResponse (await llm.invoke(responsePrompt)).content; return { aiResponse, finalResponse: aiResponse, isResolved: true }; }; // 节点3查询知识库模拟 const queryKnowledgeBaseNode async (state: WorkflowState) { console.log([知识库节点] 模拟查询); // 这里应该是向量数据库查询等真实操作我们模拟一个结果 await new Promise(resolve setTimeout(resolve, 500)); // 模拟延迟 const mockKbResult 根据知识库文档#2024-001相关问题的标准解决方案是请先尝试重启应用并检查网络连接。如果问题持续请联系技术支持。; return { kbSearchResult: mockKbResult }; }; // 节点4基于知识库生成回复 const generateResponseFromKBNode async (state: WorkflowState) { console.log([生成KB回复节点]); const responsePrompt 基于以下知识库信息回答客户的问题。 客户问题${state.customerQuery} 知识库信息${state.kbSearchResult} 请生成完整、友好的回复。; const aiResponse (await llm.invoke(responsePrompt)).content; return { aiResponse, finalResponse: aiResponse, isResolved: true }; }; // 节点5挂起等待人工干预这是一个“中断点” const waitForHumanNode async (state: WorkflowState) { console.log([等待人工节点] 工单 ${state.ticketId} 已挂起等待客服处理。); // 这个节点本身不修改状态它只是流程中的一个“暂停门”。 // 关键执行到这里检查点已经保存。工作流会停在这里。 // 恢复执行需要外部触发例如调用一个API来更新humanAgentInput状态。 return {}; }; // 节点6处理人工输入并生成最终回复 const processHumanInputNode async (state: WorkflowState) { console.log([处理人工输入节点] 收到客服指示: ${state.humanAgentInput}); if (!state.humanAgentInput) { return { finalResponse: 已转接人工请稍候。, isResolved: false }; } const responsePrompt 客服提供了以下处理意见${state.humanAgentInput}。客户的原问题是${state.customerQuery}。请结合两者生成最终回复给客户。; const finalResponse (await llm.invoke(responsePrompt)).content; return { finalResponse, isResolved: true }; };3.2 组装图并配置条件路由现在我们把节点组装起来并设置路由逻辑。// 创建图 const workflow new StateGraph({ schema: StateSchema }) // 添加所有节点 .addNode(classify, classifyNode) .addNode(handle_simple, handleSimpleQueryNode) .addNode(query_kb, queryKnowledgeBaseNode) .addNode(generate_from_kb, generateResponseFromKBNode) .addNode(wait_for_human, waitForHumanNode) .addNode(process_human_input, processHumanInputNode) // 设置入口 .addEdge(__start__, classify) // 根据分类结果路由 .addConditionalEdges( classify, // 路由函数根据state.classification的值决定下一个节点 (state: WorkflowState) state.classification!, { simple_q: handle_simple, need_kb: query_kb, need_human: wait_for_human, } ) // 简单问题处理后直接结束 .addEdge(handle_simple, __end__) // 知识库查询后进入生成回复节点然后结束 .addEdge(query_kb, generate_from_kb) .addEdge(generate_from_kb, __end__) // 等待人工后必须进入人工输入处理节点 .addEdge(wait_for_human, process_human_input) .addEdge(process_human_input, __end__); // 编译图并启用内存检查点生产环境需替换 const memorySaver new MemorySaver(); const app workflow.compile({ checkpointer: memorySaver }); console.log(AI客服工作流图已编译完成。);3.3 模拟执行与中断恢复让我们模拟一个完整的中断-恢复场景。// 场景用户提交了一个复杂的技术问题需要人工介入。 const initialTicket { ticketId: TICKET-2024-1001, customerQuery: 我的订单支付成功了但系统显示未支付而且我收到了两次扣款短信这到底怎么回事我要投诉, }; const config { configurable: { thread_id: initialTicket.ticketId } }; // 使用工单ID作为thread_id console.log( 第一次执行从开始到人工等待节点 ); try { // 第一次invoke工作流会运行到wait_for_human节点后暂停因为分类是need_human const result1 await app.invoke(initialTicket, config); console.log(当前状态:, JSON.stringify(result1, null, 2)); console.log(流程在 wait_for_human 节点中断并保存了检查点。); // 此时result1.finalResponse为空isResolved为false。 } catch (error) { console.error(执行出错:, error); } // 模拟一段时间后客服人员在后台系统查看了工单并给出了处理意见。 console.log(\n 模拟客服后台处理 ); // 客服通过另一个接口更新了该工单thread_id的状态中的humanAgentInput字段。 // 在LangGraph中我们可以通过向同一个thread_id的流程“发送消息”来更新状态。 // 一种常见模式是定义一个专门的“更新状态”节点并通过streamEvents或再次invoke时传入更新值来触发。 // 这里为了演示我们模拟直接修改状态后继续执行。 // 实际上更标准的做法是准备一个包含人工输入的新状态补丁然后再次invoke。 // 因为检查点存在再次invoke会从上次中断的节点wait_for_human之后继续。 const humanInterventionInput { humanAgentInput: 经核实该用户确实发生了重复支付。支付流水号分别为 TXN-A123 和 TXN-A124。已通知财务部门处理退款预计1-3个工作日到账。请向客户致歉并告知退款安排。, }; console.log( 第二次执行恢复工作流传入人工输入 ); // 注意我们再次调用invoke传入更新后的状态补丁并使用相同的config即thread_id const result2 await app.invoke(humanInterventionInput, config); console.log(恢复执行后的最终状态:, JSON.stringify(result2, null, 2)); console.log(工单是否解决: ${result2.isResolved}); console.log(最终回复: ${result2.finalResponse});运行这段代码你会看到工作流第一次执行在wait_for_human节点后“暂停”状态被完整保存。在“客服”提供了humanAgentInput后第二次执行并没有从头开始而是从wait_for_human之后的下一个节点process_human_input开始执行并最终生成包含人工处理意见的回复将工单标记为已解决。避坑指南在恢复执行时invoke传入的对象是对当前已保存状态的更新补丁而不是完整替换。比如第一次执行后状态是{classification: need_human, ...}第二次传入{humanAgentInput: ...}LangGraph.js会智能地合并这两个状态。这意味着你不需要在每次恢复时都传递完整初始状态只需传递发生变化的部分。这是其状态管理非常强大和易用的地方。4. 生产环境部署检查点持久化与错误处理在开发环境我们用MemorySaver但它的数据在进程重启后就消失了。生产环境必须使用持久化存储。LangGraph.js目前官方提供了MemorySaver和SqliteSaver实验性社区也在积极贡献其他后端如Redis、PostgreSQL。这里我们探讨一下核心思路和自定义实现的关键点。4.1 自定义检查点存储器你需要实现CheckpointSaver接口。它主要包含两个方法get和put。import { BaseCheckpointSaver, Checkpoint } from langchain/langgraph; interface CustomCheckpointSaverOptions { redisClient: any; // 假设使用ioredis } export class RedisCheckpointSaver extends BaseCheckpointSaver { private redisClient: any; private namespace: string; constructor(options: CustomCheckpointSaverOptions) { super(); this.redisClient options.redisClient; this.namespace langgraph:checkpoint; } // 根据 thread_id 和 checkpoint_id (可选) 获取检查点 async get(config: { configurable: { thread_id: string } }, checkpointId?: string) { const key checkpointId ? ${this.namespace}:${config.configurable.thread_id}:${checkpointId} : ${this.namespace}:${config.configurable.thread_id}:latest; // 通常取最新的 const data await this.redisClient.get(key); if (!data) return null; return JSON.parse(data) as Checkpoint; } // 保存检查点 async put(config: { configurable: { thread_id: string } }, checkpoint: Checkpoint) { const key ${this.namespace}:${config.configurable.thread_id}:${checkpoint.id}; // 同时保存一份为最新版本 const latestKey ${this.namespace}:${config.configurable.thread_id}:latest; const serialized JSON.stringify(checkpoint); // 使用事务或管道保证原子性 const multi this.redisClient.multi(); multi.set(key, serialized, EX, 86400); // 设置24小时过期 multi.set(latestKey, serialized, EX, 86400); await multi.exec(); } } // 使用自定义的存储器 import Redis from ioredis; const redisClient new Redis(); const redisSaver new RedisCheckpointSaver({ redisClient }); const app workflow.compile({ checkpointer: redisSaver });关键细节序列化检查点对象包含状态、元数据等必须是纯JSON可序列化的。确保你的状态定义Zod Schema里没有函数、循环引用等。版本管理每个检查点有一个唯一ID。通常我们总是保存并获取latest版本但保留历史版本对于调试和审计很有用。过期策略像Redis这样的内存数据库一定要设置合理的TTL生存时间避免无用数据堆积。并发安全在高并发下对同一个thread_id的检查点读写可能存在竞争。需要考虑使用乐观锁或Redis的WATCH/MULTI命令来保证一致性。4.2 错误处理与重试策略工作流执行中难免出错LLM API调用失败、工具调用超时、网络问题等。LangGraph.js本身不强制规定错误处理但我们可以利用其架构设计健壮的策略。策略一节点内部的Try-Catch在每个节点函数内部进行细致的错误捕获并选择如何更新状态。const robustQueryKBNode async (state: WorkflowState) { try { const result await callKnowledgeBaseAPI(state.customerQuery); return { kbSearchResult: result }; } catch (error) { console.error(知识库查询失败: ${error.message}); // 在状态中记录错误并可能路由到一个“错误处理”节点 return { kbSearchResult: 查询失败: ${error.message}, _error: error.message, // 使用一个特殊字段记录错误 }; } };策略二利用条件边进行错误路由你可以设计一个专门的error_handler节点并在其他节点出错时通过修改状态中的某个标志如_error让条件边路由到错误处理节点。// 修改状态Schema增加错误通道 const StateSchemaWithError Annotation.Root({ // ... 其他字段同上 _error: z.string().optional().describe(节点执行错误信息), _shouldHandleError: z.boolean().default(false).describe(是否触发错误处理), }); // 在可能出错的节点捕获错误并设置标志 const someRiskyNode async (state) { try { /* ... */ } catch (error) { return { _error: error.message, _shouldHandleError: true }; } }; // 在图中添加一个条件边检查 _shouldHandleError 标志 workflow.addConditionalEdges( someRiskyNode, (state) state._shouldHandleError ? error_handler : next_normal_node, { true: error_handler, false: next_normal_node } ); // 错误处理节点可以记录日志、发送告警、尝试补偿操作然后决定是重试、转人工还是失败结束。策略三基于检查点的外部重试这是最强大的模式。如果一个工作流实例因为不可抗力如进程崩溃完全失败由于检查点已经持久化你可以有一个外部监控进程或一个简单的cron job来扫描那些处于“执行中”但长时间没有更新的thread_id然后重新触发app.invoke({}, {configurable: {thread_id: target_id}})。工作流会从上一个成功的检查点开始重试而不是从头开始。生产环境建议对于关键业务流建议将每个工作流的thread_id和其最新状态/状态码如running,waiting,failed,completed记录在业务数据库的一张表里。这样你可以很方便地做健康检查、手动干预和报表统计。5. 进阶模式动态分支、人工审批与外部事件驱动掌握了基础的中断恢复后我们可以探索更复杂的模式这些模式在真实业务系统中非常常见。5.1 动态分支根据LLM输出决定多步路径有时下一个步骤不是简单的枚举分类而是需要LLM动态生成一个计划plan。我们可以让一个节点输出一个“任务列表”然后动态创建后续的执行路径。这需要更灵活的状态设计和节点调度。思路planning_node生成一个任务列表[search_web, analyze_data, write_report]并存入状态。然后一个orchestrator_node负责从列表中取出下一个任务并路由到对应的执行节点。每完成一个任务就更新状态如标记任务完成并循环回到orchestrator_node直到所有任务完成。这本质上实现了一个动态的、长度不确定的循环。// 状态扩展 const DynamicWorkflowState Annotation.Root({ objective: z.string(), plan: z.array(z.string()).optional(), // 动态计划如 [search, analyze, write] currentTaskIndex: z.number().default(0), taskResults: z.array(z.string()).optional(), finalOutput: z.string().optional(), }); // 规划节点 const plannerNode async (state) { const planPrompt 针对目标${state.objective}请列出需要执行的步骤每个步骤用简单动词描述以JSON数组格式输出例如[search_news, summarize, evaluate]; const planStr await llm.invoke(planPrompt); const plan JSON.parse(planStr.content); // 注意实际中需要更健壮的解析 return { plan, currentTaskIndex: 0 }; }; // 调度节点 const orchestratorNode async (state) { const { plan, currentTaskIndex, taskResults [] } state; if (currentTaskIndex plan.length) { return { _next: __end__ }; // 所有任务完成结束 } const currentTask plan[currentTaskIndex]; return { _next: execute_${currentTask} }; // 动态决定下一个节点名 }; // 任务执行节点示例搜索 const execute_searchNode async (state) { // 执行搜索逻辑... const result 关于${state.objective}的搜索结果摘要...; const newTaskResults [...(state.taskResults || []), result]; const nextIndex state.currentTaskIndex 1; // 更新结果和索引并指示返回调度器 return { taskResults: newTaskResults, currentTaskIndex: nextIndex, _next: orchestrator }; }; // 在图中需要将orchestratorNode连接到所有可能的execute_*节点这可以通过动态添加节点或使用一个“路由映射”来实现。这种模式非常强大可以构建出能自主规划复杂任务的AI Agent。关键在于orchestratorNode如何根据状态动态决定下一跳。5.2 集成人工审批节点在很多企业流程中AI可以处理大部分工作但关键决策需要人工审批。这可以建模为一个特殊的“中断”节点。与之前wait_for_human被动等待不同审批节点需要与外部系统如OA、邮件、钉钉/飞书审批集成。实现模式审批节点工作流执行到此节点时状态中包含需要审批的“提案”例如“是否批准该笔报销”、“是否发布这篇稿件”。该节点会调用外部API在审批系统中创建一个待办事项。将工作流状态或关键信息与这个待办事项关联例如存入数据库或用thread_id关联。然后工作流主动暂停通过到达一个没有出边的节点或者抛出一个特殊的中断信号。外部回调当审批人在外部系统完成操作批准/拒绝后该系统需要回调你的服务的一个特定API。恢复执行这个回调API收到结果后根据关联的thread_id更新工作流状态如approvalResult: approved然后再次调用app.invoke()恢复执行。// 伪代码示例 const approvalNode async (state: WorkflowState, config: any) { const { thread_id } config.configurable; const proposal state.proposalForApproval; // 1. 调用内部或外部API创建审批单并将thread_id作为关联ID const approvalTicketId await createApprovalTicket({ title: AI工作流审批: ${thread_id}, content: proposal, metadata: { langgraph_thread_id: thread_id } }); // 2. 将审批单ID也存入状态方便后续查询 // 3. 此节点执行完毕工作流进入等待。没有直接的出边或者指向一个虚拟的“等待”节点。 // 通常这里会抛出一个自定义的“中断异常”由外层逻辑捕获并处理暂停。 // 为了简化我们可以更新状态并让路由逻辑进入一个“等待循环”。 return { approvalTicketId, status: pending_approval, _pause: true // 自定义标志供条件边判断 }; }; // 在图中可以设置条件边如果 _pause 为 true则路由到一个不执行任何操作、也没有出边的“挂起”节点实现暂停。 // 或者更优雅的方式是利用LangGraph的“中断”机制如果未来版本提供更直接的支持。5.3 外部事件驱动与消息队列集成对于高吞吐量或需要与多个外部系统集成的场景可以将LangGraph.js工作流作为“消息处理器”来运行。每个thread_id对应一个独立的业务流程实例。消费消息使用Kafka、RabbitMQ或AWS SQS等消息队列。消费者从队列中取出消息消息体中包含thread_id和需要更新的状态数据stateUpdate。调用工作流消费者调用app.invoke(stateUpdate, { configurable: { thread_id } })。由于检查点存在工作流会从上次中断处继续执行。产生新消息工作流执行到某个节点时可能需要触发外部操作如发送邮件、调用API。这个节点可以不直接执行而是将需要执行的任务作为一条新消息发送到另一个队列然后自身暂停。由专门的服务消费那个队列完成任务后再发送一条“任务完成”的消息回来驱动工作流恢复。这种架构将工作流引擎变成了一个状态驱动的消息路由器实现了极高的解耦和可扩展性。LangGraph.js的检查点机制保证了即使在消息处理过程中发生故障状态也不会丢失可以安全重试。6. 调试、监控与性能考量构建复杂的工作流调试和监控是必不可少的。6.1 可视化与调试LangGraph Studio是一个官方的可视化调试工具目前对Python支持更好但JS生态也在跟进。对于JS版本目前可以依靠以下方式日志记录在每个节点的开始和结束添加详细的日志打印thread_id、节点名、输入/输出状态片段。结构化日志输出为JSON便于后续收集到ELK或Datadog等系统。状态快照利用检查点存储器你可以随时查询任意thread_id的最新状态这是最直接的调试手段。手动执行与追踪在开发时可以使用app.stream()或app.streamEvents()方法来逐步执行工作流并观察每个节点前后的状态变化。streamEvents提供了更细粒度的事件流非常适合调试。// 使用streamEvents进行调试 const events app.streamEvents( initialTicket, { configurable: { thread_id: debug_1 } }, { version: v1 } ); for await (const event of events) { // 事件类型包括on_chain_start, on_chain_end, on_tool_start, on_tool_end 等 console.log([${event.event}] ${event.name}, event.data || {}); // 可以在这里记录或检查状态 }6.2 性能优化要点状态大小状态对象会被频繁序列化/反序列化并持久化。务必保持状态精简只存储必要数据。避免将大型文件内容如图片、长文本直接放在状态里可以存储引用如文件ID、URL。检查点频率默认每个节点后都保存检查点。对于性能极其敏感、且节点失败风险低的场景可以考虑自定义检查点策略例如只在关键节点或“等待”节点保存。这需要更底层的控制可能需要对LangGraph.js进行扩展。LLM调用优化工作流中往往包含多个LLM调用这是主要的耗时和成本来源。考虑缓存对具有确定性的LLM查询如分类、标准化结果进行缓存。并行化如果多个节点间没有数据依赖可以考虑使用Promise.all在一个节点内并行执行多个LLM调用或工具调用而不是设计成串行节点。模型选型非核心的、简单的分类或提取任务使用小型/快速的模型如gpt-4o-mini把大模型如GPT-4留给需要复杂推理的环节。节点粒度节点的粒度要适中。太粗一个节点做太多事不利于复用和调试太细每个小操作都是一个节点会增加图的管理开销和序列化成本。一个经验法则是一个节点应该完成一个逻辑上连贯的、可以独立描述的任务。6.3 与LangChain的协同LangGraph.js和LangChain是绝佳搭档。你可以直接在你的LangGraph节点函数中使用LangChain的组件ChatOpenAI,ChatAnthropic等LLM集成。SerpAPI,RequestsToolkit等工具。ConversationSummaryBufferMemory等记忆组件不过LangGraph的状态管理通常更强大。RecursiveCharacterTextSplitter,VectorStoreRetriever等RAG相关组件。实际上你可以把LangChain看成是“零件箱”而LangGraph.js是组装这些零件并赋予其可控流程的“流水线图纸和控制器”。在构建复杂AI应用时我通常会先用LangChain快速验证想法的各个部分然后用LangGraph.js将它们组织成一个健壮、可维护的工作流。从我自己的几个生产项目迁移经验来看从纯LangChain链式结构转向LangGraph.js初期会有一些概念转换的成本但一旦熟悉了“图”和“状态”的思维方式代码的可读性、可维护性和系统的可靠性都会得到显著提升。尤其是当你的AI应用开始需要处理多轮交互、复杂决策和外部系统集成时LangGraph.js提供的这套范式几乎是必然的选择。
LangGraph.js：构建可中断、可恢复的AI工作流与智能体
1. 从LangChain到LangGraph为什么我们需要“可中断”的AI工作流如果你在过去一两年里折腾过AI应用开发尤其是基于大语言模型LLM构建一些自动化流程那么“LangChain”这个名字你一定不陌生。它像是一套乐高积木把提示词模板、记忆、工具调用这些组件串起来让我们能相对容易地搭建起一个AI驱动的对话机器人或者文档分析工具。我早期很多原型项目都是基于LangChain快速搭起来的它确实极大地降低了入门门槛。但用久了尤其是在尝试构建一些稍微复杂点的、带点“业务流程”味道的应用时痛点就来了状态管理太麻烦流程一旦跑起来就像脱缰的野马想中途干预一下、暂停一下、或者失败了从某个点重试简直是一场噩梦。这就像你写一个简单的脚本A - B - C顺序执行没问题。但现实中的业务逻辑往往是这样的A - (根据结果判断) - 要么走B1要么走B2 - 等待外部输入比如用户确认- 继续执行C - 如果C失败回退到B2重试。用传统的链式Chain思维来硬套代码会迅速变得臃肿不堪各种if-else嵌套在回调函数里状态散落在各个角落。更头疼的是“可恢复性”想象一个处理长篇文档的AI工作流运行到一半服务器重启了你难道要从头再处理一遍吗这就是LangGraph.js切入的战场。它不是一个替代LangChain的全新框架而是一个专门为构建有状态、多步骤、可循环且最重要的是可中断与可恢复的AI智能体Agent或工作流而设计的库。你可以把它理解为在LangChain的“积木块”之上提供了一套可视化、可编程的“流程图”绘制和引擎。它的核心模型是一个有向图节点Node是你的处理函数可以是LLM调用、工具执行、条件判断边Edge定义了节点间的流转逻辑。最关键的是它内置了完整的状态管理和检查点Checkpoint机制这让“暂停-继续”和“失败重试”从概念变成了几行配置就能实现的功能。所以当我们谈论“LangGraph.js可中断可恢复的AI工作流”时我们本质上是在讨论如何将那些脆弱、冗长、不可控的AI自动化脚本升级为健壮、灵活、像传统工作流引擎如Airflow、Camunda一样可靠的应用。这对于构建复杂的AI助手、自动化客服、内容审核流水线、多步骤数据分析Agent等场景是质的飞跃。2. LangGraph.js核心架构拆解图、状态与检查点要玩转可中断的工作流必须吃透LangGraph.js的三个核心概念图Graph、状态State和检查点Checkpoint。这三者构成了其可中断、可恢复能力的基石。2.1 图Graph工作流的骨架LangGraph.js中的“图”就是你的工作流蓝图。它由两种基本元素构成节点Nodes这是实际干活的地方。每个节点是一个异步函数它接收当前整个工作流的“状态”对象执行一些操作比如调用LLM、查询数据库、运行计算然后返回一个对状态的更新。在JavaScript/TypeScript中它通常长这样const myNode async (state: StateType) { // 从state中读取所需数据 const userQuestion state.user_input; // 执行核心逻辑例如调用LLM const llmResponse await chatModel.invoke(请回答${userQuestion}); // 返回一个对象这个对象会用来更新全局state return { assistant_response: llmResponse.content }; };关键点在于节点函数不直接修改传入的state而是返回一个更新“补丁”。这保证了状态变化的可预测性和可序列化。边Edges决定了工作流的走向。分为两种条件边Conditional Edges根据当前状态的值决定下一步去哪个节点。这是实现分支if-else和循环的关键。普通边Regular Edges无条件地从一个节点指向下一个节点。通过组合节点和边你可以定义出非常复杂的流程比如经典的“ReActReasoning Acting代理”循环思考 - 判断是否需要行动 - 是则执行工具 - 合并结果并循环。2.2 状态State工作流的记忆中枢状态是一个中心化的、类型化的对象它随着工作流的执行而演变。LangGraph.js强烈推荐在TypeScript中几乎是必须使用Zod库来定义状态的模式Schema。这不仅是类型安全的需要更是可序列化的前提。import { z } from zod; // 1. 使用Zod定义状态结构 const StateSchema z.object({ // 输入 user_input: z.string(), // 中间过程 ai_thoughts: z.string().optional(), // 代理的“思考”过程 tools_to_call: z.array(z.string()).optional(), // 决定要调用的工具 tool_results: z.array(z.any()).optional(), // 工具执行结果 // 输出 final_answer: z.string().optional(), // 元数据对中断/恢复很有用 current_step: z.string().optional(), // 当前所在节点名 error: z.string().optional(), // 错误信息 }); // 2. 推导出TypeScript类型 type MyWorkflowState z.infertypeof StateSchema;所有节点都读写这个统一的状态对象。当工作流被中断时LangGraph.js需要将整个状态对象可能包含LLM的对话历史、工具调用结果等完整地保存下来。这就要求状态里的所有值都必须是可序列化为JSON的。像函数、DOM元素这类不可序列化的东西绝对不能直接放在状态里。2.3 检查点Checkpoint实现可中断与可恢复的魔法这是LangGraph.js最精髓的部分。检查点机制允许你在工作流执行到任何一个节点后将当前完整的状态State以及工作流所处的上下文比如接下来该执行哪个节点持久化保存起来。它是如何工作的配置检查点存储器你需要提供一个“检查点存储器”CheckpointSaver的实现。LangGraph.js提供了内存存储MemorySaver用于开发测试但对于生产环境你需要将其存储到数据库如Redis、PostgreSQL或文件系统中。import { MemorySaver } from langchain/langgraph; const checkpointSaver new MemorySaver(); // 开发用 // 生产环境可能需要 new RedisSaver(redisClient) 之类的自定义实现在图中启用检查点创建图时将存储器传入。const workflow new StateGraph({ schema: StateSchema, }) .addNode(process_input, processInputNode) .addEdge(start, process_input) .compile({ checkpointer: checkpointSaver, // 关键启用检查点 });执行与中断当你调用workflow.invoke()时可以传入一个config对象其中包含一个configurable字段通常用来指定这次运行的“线程ID”thread_id。const initialInput { user_input: 今天的天气怎么样 }; const config { configurable: { thread_id: user_123_session_1 } }; // 第一次执行可能只执行了几步就被主动暂停或意外中断 const result1 await workflow.invoke(initialInput, config);执行过程中每经过一个节点或你配置的特定节点引擎都会自动调用检查点存储器将当前状态快照保存起来并与这个thread_id关联。恢复执行当需要恢复时你不需要重新构造初始状态。只需使用**相同的thread_id**再次调用invoke甚至可以传入新的输入来更新状态。// 一段时间后恢复执行。注意这里没有传initialInput因为状态已保存。 const result2 await workflow.invoke({}, config); // 从上次中断处继续 // 或者提供新的输入来更新状态后再继续 const result3 await workflow.invoke({ user_input: 那么明天呢 }, config);引擎会根据thread_id从检查点存储器加载最新的状态和进度然后从上次中断的节点之后继续执行。这对于处理长对话、分步任务和错误恢复至关重要。实操心得thread_id的设计非常巧妙。它可以是用户ID、会话ID、或任务ID。这让你能轻松管理同一个工作流的多个并行实例。例如一个客服机器人每个用户对话就是一个独立的、可随时暂停和恢复的工作流线程。3. 构建一个可中断的AI客服工单处理工作流理论说再多不如动手。我们来设计一个稍微贴近实际场景的例子一个AI客服工单自动处理工作流。它的流程是1) 分类用户问题2) 根据分类要么直接回答简单问题要么查询知识库要么在需要人工时暂停并等待3) 最终生成回复。这个流程天然需要“可中断”——因为在“等待人工”节点工作流必须暂停直到客服人员提供了干预信息后才能继续。3.1 定义状态与节点首先定义工作流的状态。我们需要记录用户问题、AI分类结果、查询到的知识、人工干预输入以及最终回复。import { z } from zod; import { StateGraph, Annotation } from langchain/langgraph; import { ChatOpenAI } from langchain/openai; import { MemorySaver } from langchain/langgraph; // 使用Annotation来方便地定义带默认值的状态模式 const StateSchema Annotation.Root({ // 输入 ticketId: z.string().describe(工单唯一ID), customerQuery: z.string().describe(客户原始问题), // 处理过程 classification: z.enum([simple_q, need_kb, need_human]).optional().describe(AI分类结果), kbSearchResult: z.string().optional().describe(知识库查询结果), humanAgentInput: z.string().optional().describe(人工客服的补充输入或指示), // 输出与元数据 aiResponse: z.string().optional().describe(AI生成的回复), finalResponse: z.string().optional().describe(最终发给客户的回复), isResolved: z.boolean().default(false).describe(工单是否已解决), }); type WorkflowState typeof StateSchema.State; // 初始化LLM const llm new ChatOpenAI({ modelName: gpt-4o-mini, temperature: 0, }); // 节点1分类用户问题 const classifyNode async (state: WorkflowState) { console.log([分类节点] 处理工单: ${state.ticketId}); const classifyPrompt 请将以下客户问题分类 - simple_q: 简单问候、感谢或非常基础的问题如“你们上班时间”。 - need_kb: 需要查询产品文档、政策条款才能回答的具体问题。 - need_human: 涉及投诉、退款、复杂技术问题或需要人工判断的情感化问题。 客户问题${state.customerQuery} 只输出分类标签simple_q, need_kb, need_human不要任何其他文字。 ; const classification (await llm.invoke(classifyPrompt)).content.trim() as WorkflowState[classification]; // 返回状态更新补丁 return { classification }; }; // 节点2处理简单问题 const handleSimpleQueryNode async (state: WorkflowState) { console.log([简单问题节点] 直接生成回复); const responsePrompt 客户问了一个简单问题${state.customerQuery}。请以友好、专业的客服口吻直接回答。; const aiResponse (await llm.invoke(responsePrompt)).content; return { aiResponse, finalResponse: aiResponse, isResolved: true }; }; // 节点3查询知识库模拟 const queryKnowledgeBaseNode async (state: WorkflowState) { console.log([知识库节点] 模拟查询); // 这里应该是向量数据库查询等真实操作我们模拟一个结果 await new Promise(resolve setTimeout(resolve, 500)); // 模拟延迟 const mockKbResult 根据知识库文档#2024-001相关问题的标准解决方案是请先尝试重启应用并检查网络连接。如果问题持续请联系技术支持。; return { kbSearchResult: mockKbResult }; }; // 节点4基于知识库生成回复 const generateResponseFromKBNode async (state: WorkflowState) { console.log([生成KB回复节点]); const responsePrompt 基于以下知识库信息回答客户的问题。 客户问题${state.customerQuery} 知识库信息${state.kbSearchResult} 请生成完整、友好的回复。; const aiResponse (await llm.invoke(responsePrompt)).content; return { aiResponse, finalResponse: aiResponse, isResolved: true }; }; // 节点5挂起等待人工干预这是一个“中断点” const waitForHumanNode async (state: WorkflowState) { console.log([等待人工节点] 工单 ${state.ticketId} 已挂起等待客服处理。); // 这个节点本身不修改状态它只是流程中的一个“暂停门”。 // 关键执行到这里检查点已经保存。工作流会停在这里。 // 恢复执行需要外部触发例如调用一个API来更新humanAgentInput状态。 return {}; }; // 节点6处理人工输入并生成最终回复 const processHumanInputNode async (state: WorkflowState) { console.log([处理人工输入节点] 收到客服指示: ${state.humanAgentInput}); if (!state.humanAgentInput) { return { finalResponse: 已转接人工请稍候。, isResolved: false }; } const responsePrompt 客服提供了以下处理意见${state.humanAgentInput}。客户的原问题是${state.customerQuery}。请结合两者生成最终回复给客户。; const finalResponse (await llm.invoke(responsePrompt)).content; return { finalResponse, isResolved: true }; };3.2 组装图并配置条件路由现在我们把节点组装起来并设置路由逻辑。// 创建图 const workflow new StateGraph({ schema: StateSchema }) // 添加所有节点 .addNode(classify, classifyNode) .addNode(handle_simple, handleSimpleQueryNode) .addNode(query_kb, queryKnowledgeBaseNode) .addNode(generate_from_kb, generateResponseFromKBNode) .addNode(wait_for_human, waitForHumanNode) .addNode(process_human_input, processHumanInputNode) // 设置入口 .addEdge(__start__, classify) // 根据分类结果路由 .addConditionalEdges( classify, // 路由函数根据state.classification的值决定下一个节点 (state: WorkflowState) state.classification!, { simple_q: handle_simple, need_kb: query_kb, need_human: wait_for_human, } ) // 简单问题处理后直接结束 .addEdge(handle_simple, __end__) // 知识库查询后进入生成回复节点然后结束 .addEdge(query_kb, generate_from_kb) .addEdge(generate_from_kb, __end__) // 等待人工后必须进入人工输入处理节点 .addEdge(wait_for_human, process_human_input) .addEdge(process_human_input, __end__); // 编译图并启用内存检查点生产环境需替换 const memorySaver new MemorySaver(); const app workflow.compile({ checkpointer: memorySaver }); console.log(AI客服工作流图已编译完成。);3.3 模拟执行与中断恢复让我们模拟一个完整的中断-恢复场景。// 场景用户提交了一个复杂的技术问题需要人工介入。 const initialTicket { ticketId: TICKET-2024-1001, customerQuery: 我的订单支付成功了但系统显示未支付而且我收到了两次扣款短信这到底怎么回事我要投诉, }; const config { configurable: { thread_id: initialTicket.ticketId } }; // 使用工单ID作为thread_id console.log( 第一次执行从开始到人工等待节点 ); try { // 第一次invoke工作流会运行到wait_for_human节点后暂停因为分类是need_human const result1 await app.invoke(initialTicket, config); console.log(当前状态:, JSON.stringify(result1, null, 2)); console.log(流程在 wait_for_human 节点中断并保存了检查点。); // 此时result1.finalResponse为空isResolved为false。 } catch (error) { console.error(执行出错:, error); } // 模拟一段时间后客服人员在后台系统查看了工单并给出了处理意见。 console.log(\n 模拟客服后台处理 ); // 客服通过另一个接口更新了该工单thread_id的状态中的humanAgentInput字段。 // 在LangGraph中我们可以通过向同一个thread_id的流程“发送消息”来更新状态。 // 一种常见模式是定义一个专门的“更新状态”节点并通过streamEvents或再次invoke时传入更新值来触发。 // 这里为了演示我们模拟直接修改状态后继续执行。 // 实际上更标准的做法是准备一个包含人工输入的新状态补丁然后再次invoke。 // 因为检查点存在再次invoke会从上次中断的节点wait_for_human之后继续。 const humanInterventionInput { humanAgentInput: 经核实该用户确实发生了重复支付。支付流水号分别为 TXN-A123 和 TXN-A124。已通知财务部门处理退款预计1-3个工作日到账。请向客户致歉并告知退款安排。, }; console.log( 第二次执行恢复工作流传入人工输入 ); // 注意我们再次调用invoke传入更新后的状态补丁并使用相同的config即thread_id const result2 await app.invoke(humanInterventionInput, config); console.log(恢复执行后的最终状态:, JSON.stringify(result2, null, 2)); console.log(工单是否解决: ${result2.isResolved}); console.log(最终回复: ${result2.finalResponse});运行这段代码你会看到工作流第一次执行在wait_for_human节点后“暂停”状态被完整保存。在“客服”提供了humanAgentInput后第二次执行并没有从头开始而是从wait_for_human之后的下一个节点process_human_input开始执行并最终生成包含人工处理意见的回复将工单标记为已解决。避坑指南在恢复执行时invoke传入的对象是对当前已保存状态的更新补丁而不是完整替换。比如第一次执行后状态是{classification: need_human, ...}第二次传入{humanAgentInput: ...}LangGraph.js会智能地合并这两个状态。这意味着你不需要在每次恢复时都传递完整初始状态只需传递发生变化的部分。这是其状态管理非常强大和易用的地方。4. 生产环境部署检查点持久化与错误处理在开发环境我们用MemorySaver但它的数据在进程重启后就消失了。生产环境必须使用持久化存储。LangGraph.js目前官方提供了MemorySaver和SqliteSaver实验性社区也在积极贡献其他后端如Redis、PostgreSQL。这里我们探讨一下核心思路和自定义实现的关键点。4.1 自定义检查点存储器你需要实现CheckpointSaver接口。它主要包含两个方法get和put。import { BaseCheckpointSaver, Checkpoint } from langchain/langgraph; interface CustomCheckpointSaverOptions { redisClient: any; // 假设使用ioredis } export class RedisCheckpointSaver extends BaseCheckpointSaver { private redisClient: any; private namespace: string; constructor(options: CustomCheckpointSaverOptions) { super(); this.redisClient options.redisClient; this.namespace langgraph:checkpoint; } // 根据 thread_id 和 checkpoint_id (可选) 获取检查点 async get(config: { configurable: { thread_id: string } }, checkpointId?: string) { const key checkpointId ? ${this.namespace}:${config.configurable.thread_id}:${checkpointId} : ${this.namespace}:${config.configurable.thread_id}:latest; // 通常取最新的 const data await this.redisClient.get(key); if (!data) return null; return JSON.parse(data) as Checkpoint; } // 保存检查点 async put(config: { configurable: { thread_id: string } }, checkpoint: Checkpoint) { const key ${this.namespace}:${config.configurable.thread_id}:${checkpoint.id}; // 同时保存一份为最新版本 const latestKey ${this.namespace}:${config.configurable.thread_id}:latest; const serialized JSON.stringify(checkpoint); // 使用事务或管道保证原子性 const multi this.redisClient.multi(); multi.set(key, serialized, EX, 86400); // 设置24小时过期 multi.set(latestKey, serialized, EX, 86400); await multi.exec(); } } // 使用自定义的存储器 import Redis from ioredis; const redisClient new Redis(); const redisSaver new RedisCheckpointSaver({ redisClient }); const app workflow.compile({ checkpointer: redisSaver });关键细节序列化检查点对象包含状态、元数据等必须是纯JSON可序列化的。确保你的状态定义Zod Schema里没有函数、循环引用等。版本管理每个检查点有一个唯一ID。通常我们总是保存并获取latest版本但保留历史版本对于调试和审计很有用。过期策略像Redis这样的内存数据库一定要设置合理的TTL生存时间避免无用数据堆积。并发安全在高并发下对同一个thread_id的检查点读写可能存在竞争。需要考虑使用乐观锁或Redis的WATCH/MULTI命令来保证一致性。4.2 错误处理与重试策略工作流执行中难免出错LLM API调用失败、工具调用超时、网络问题等。LangGraph.js本身不强制规定错误处理但我们可以利用其架构设计健壮的策略。策略一节点内部的Try-Catch在每个节点函数内部进行细致的错误捕获并选择如何更新状态。const robustQueryKBNode async (state: WorkflowState) { try { const result await callKnowledgeBaseAPI(state.customerQuery); return { kbSearchResult: result }; } catch (error) { console.error(知识库查询失败: ${error.message}); // 在状态中记录错误并可能路由到一个“错误处理”节点 return { kbSearchResult: 查询失败: ${error.message}, _error: error.message, // 使用一个特殊字段记录错误 }; } };策略二利用条件边进行错误路由你可以设计一个专门的error_handler节点并在其他节点出错时通过修改状态中的某个标志如_error让条件边路由到错误处理节点。// 修改状态Schema增加错误通道 const StateSchemaWithError Annotation.Root({ // ... 其他字段同上 _error: z.string().optional().describe(节点执行错误信息), _shouldHandleError: z.boolean().default(false).describe(是否触发错误处理), }); // 在可能出错的节点捕获错误并设置标志 const someRiskyNode async (state) { try { /* ... */ } catch (error) { return { _error: error.message, _shouldHandleError: true }; } }; // 在图中添加一个条件边检查 _shouldHandleError 标志 workflow.addConditionalEdges( someRiskyNode, (state) state._shouldHandleError ? error_handler : next_normal_node, { true: error_handler, false: next_normal_node } ); // 错误处理节点可以记录日志、发送告警、尝试补偿操作然后决定是重试、转人工还是失败结束。策略三基于检查点的外部重试这是最强大的模式。如果一个工作流实例因为不可抗力如进程崩溃完全失败由于检查点已经持久化你可以有一个外部监控进程或一个简单的cron job来扫描那些处于“执行中”但长时间没有更新的thread_id然后重新触发app.invoke({}, {configurable: {thread_id: target_id}})。工作流会从上一个成功的检查点开始重试而不是从头开始。生产环境建议对于关键业务流建议将每个工作流的thread_id和其最新状态/状态码如running,waiting,failed,completed记录在业务数据库的一张表里。这样你可以很方便地做健康检查、手动干预和报表统计。5. 进阶模式动态分支、人工审批与外部事件驱动掌握了基础的中断恢复后我们可以探索更复杂的模式这些模式在真实业务系统中非常常见。5.1 动态分支根据LLM输出决定多步路径有时下一个步骤不是简单的枚举分类而是需要LLM动态生成一个计划plan。我们可以让一个节点输出一个“任务列表”然后动态创建后续的执行路径。这需要更灵活的状态设计和节点调度。思路planning_node生成一个任务列表[search_web, analyze_data, write_report]并存入状态。然后一个orchestrator_node负责从列表中取出下一个任务并路由到对应的执行节点。每完成一个任务就更新状态如标记任务完成并循环回到orchestrator_node直到所有任务完成。这本质上实现了一个动态的、长度不确定的循环。// 状态扩展 const DynamicWorkflowState Annotation.Root({ objective: z.string(), plan: z.array(z.string()).optional(), // 动态计划如 [search, analyze, write] currentTaskIndex: z.number().default(0), taskResults: z.array(z.string()).optional(), finalOutput: z.string().optional(), }); // 规划节点 const plannerNode async (state) { const planPrompt 针对目标${state.objective}请列出需要执行的步骤每个步骤用简单动词描述以JSON数组格式输出例如[search_news, summarize, evaluate]; const planStr await llm.invoke(planPrompt); const plan JSON.parse(planStr.content); // 注意实际中需要更健壮的解析 return { plan, currentTaskIndex: 0 }; }; // 调度节点 const orchestratorNode async (state) { const { plan, currentTaskIndex, taskResults [] } state; if (currentTaskIndex plan.length) { return { _next: __end__ }; // 所有任务完成结束 } const currentTask plan[currentTaskIndex]; return { _next: execute_${currentTask} }; // 动态决定下一个节点名 }; // 任务执行节点示例搜索 const execute_searchNode async (state) { // 执行搜索逻辑... const result 关于${state.objective}的搜索结果摘要...; const newTaskResults [...(state.taskResults || []), result]; const nextIndex state.currentTaskIndex 1; // 更新结果和索引并指示返回调度器 return { taskResults: newTaskResults, currentTaskIndex: nextIndex, _next: orchestrator }; }; // 在图中需要将orchestratorNode连接到所有可能的execute_*节点这可以通过动态添加节点或使用一个“路由映射”来实现。这种模式非常强大可以构建出能自主规划复杂任务的AI Agent。关键在于orchestratorNode如何根据状态动态决定下一跳。5.2 集成人工审批节点在很多企业流程中AI可以处理大部分工作但关键决策需要人工审批。这可以建模为一个特殊的“中断”节点。与之前wait_for_human被动等待不同审批节点需要与外部系统如OA、邮件、钉钉/飞书审批集成。实现模式审批节点工作流执行到此节点时状态中包含需要审批的“提案”例如“是否批准该笔报销”、“是否发布这篇稿件”。该节点会调用外部API在审批系统中创建一个待办事项。将工作流状态或关键信息与这个待办事项关联例如存入数据库或用thread_id关联。然后工作流主动暂停通过到达一个没有出边的节点或者抛出一个特殊的中断信号。外部回调当审批人在外部系统完成操作批准/拒绝后该系统需要回调你的服务的一个特定API。恢复执行这个回调API收到结果后根据关联的thread_id更新工作流状态如approvalResult: approved然后再次调用app.invoke()恢复执行。// 伪代码示例 const approvalNode async (state: WorkflowState, config: any) { const { thread_id } config.configurable; const proposal state.proposalForApproval; // 1. 调用内部或外部API创建审批单并将thread_id作为关联ID const approvalTicketId await createApprovalTicket({ title: AI工作流审批: ${thread_id}, content: proposal, metadata: { langgraph_thread_id: thread_id } }); // 2. 将审批单ID也存入状态方便后续查询 // 3. 此节点执行完毕工作流进入等待。没有直接的出边或者指向一个虚拟的“等待”节点。 // 通常这里会抛出一个自定义的“中断异常”由外层逻辑捕获并处理暂停。 // 为了简化我们可以更新状态并让路由逻辑进入一个“等待循环”。 return { approvalTicketId, status: pending_approval, _pause: true // 自定义标志供条件边判断 }; }; // 在图中可以设置条件边如果 _pause 为 true则路由到一个不执行任何操作、也没有出边的“挂起”节点实现暂停。 // 或者更优雅的方式是利用LangGraph的“中断”机制如果未来版本提供更直接的支持。5.3 外部事件驱动与消息队列集成对于高吞吐量或需要与多个外部系统集成的场景可以将LangGraph.js工作流作为“消息处理器”来运行。每个thread_id对应一个独立的业务流程实例。消费消息使用Kafka、RabbitMQ或AWS SQS等消息队列。消费者从队列中取出消息消息体中包含thread_id和需要更新的状态数据stateUpdate。调用工作流消费者调用app.invoke(stateUpdate, { configurable: { thread_id } })。由于检查点存在工作流会从上次中断处继续执行。产生新消息工作流执行到某个节点时可能需要触发外部操作如发送邮件、调用API。这个节点可以不直接执行而是将需要执行的任务作为一条新消息发送到另一个队列然后自身暂停。由专门的服务消费那个队列完成任务后再发送一条“任务完成”的消息回来驱动工作流恢复。这种架构将工作流引擎变成了一个状态驱动的消息路由器实现了极高的解耦和可扩展性。LangGraph.js的检查点机制保证了即使在消息处理过程中发生故障状态也不会丢失可以安全重试。6. 调试、监控与性能考量构建复杂的工作流调试和监控是必不可少的。6.1 可视化与调试LangGraph Studio是一个官方的可视化调试工具目前对Python支持更好但JS生态也在跟进。对于JS版本目前可以依靠以下方式日志记录在每个节点的开始和结束添加详细的日志打印thread_id、节点名、输入/输出状态片段。结构化日志输出为JSON便于后续收集到ELK或Datadog等系统。状态快照利用检查点存储器你可以随时查询任意thread_id的最新状态这是最直接的调试手段。手动执行与追踪在开发时可以使用app.stream()或app.streamEvents()方法来逐步执行工作流并观察每个节点前后的状态变化。streamEvents提供了更细粒度的事件流非常适合调试。// 使用streamEvents进行调试 const events app.streamEvents( initialTicket, { configurable: { thread_id: debug_1 } }, { version: v1 } ); for await (const event of events) { // 事件类型包括on_chain_start, on_chain_end, on_tool_start, on_tool_end 等 console.log([${event.event}] ${event.name}, event.data || {}); // 可以在这里记录或检查状态 }6.2 性能优化要点状态大小状态对象会被频繁序列化/反序列化并持久化。务必保持状态精简只存储必要数据。避免将大型文件内容如图片、长文本直接放在状态里可以存储引用如文件ID、URL。检查点频率默认每个节点后都保存检查点。对于性能极其敏感、且节点失败风险低的场景可以考虑自定义检查点策略例如只在关键节点或“等待”节点保存。这需要更底层的控制可能需要对LangGraph.js进行扩展。LLM调用优化工作流中往往包含多个LLM调用这是主要的耗时和成本来源。考虑缓存对具有确定性的LLM查询如分类、标准化结果进行缓存。并行化如果多个节点间没有数据依赖可以考虑使用Promise.all在一个节点内并行执行多个LLM调用或工具调用而不是设计成串行节点。模型选型非核心的、简单的分类或提取任务使用小型/快速的模型如gpt-4o-mini把大模型如GPT-4留给需要复杂推理的环节。节点粒度节点的粒度要适中。太粗一个节点做太多事不利于复用和调试太细每个小操作都是一个节点会增加图的管理开销和序列化成本。一个经验法则是一个节点应该完成一个逻辑上连贯的、可以独立描述的任务。6.3 与LangChain的协同LangGraph.js和LangChain是绝佳搭档。你可以直接在你的LangGraph节点函数中使用LangChain的组件ChatOpenAI,ChatAnthropic等LLM集成。SerpAPI,RequestsToolkit等工具。ConversationSummaryBufferMemory等记忆组件不过LangGraph的状态管理通常更强大。RecursiveCharacterTextSplitter,VectorStoreRetriever等RAG相关组件。实际上你可以把LangChain看成是“零件箱”而LangGraph.js是组装这些零件并赋予其可控流程的“流水线图纸和控制器”。在构建复杂AI应用时我通常会先用LangChain快速验证想法的各个部分然后用LangGraph.js将它们组织成一个健壮、可维护的工作流。从我自己的几个生产项目迁移经验来看从纯LangChain链式结构转向LangGraph.js初期会有一些概念转换的成本但一旦熟悉了“图”和“状态”的思维方式代码的可读性、可维护性和系统的可靠性都会得到显著提升。尤其是当你的AI应用开始需要处理多轮交互、复杂决策和外部系统集成时LangGraph.js提供的这套范式几乎是必然的选择。 
