1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一两年里折腾过LLM应用开发大概率听说过或者用过LangChain。它确实是个好东西帮我们解决了LLM调用、工具集成、记忆管理等一系列繁琐问题让构建一个能聊天的AI应用变得像搭积木一样简单。但不知道你有没有遇到过这样的场景你想构建一个稍微复杂点的流程比如一个能根据用户问题自动决定是去查数据库、调用API还是直接生成答案的客服机器人。用LangChain的Chain和Agent来搞代码写着写着就感觉有点“拧巴”——状态管理分散在各个节点流程控制依赖大量的if-else想要实现个循环或者让某个节点根据条件重复执行更是得费一番功夫。这就是LangGraph诞生的背景。它不是要取代LangChain而是LangChain生态中的一个新成员专门用来解决复杂、有状态的工作流问题。你可以把它理解成从“链式思维”到“图式思维”的升级。LangChain的Chain是线性的一个接一个而LangGraph是图结构的节点Node之间可以有分支、循环、并行。这特别适合那些需要根据中间结果动态决定下一步做什么的应用比如多轮对话、复杂决策、带有审核或修正环节的自动化流程。我最初接触LangGraph的Python版时就被它的清晰和强大吸引了。但作为主要工作在Node.js/TypeScript环境下的开发者我一直在等它的TypeScript版本变得足够成熟。现在这一天已经来了。TypeScript版的LangGraph不仅功能上向Python版看齐而且完美融入了TypeScript的类型系统让你在享受灵活图计算的同时还能获得强大的类型安全和IDE提示。这篇内容我就结合自己从Python迁移到TypeScript的实践经验带你从零开始彻底搞懂LangGraph TypeScript版的核心概念、最佳实践以及那些官方文档里可能没细说的“坑”。2. LangGraph核心三要素State、Node、Edge要玩转LangGraph首先得吃透它的三个核心概念State、Node和Edge。这是构建任何图的基础理解透了后面写代码就是水到渠成。2.1 State工作流的“记忆中枢”State是LangGraph里最核心也最容易让人一开始困惑的概念。它不是一个简单的变量而是一个定义了整个工作流所有共享数据结构的类型。你可以把它想象成游戏里角色的属性面板里面记录了生命值、魔法值、装备、任务进度等所有需要在不同场景节点间传递和修改的信息。在TypeScript里我们用一个接口Interface或类型别名Type Alias来定义State。LangGraph要求这个State必须包含一个特殊的字段messages类型是BaseMessage[]来自langchain/core。这是为了与LangChain的聊天模型兼容。但除此之外你可以自由添加任何字段。import { BaseMessage } from langchain/core/messages; // 定义工作流的状态类型 interface AgentState { // 必需字段消息历史 messages: BaseMessage[]; // 自定义字段用户问题 question: string; // 自定义字段从知识库检索到的上下文 context?: string; // 自定义字段LLM生成的最终答案 answer?: string; // 自定义字段记录调用了哪些工具 toolsUsed: string[]; // 自定义字段控制流程的标记比如是否需要人工审核 needsHumanReview?: boolean; }关键理解这个AgentState接口定义了一个对象“形状”。之后图里的每一个Node节点函数接收到的第一个参数就是当前这个State对象的一个代理Proxy。你在Node函数里修改这个State比如state.answer “某答案”修改的是共享的、全局的状态。这保证了数据在节点间的流动。实操心得在设计State时尽量保持扁平结构避免嵌套过深的对象。因为State会在每个节点被序列化和传递尤其在分布式场景下复杂的嵌套会影响性能。同时为每个字段想好明确的用途避免变成“杂物间”。使用可选属性?来区分哪些是中间过程可能缺失的字段。2.2 Node执行具体任务的“工作单元”Node节点就是一个普通的异步函数它接收两个参数当前的State和一个可选的配置对象。它的职责是执行一段逻辑并返回一个部分State更新。这个“返回部分更新”的概念很重要。你不需要返回完整的State对象只需要返回一个对象里面包含你这次执行想要修改的那些字段。LangGraph会帮你自动合并到全局State中。import { BaseMessage, HumanMessage } from langchain/core/messages; // 一个节点处理用户输入 async function processInput(state: AgentState): PromisePartialAgentState { // 从消息历史中取出最新的一条假设是用户输入 const lastMessage state.messages[state.messages.length - 1]; // 假设我们约定用户的最新问题在最后一条HumanMessage里 if (lastMessage instanceof HumanMessage) { const userQuestion lastMessage.content.toString(); // 返回要更新的部分State return { question: userQuestion, // 更新问题字段 toolsUsed: [], // 初始化工具使用记录 }; } // 如果没有找到用户输入返回空对象不更新任何State字段 return {}; }Node函数的精髓它应该是纯净的或者说副作用是可控的。它的主要任务是根据输入State计算输出State的更新。虽然你可以在Node里调用网络IO如数据库查询、API调用但最好把这些副作用封装成“工具”Tool这样更利于测试和复用。我们后面会讲到如何集成工具。2.3 Edge决定流程走向的“路标”Edge边决定了在一个Node执行完毕后下一步应该去哪个Node。这是实现分支、循环的关键。Edge通常是一个函数它检查当前的State然后返回下一个要执行的Node的名称字符串。最简单的Edge是“条件边”Conditional Edge。LangGraph提供了conditionalEdge辅助函数来创建它。import { conditionalEdge } from langchain/langgraph; // 假设我们有两个节点名”search_web“ 和 ”generate_answer“ // 定义一个条件边如果问题需要搜索就去”search_web“否则去”generate_answer“ const routeQuestion (state: AgentState) { // 这里可以是一些复杂的判断逻辑比如用一个小型分类器或者简单的关键词匹配 const needsSearch state.question.includes(“最新”) || state.question.includes(“今天”); return needsSearch ? “search_web” : “generate_answer”; }; // 使用 conditionalEdge 创建边 const conditionalEdge conditionalEdge( [“search_web”, “generate_answer”], // 可能的目标节点列表 routeQuestion // 路由函数 );除了条件边还有固定边从一个节点固定指向另一个节点和入口边定义图的开始节点。通过组合不同的边你就能构建出任意复杂的流程图比如先执行A然后根据A的结果决定走B还是CC执行完后可能又跳回A进行循环直到某个条件满足才结束。3. 构建你的第一个LangGraph应用智能问答助手理论说再多不如动手做一遍。我们来构建一个相对完整的智能问答助手流程图。这个助手能1) 理解用户问题2) 判断是否需要联网搜索3) 需要则搜索并整合信息4) 最终生成回答。3.1 项目初始化与依赖安装首先创建一个新的TypeScript项目并安装核心依赖。mkdir my-langgraph-agent cd my-langgraph-agent npm init -y npm install typescript ts-node types/node --save-dev npm install langchain/langgraph langchain/core langchain/openailangchain/langgraph: LangGraph核心库。langchain/core: LangChain核心消息和工具等基础类型。langchain/openai: OpenAI模型集成我们用它作为LLM。配置tsconfig.json{ “compilerOptions”: { “target”: “ES2020”, “module”: “commonjs”, “lib”: [“ES2020”], “outDir”: “./dist”, “rootDir”: “./src”, “strict”: true, “esModuleInterop”: true, “skipLibCheck”: true, “forceConsistentCasingInFileNames”: true, “resolveJsonModule”: true }, “include”: [“src/**/*”], “exclude”: [“node_modules”] }3.2 定义State与工具在src/state.ts中定义State在src/tools.ts中定义我们可能用到的“工具”。工具是LangChain的概念可以被Node调用。// src/state.ts import { BaseMessage } from “langchain/core/messages”; export interface AssistantState { messages: BaseMessage[]; question: string; // 判断是否需要搜索的结果 shouldSearch?: boolean; // 搜索得到的上下文 searchContext?: string; // 最终答案 finalAnswer?: string; // 错误信息如果有 error?: string; }// src/tools.ts import { tool } from “langchain/core/tools”; // 假设我们有一个模拟的搜索函数 async function mockWebSearch(query: string): Promisestring { // 模拟网络延迟 await new Promise(resolve setTimeout(resolve, 100)); return 这是关于“${query}”的模拟搜索结果。当前时间是${new Date().toLocaleTimeString()}。信息仅供参考。; } // 使用 tool 装饰器定义一个工具 export const webSearchTool tool( async ({ query }: { query: string }) { const result await mockWebSearch(query); return result; }, { name: “web_search”, description: “在互联网上搜索最新信息。当用户问题涉及实时、最新或未知领域知识时使用。”, schema: { type: “object”, properties: { query: { type: “string”, description: “要搜索的查询词”, }, }, required: [“query”], }, } );3.3 实现各个功能节点现在我们来创建图的核心节点函数。我们在src/nodes.ts中实现它们。// src/nodes.ts import { HumanMessage } from “langchain/core/messages”; import { AssistantState } from “./state”; import { webSearchTool } from “./tools”; // 注意这里我们先不引入LLM下一个节点再用 // 节点1提取并分类问题 export async function classifyQuestion( state: AssistantState ): PromisePartialAssistantState { const lastMessage state.messages[state.messages.length - 1]; let userQuestion “”; if (lastMessage instanceof HumanMessage) { userQuestion lastMessage.content.toString(); } // 简单的规则判断是否需要搜索实际应用可用小模型分类 const shouldSearch userQuestion.includes(“最新”) || userQuestion.includes(“今天”) || userQuestion.includes(“2024”) || userQuestion.includes(“新闻”); console.log([Classify] 问题“${userQuestion}” 需要搜索${shouldSearch}); return { question: userQuestion, shouldSearch, }; } // 节点2执行网络搜索如果需要 export async function searchWeb( state: AssistantState ): PromisePartialAssistantState { // 只有需要搜索时才执行 if (!state.shouldSearch) { console.log(“[Search] 跳过搜索环节。”); return { searchContext: “未进行搜索。” }; } console.log([Search] 正在搜索“${state.question}”); try { // 调用我们之前定义的搜索工具 const context await webSearchTool.invoke({ query: state.question }); return { searchContext: context }; } catch (error) { console.error(“[Search] 搜索失败”, error); return { searchContext: “搜索服务暂时不可用。”, error: 搜索失败${error} }; } } // 节点3调用LLM生成最终答案 export async function generateAnswer( state: AssistantState ): PromisePartialAssistantState { console.log([Generate] 正在生成答案上下文长度${state.searchContext?.length || 0}); // 动态引入避免没有API Key时初始化报错 const { ChatOpenAI } await import(“langchain/openai”); const { SystemMessage } await import(“langchain/core/messages”); const llm new ChatOpenAI({ modelName: “gpt-3.5-turbo”, temperature: 0.7, // apiKey: process.env.OPENAI_API_KEY, // 应从环境变量读取 }); // 构建给LLM的提示 let systemPrompt “你是一个有帮助的助手。请根据以下信息回答用户问题。\n”; if (state.searchContext) { systemPrompt [检索到的信息]${state.searchContext}\n; } else { systemPrompt “本次回答未使用实时搜索信息仅基于你的内部知识。\n”; } systemPrompt “请用中文回答保持友好和专业。”; const messages [ new SystemMessage(systemPrompt), ...state.messages, // 包含用户的历史消息 ]; try { const response await llm.invoke(messages); const finalAnswer response.content.toString(); console.log([Generate] 答案生成完毕长度${finalAnswer.length}); return { finalAnswer }; } catch (error) { console.error(“[Generate] LLM调用失败”, error); return { finalAnswer: “抱歉思考过程出现了问题。”, error: LLM调用失败${error} }; } } // 节点4组装最终回复并更新消息历史 export async function formatOutput( state: AssistantState ): PromisePartialAssistantState { const { finalAnswer } state; const { AIMessage } await import(“langchain/core/messages”); if (!finalAnswer) { return { messages: [...state.messages, new AIMessage(“未能生成有效回答。”)] }; } const aiMessage new AIMessage(finalAnswer); // 将AI的回复追加到消息历史中 const updatedMessages [...state.messages, aiMessage]; console.log([Format] 最终回答已添加到消息历史。); return { messages: updatedMessages }; }3.4 组装图并运行最后在src/index.ts中我们将所有节点和边组装成图并运行它。// src/index.ts import { StateGraph, START, END } from “langchain/langgraph”; import { AssistantState } from “./state”; import { classifyQuestion, searchWeb, generateAnswer, formatOutput } from “./nodes”; import { HumanMessage } from “langchain/core/messages”; // 1. 创建一个图构建器并传入我们的State类型 const workflow new StateGraphAssistantState(); // 2. 添加节点 workflow.addNode(“classify_question”, classifyQuestion); workflow.addNode(“search_web”, searchWeb); workflow.addNode(“generate_answer”, generateAnswer); workflow.addNode(“format_output”, formatOutput); // 3. 添加边定义流程 // 起点 - 问题分类 workflow.addEdge(START, “classify_question”); // 问题分类 - 条件边 - 搜索 或 直接生成 workflow.addConditionalEdges( “classify_question”, // 路由函数根据state.shouldSearch决定下一个节点 (state: AssistantState) { return state.shouldSearch ? “search_web” : “generate_answer”; }, // 指定可能的目标节点映射可选但让流程更清晰 { search_web: “search_web”, generate_answer: “generate_answer”, } ); // 搜索节点 - 生成答案节点固定边 workflow.addEdge(“search_web”, “generate_answer”); // 生成答案节点 - 格式化输出节点固定边 workflow.addEdge(“generate_answer”, “format_output”); // 格式化输出节点 - 结束 workflow.addEdge(“format_output”, END); // 4. 编译图 const app workflow.compile(); // 5. 运行图 async function runAgent(question: string) { console.log(\n 用户问题${question} ); const initialState: AssistantState { messages: [new HumanMessage(question)], // 初始状态包含用户问题 question: “”, }; // stream() 方法返回一个异步生成器可以逐步获取每个节点的输出 const stream app.stream(initialState, { subgraphs: true }); for await (const chunk of stream) { // chunk 是一个记录键是节点名值是该节点执行后的状态快照 for (const [nodeName, nodeState] of Object.entries(chunk)) { console.log(\n[节点更新] ${nodeName}:); // 只打印我们关心的部分避免日志过长 if (nodeName “classify_question”) { console.log( - shouldSearch: ${nodeState.shouldSearch}); } else if (nodeName “search_web” nodeState.searchContext) { console.log( - searchContext: ${nodeState.searchContext.substring(0, 100)}...); } else if (nodeName “generate_answer” nodeState.finalAnswer) { console.log( - finalAnswer: ${nodeState.finalAnswer.substring(0, 150)}...); } } } // 获取最终状态 const finalState await app.invoke(initialState); console.log(“\n 最终回答 ); const lastMessage finalState.messages[finalState.messages.length - 1]; console.log(lastMessage.content); console.log(“\n”); } // 运行测试 (async () { await runAgent(“今天的天气怎么样”); await runAgent(“请介绍一下你自己。”); })();运行npx ts-node src/index.ts你会看到控制台打印出整个工作流的执行步骤和中间状态最终输出AI的回答。第一个问题触发了搜索分支第二个问题则直接跳过了搜索。4. 高级特性与实战技巧掌握了基础构建之后我们来看看LangGraph TypeScript版的一些高级特性和能让你事半功倍的实战技巧。4.1 子图Subgraph模块化与复用当你的工作流变得非常复杂时把所有节点都放在一个主图里会难以维护。子图允许你将一部分节点和边打包成一个独立的、可复用的“模块”这个模块在主图中看起来就像一个普通的节点。典型场景一个“研究助手”工作流其中“搜集资料”这个步骤本身可能又包含“学术搜索”、“新闻搜索”、“社交媒体搜索”等多个并行或串行的子任务。我们可以把“搜集资料”做成一个子图。import { StateGraph, START, END } from “langchain/langgraph”; // 假设我们已定义 ResearchState 和若干节点函数 // 1. 先创建一个“资料搜集”子图 const researchSubgraph new StateGraphResearchState() .addNode(“search_academic”, searchAcademicNode) .addNode(“search_news”, searchNewsNode) .addNode(“aggregate_results”, aggregateResultsNode) .addEdge(START, “search_academic”) .addEdge(START, “search_news”) // 并行执行 .addConditionalEdges(“search_academic”, routeToAggregate) .addConditionalEdges(“search_news”, routeToAggregate) .addEdge(“aggregate_results”, END); const compiledResearch researchSubgraph.compile(); // 2. 在主图中将这个编译好的子图作为一个节点添加 const mainWorkflow new StateGraphMainState() .addNode(“understand_task”, understandTaskNode) .addNode(“conduct_research”, compiledResearch) // 关键添加子图作为节点 .addNode(“write_report”, writeReportNode) .addEdge(START, “understand_task”) .addEdge(“understand_task”, “conduct_research”) .addEdge(“conduct_research”, “write_report”) .addEdge(“write_report”, END);优势封装性主图逻辑更清晰。可复用同一个子图可以在不同主图中使用。独立测试子图可以单独编译和测试。4.2 持久化与长期记忆CheckpointerLangGraph一个强大的特性是支持检查点Checkpointer。它可以将工作流的完整状态State持久化到数据库如内存、Redis、PostgreSQL等并允许从任意一个保存的检查点恢复执行。这是实现“长期对话”、“异步任务”、“人类审核中断后继续”等场景的基石。TypeScript版通过MemorySaver提供了一个简单的内存检查点实现。对于生产环境你需要实现自己的BaseCheckpointSaver接口连接到你的数据库。import { MemorySaver } from “langchain/langgraph”; // 创建内存检查点管理器 const memory new MemorySaver(); // 在编译图时传入 checkpointer const app workflow.compile({ checkpointer: memory, }); // 运行图时可以指定一个线程IDthread_id用于标识这次会话或任务 const config { configurable: { thread_id: “user_123_session_1” } }; const initialState { messages: [new HumanMessage(“Hi”)] }; // 第一次调用会创建检查点 const stream1 app.stream(initialState, config); // ... 处理 stream1 // 模拟一段时间后第二次调用如用户发送下一条消息 // 注意这里传入的初始状态会被忽略因为会从检查点恢复 const newMessageState { messages: [new HumanMessage(“What’s next?”)] }; // 使用相同的 thread_idLangGraph会加载上次保存的状态并在此基础上追加新消息然后继续执行 const stream2 app.stream(newMessageState, config);核心机制Checkpointer不仅保存State还保存了图的执行位置。恢复时它能知道上次执行到了哪个节点然后从那里继续。这对于多轮对话机器人、需要长时间运行并可能被中断的自动化流程如客服工单处理来说是必不可少的特性。4.3 与LangChain Agent的深度集成LangGraph的Node可以无缝集成LangChain的Agent。你可以把一个完整的Agent它本身能调用工具、思考作为一个Node嵌入到你的图中。这样你就拥有了“宏流程”图控制“微流程”Agent的能力。import { createReactAgent } from “langchain/langgraph/prebuilt”; import { ChatOpenAI } from “langchain/openai”; import { tool } from “langchain/core/tools”; // 1. 定义一些工具 const calculatorTool tool( ({ a, b }: { a: number; b: number }) a b, { name: “add”, description: “Adds two numbers”, schema: ... } ); const tools [calculatorTool]; // 2. 创建一个React Agent const llm new ChatOpenAI({ temperature: 0 }); const agent createReactAgent({ llm, tools, // 可以自定义Agent的提示词等 }); // 3. 在你的图状态中可能需要一个字段来存储Agent的中间步骤 interface StateWithAgent { messages: BaseMessage[]; agentSteps?: any[]; // 用于存储Agent的思考过程 finalOutput?: string; } // 4. 创建一个Node这个Node内部运行的是LangChain Agent async function runAgentNode(state: StateWithAgent): PromisePartialStateWithAgent { // 将最新的用户消息交给Agent处理 const response await agent.invoke({ messages: state.messages, }); // response 里包含了Agent生成的消息和可能的中间步骤 return { messages: [...state.messages, response.messages[response.messages.length - 1]], agentSteps: response.steps, // 保存步骤供后续节点分析 }; } // 5. 将这个Node加入到你的主图中这种模式非常强大比如你可以设计一个图第一个节点用简单规则做意图分类如果是简单查询直接走快速回答节点如果是复杂任务则路由到“Agent节点”让这个拥有强大工具调用能力的Agent去自由发挥Agent执行完后再进入一个“结果审核节点”进行检查和格式化。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种问题。下面是我踩过的一些坑和总结的经验。5.1 常见问题速查表问题现象可能原因解决方案编译错误Property ‘messages’ is missing…State接口没有定义messages: BaseMessage[]字段。确保State接口中必须包含messages: BaseMessage[]。这是LangGraph的强制要求。运行时错误Cannot read property ‘xxx’ of undefined在Node中访问了State中可能为undefined的字段没有做安全判断。1. 在State定义中使用可选属性?。2. 在Node代码中访问前先检查if (state.someField) { … }。流程没有按预期分支条件边conditionalEdge的路由函数逻辑有误或者返回的节点名与addNode时注册的名称不匹配。1. 在路由函数中多加console.log打印判断逻辑和返回值。2. 检查节点名拼写确保完全一致区分大小写。Stream输出看不到某个节点的更新1. 该节点可能因为条件边被跳过。2. 该节点执行出错但错误被吞掉。3.stream()迭代时只打印了部分状态字段。1. 检查条件边逻辑。2. 在Node函数内部用try-catch包裹并console.error错误。3. 在stream循环中打印完整的chunk对象看看。TypeScript类型报错说State类型不兼容在addNode或addEdge时传入的函数参数类型与StateGraphMyState中定义的泛型不匹配。确保所有节点函数、条件边函数的第一个参数类型都是MyState或它的只读版本。使用统一的State接口导入。检查点Checkpointer没有生效1. 编译图时没有传入checkpointer配置。2. 每次调用stream/invoke时没有传入相同的configurable.thread_id。1. 确认compile({ checkpointer: mySaver })。2. 确保恢复执行时使用了与创建时相同的thread_id。5.2 调试技巧充分利用stream()方法app.stream()返回的异步生成器会产出每个节点执行后的完整状态快照。这是最好的调试工具。你可以像示例中那样遍历并打印每个节点的状态变化清晰地看到数据是如何流动的。为节点函数添加详细日志在每个节点函数的开始和结束以及关键分支处使用console.log输出当前State的相关字段。给日志加上节点名前缀方便过滤。简化复现当遇到复杂问题时尝试创建一个最小的、可复现的示例图。只保留能触发问题的最少节点和边这能帮你快速定位是逻辑错误还是环境配置问题。使用TypeScript严格模式在tsconfig.json中开启“strict”: true。这能帮助你在编码阶段就捕获许多潜在的类型错误比如访问未定义的属性。5.3 性能优化考量State设计要精简State会在每个节点间被序列化/反序列化尤其是在使用远程Checkpointer时。避免在State中存储过大的对象如图片Buffer、长文本。可以考虑只存储引用ID在节点内部再去查询详细数据。节点的幂等性与副作用理想情况下节点函数应该是幂等的给定相同输入产生相同输出。对于有副作用的操作如发送邮件、写入数据库要做好错误处理和重试机制并考虑这些操作失败时如何更新State来反映失败状态。并行执行LangGraph支持并行节点。通过addEdge(START, [“node_a”, “node_b”])可以让node_a和node_b同时开始执行。但要注意它们之间的数据依赖关系只有不依赖彼此输出的节点才适合并行。LLM调用优化LLM调用通常是瓶颈。考虑缓存对相似的LLM提示词和参数进行结果缓存。批处理如果多个节点都需要调用LLM看是否能合并请求。模型选择在不需要很强推理能力的节点如分类、提取使用更小、更快的模型。6. 从Demo到生产架构与部署思考当你用LangGraph做出了一个效果不错的原型后如何将它变成一个稳定、可维护的生产级服务状态存储Checkpointer将内存检查点 (MemorySaver) 替换为持久化存储。实现一个PostgresCheckpointSaver或RedisCheckpointSaver。你需要考虑状态数据的序列化通常用JSON和版本兼容性。错误处理与重试在图层面增加错误处理节点。可以设计一个全局的“错误处理”节点其他节点在抛出特定错误时通过条件边路由到这里进行统一的重试、降级或通知人工。可观测性在生产环境中光靠console.log是不够的。需要将节点的执行时间、输入输出、LLM的Token使用情况等指标发送到监控系统如Prometheus、Datadog。可以在每个节点函数外层包裹一个装饰器来统一收集指标。图的版本管理你的工作流逻辑图的结构可能会迭代。你需要一种机制来管理不同版本的图并能将旧的检查点状态迁移到新版本的图上执行。这通常需要自定义的迁移脚本。API封装将你的LangGraph应用封装成一个标准的Web API如使用Fastify、Express或NestJS。API端点接收用户输入创建或加载对应的thread_id调用图并返回结果。处理好异步、长轮询或WebSocket连接以支持流式响应。测试策略单元测试单独测试每个节点函数模拟输入State断言输出State。集成测试测试整个编译好的图用固定的输入验证输出是否符合预期。端到端测试模拟真实用户对话流测试整个系统。LangGraph TypeScript版为我们提供了一个极其强大的抽象来管理复杂的、有状态的LLM工作流。它强迫我们以“状态流”和“图结构”的方式思考问题这本身就能让系统设计变得更清晰。从简单的线性链到复杂的、带循环和分支的智能体它都能优雅地胜任。
1. 从LangChain到LangGraph为什么我们需要一个新的范式如果你在过去一两年里折腾过LLM应用开发大概率听说过或者用过LangChain。它确实是个好东西帮我们解决了LLM调用、工具集成、记忆管理等一系列繁琐问题让构建一个能聊天的AI应用变得像搭积木一样简单。但不知道你有没有遇到过这样的场景你想构建一个稍微复杂点的流程比如一个能根据用户问题自动决定是去查数据库、调用API还是直接生成答案的客服机器人。用LangChain的Chain和Agent来搞代码写着写着就感觉有点“拧巴”——状态管理分散在各个节点流程控制依赖大量的if-else想要实现个循环或者让某个节点根据条件重复执行更是得费一番功夫。这就是LangGraph诞生的背景。它不是要取代LangChain而是LangChain生态中的一个新成员专门用来解决复杂、有状态的工作流问题。你可以把它理解成从“链式思维”到“图式思维”的升级。LangChain的Chain是线性的一个接一个而LangGraph是图结构的节点Node之间可以有分支、循环、并行。这特别适合那些需要根据中间结果动态决定下一步做什么的应用比如多轮对话、复杂决策、带有审核或修正环节的自动化流程。我最初接触LangGraph的Python版时就被它的清晰和强大吸引了。但作为主要工作在Node.js/TypeScript环境下的开发者我一直在等它的TypeScript版本变得足够成熟。现在这一天已经来了。TypeScript版的LangGraph不仅功能上向Python版看齐而且完美融入了TypeScript的类型系统让你在享受灵活图计算的同时还能获得强大的类型安全和IDE提示。这篇内容我就结合自己从Python迁移到TypeScript的实践经验带你从零开始彻底搞懂LangGraph TypeScript版的核心概念、最佳实践以及那些官方文档里可能没细说的“坑”。2. LangGraph核心三要素State、Node、Edge要玩转LangGraph首先得吃透它的三个核心概念State、Node和Edge。这是构建任何图的基础理解透了后面写代码就是水到渠成。2.1 State工作流的“记忆中枢”State是LangGraph里最核心也最容易让人一开始困惑的概念。它不是一个简单的变量而是一个定义了整个工作流所有共享数据结构的类型。你可以把它想象成游戏里角色的属性面板里面记录了生命值、魔法值、装备、任务进度等所有需要在不同场景节点间传递和修改的信息。在TypeScript里我们用一个接口Interface或类型别名Type Alias来定义State。LangGraph要求这个State必须包含一个特殊的字段messages类型是BaseMessage[]来自langchain/core。这是为了与LangChain的聊天模型兼容。但除此之外你可以自由添加任何字段。import { BaseMessage } from langchain/core/messages; // 定义工作流的状态类型 interface AgentState { // 必需字段消息历史 messages: BaseMessage[]; // 自定义字段用户问题 question: string; // 自定义字段从知识库检索到的上下文 context?: string; // 自定义字段LLM生成的最终答案 answer?: string; // 自定义字段记录调用了哪些工具 toolsUsed: string[]; // 自定义字段控制流程的标记比如是否需要人工审核 needsHumanReview?: boolean; }关键理解这个AgentState接口定义了一个对象“形状”。之后图里的每一个Node节点函数接收到的第一个参数就是当前这个State对象的一个代理Proxy。你在Node函数里修改这个State比如state.answer “某答案”修改的是共享的、全局的状态。这保证了数据在节点间的流动。实操心得在设计State时尽量保持扁平结构避免嵌套过深的对象。因为State会在每个节点被序列化和传递尤其在分布式场景下复杂的嵌套会影响性能。同时为每个字段想好明确的用途避免变成“杂物间”。使用可选属性?来区分哪些是中间过程可能缺失的字段。2.2 Node执行具体任务的“工作单元”Node节点就是一个普通的异步函数它接收两个参数当前的State和一个可选的配置对象。它的职责是执行一段逻辑并返回一个部分State更新。这个“返回部分更新”的概念很重要。你不需要返回完整的State对象只需要返回一个对象里面包含你这次执行想要修改的那些字段。LangGraph会帮你自动合并到全局State中。import { BaseMessage, HumanMessage } from langchain/core/messages; // 一个节点处理用户输入 async function processInput(state: AgentState): PromisePartialAgentState { // 从消息历史中取出最新的一条假设是用户输入 const lastMessage state.messages[state.messages.length - 1]; // 假设我们约定用户的最新问题在最后一条HumanMessage里 if (lastMessage instanceof HumanMessage) { const userQuestion lastMessage.content.toString(); // 返回要更新的部分State return { question: userQuestion, // 更新问题字段 toolsUsed: [], // 初始化工具使用记录 }; } // 如果没有找到用户输入返回空对象不更新任何State字段 return {}; }Node函数的精髓它应该是纯净的或者说副作用是可控的。它的主要任务是根据输入State计算输出State的更新。虽然你可以在Node里调用网络IO如数据库查询、API调用但最好把这些副作用封装成“工具”Tool这样更利于测试和复用。我们后面会讲到如何集成工具。2.3 Edge决定流程走向的“路标”Edge边决定了在一个Node执行完毕后下一步应该去哪个Node。这是实现分支、循环的关键。Edge通常是一个函数它检查当前的State然后返回下一个要执行的Node的名称字符串。最简单的Edge是“条件边”Conditional Edge。LangGraph提供了conditionalEdge辅助函数来创建它。import { conditionalEdge } from langchain/langgraph; // 假设我们有两个节点名”search_web“ 和 ”generate_answer“ // 定义一个条件边如果问题需要搜索就去”search_web“否则去”generate_answer“ const routeQuestion (state: AgentState) { // 这里可以是一些复杂的判断逻辑比如用一个小型分类器或者简单的关键词匹配 const needsSearch state.question.includes(“最新”) || state.question.includes(“今天”); return needsSearch ? “search_web” : “generate_answer”; }; // 使用 conditionalEdge 创建边 const conditionalEdge conditionalEdge( [“search_web”, “generate_answer”], // 可能的目标节点列表 routeQuestion // 路由函数 );除了条件边还有固定边从一个节点固定指向另一个节点和入口边定义图的开始节点。通过组合不同的边你就能构建出任意复杂的流程图比如先执行A然后根据A的结果决定走B还是CC执行完后可能又跳回A进行循环直到某个条件满足才结束。3. 构建你的第一个LangGraph应用智能问答助手理论说再多不如动手做一遍。我们来构建一个相对完整的智能问答助手流程图。这个助手能1) 理解用户问题2) 判断是否需要联网搜索3) 需要则搜索并整合信息4) 最终生成回答。3.1 项目初始化与依赖安装首先创建一个新的TypeScript项目并安装核心依赖。mkdir my-langgraph-agent cd my-langgraph-agent npm init -y npm install typescript ts-node types/node --save-dev npm install langchain/langgraph langchain/core langchain/openailangchain/langgraph: LangGraph核心库。langchain/core: LangChain核心消息和工具等基础类型。langchain/openai: OpenAI模型集成我们用它作为LLM。配置tsconfig.json{ “compilerOptions”: { “target”: “ES2020”, “module”: “commonjs”, “lib”: [“ES2020”], “outDir”: “./dist”, “rootDir”: “./src”, “strict”: true, “esModuleInterop”: true, “skipLibCheck”: true, “forceConsistentCasingInFileNames”: true, “resolveJsonModule”: true }, “include”: [“src/**/*”], “exclude”: [“node_modules”] }3.2 定义State与工具在src/state.ts中定义State在src/tools.ts中定义我们可能用到的“工具”。工具是LangChain的概念可以被Node调用。// src/state.ts import { BaseMessage } from “langchain/core/messages”; export interface AssistantState { messages: BaseMessage[]; question: string; // 判断是否需要搜索的结果 shouldSearch?: boolean; // 搜索得到的上下文 searchContext?: string; // 最终答案 finalAnswer?: string; // 错误信息如果有 error?: string; }// src/tools.ts import { tool } from “langchain/core/tools”; // 假设我们有一个模拟的搜索函数 async function mockWebSearch(query: string): Promisestring { // 模拟网络延迟 await new Promise(resolve setTimeout(resolve, 100)); return 这是关于“${query}”的模拟搜索结果。当前时间是${new Date().toLocaleTimeString()}。信息仅供参考。; } // 使用 tool 装饰器定义一个工具 export const webSearchTool tool( async ({ query }: { query: string }) { const result await mockWebSearch(query); return result; }, { name: “web_search”, description: “在互联网上搜索最新信息。当用户问题涉及实时、最新或未知领域知识时使用。”, schema: { type: “object”, properties: { query: { type: “string”, description: “要搜索的查询词”, }, }, required: [“query”], }, } );3.3 实现各个功能节点现在我们来创建图的核心节点函数。我们在src/nodes.ts中实现它们。// src/nodes.ts import { HumanMessage } from “langchain/core/messages”; import { AssistantState } from “./state”; import { webSearchTool } from “./tools”; // 注意这里我们先不引入LLM下一个节点再用 // 节点1提取并分类问题 export async function classifyQuestion( state: AssistantState ): PromisePartialAssistantState { const lastMessage state.messages[state.messages.length - 1]; let userQuestion “”; if (lastMessage instanceof HumanMessage) { userQuestion lastMessage.content.toString(); } // 简单的规则判断是否需要搜索实际应用可用小模型分类 const shouldSearch userQuestion.includes(“最新”) || userQuestion.includes(“今天”) || userQuestion.includes(“2024”) || userQuestion.includes(“新闻”); console.log([Classify] 问题“${userQuestion}” 需要搜索${shouldSearch}); return { question: userQuestion, shouldSearch, }; } // 节点2执行网络搜索如果需要 export async function searchWeb( state: AssistantState ): PromisePartialAssistantState { // 只有需要搜索时才执行 if (!state.shouldSearch) { console.log(“[Search] 跳过搜索环节。”); return { searchContext: “未进行搜索。” }; } console.log([Search] 正在搜索“${state.question}”); try { // 调用我们之前定义的搜索工具 const context await webSearchTool.invoke({ query: state.question }); return { searchContext: context }; } catch (error) { console.error(“[Search] 搜索失败”, error); return { searchContext: “搜索服务暂时不可用。”, error: 搜索失败${error} }; } } // 节点3调用LLM生成最终答案 export async function generateAnswer( state: AssistantState ): PromisePartialAssistantState { console.log([Generate] 正在生成答案上下文长度${state.searchContext?.length || 0}); // 动态引入避免没有API Key时初始化报错 const { ChatOpenAI } await import(“langchain/openai”); const { SystemMessage } await import(“langchain/core/messages”); const llm new ChatOpenAI({ modelName: “gpt-3.5-turbo”, temperature: 0.7, // apiKey: process.env.OPENAI_API_KEY, // 应从环境变量读取 }); // 构建给LLM的提示 let systemPrompt “你是一个有帮助的助手。请根据以下信息回答用户问题。\n”; if (state.searchContext) { systemPrompt [检索到的信息]${state.searchContext}\n; } else { systemPrompt “本次回答未使用实时搜索信息仅基于你的内部知识。\n”; } systemPrompt “请用中文回答保持友好和专业。”; const messages [ new SystemMessage(systemPrompt), ...state.messages, // 包含用户的历史消息 ]; try { const response await llm.invoke(messages); const finalAnswer response.content.toString(); console.log([Generate] 答案生成完毕长度${finalAnswer.length}); return { finalAnswer }; } catch (error) { console.error(“[Generate] LLM调用失败”, error); return { finalAnswer: “抱歉思考过程出现了问题。”, error: LLM调用失败${error} }; } } // 节点4组装最终回复并更新消息历史 export async function formatOutput( state: AssistantState ): PromisePartialAssistantState { const { finalAnswer } state; const { AIMessage } await import(“langchain/core/messages”); if (!finalAnswer) { return { messages: [...state.messages, new AIMessage(“未能生成有效回答。”)] }; } const aiMessage new AIMessage(finalAnswer); // 将AI的回复追加到消息历史中 const updatedMessages [...state.messages, aiMessage]; console.log([Format] 最终回答已添加到消息历史。); return { messages: updatedMessages }; }3.4 组装图并运行最后在src/index.ts中我们将所有节点和边组装成图并运行它。// src/index.ts import { StateGraph, START, END } from “langchain/langgraph”; import { AssistantState } from “./state”; import { classifyQuestion, searchWeb, generateAnswer, formatOutput } from “./nodes”; import { HumanMessage } from “langchain/core/messages”; // 1. 创建一个图构建器并传入我们的State类型 const workflow new StateGraphAssistantState(); // 2. 添加节点 workflow.addNode(“classify_question”, classifyQuestion); workflow.addNode(“search_web”, searchWeb); workflow.addNode(“generate_answer”, generateAnswer); workflow.addNode(“format_output”, formatOutput); // 3. 添加边定义流程 // 起点 - 问题分类 workflow.addEdge(START, “classify_question”); // 问题分类 - 条件边 - 搜索 或 直接生成 workflow.addConditionalEdges( “classify_question”, // 路由函数根据state.shouldSearch决定下一个节点 (state: AssistantState) { return state.shouldSearch ? “search_web” : “generate_answer”; }, // 指定可能的目标节点映射可选但让流程更清晰 { search_web: “search_web”, generate_answer: “generate_answer”, } ); // 搜索节点 - 生成答案节点固定边 workflow.addEdge(“search_web”, “generate_answer”); // 生成答案节点 - 格式化输出节点固定边 workflow.addEdge(“generate_answer”, “format_output”); // 格式化输出节点 - 结束 workflow.addEdge(“format_output”, END); // 4. 编译图 const app workflow.compile(); // 5. 运行图 async function runAgent(question: string) { console.log(\n 用户问题${question} ); const initialState: AssistantState { messages: [new HumanMessage(question)], // 初始状态包含用户问题 question: “”, }; // stream() 方法返回一个异步生成器可以逐步获取每个节点的输出 const stream app.stream(initialState, { subgraphs: true }); for await (const chunk of stream) { // chunk 是一个记录键是节点名值是该节点执行后的状态快照 for (const [nodeName, nodeState] of Object.entries(chunk)) { console.log(\n[节点更新] ${nodeName}:); // 只打印我们关心的部分避免日志过长 if (nodeName “classify_question”) { console.log( - shouldSearch: ${nodeState.shouldSearch}); } else if (nodeName “search_web” nodeState.searchContext) { console.log( - searchContext: ${nodeState.searchContext.substring(0, 100)}...); } else if (nodeName “generate_answer” nodeState.finalAnswer) { console.log( - finalAnswer: ${nodeState.finalAnswer.substring(0, 150)}...); } } } // 获取最终状态 const finalState await app.invoke(initialState); console.log(“\n 最终回答 ); const lastMessage finalState.messages[finalState.messages.length - 1]; console.log(lastMessage.content); console.log(“\n”); } // 运行测试 (async () { await runAgent(“今天的天气怎么样”); await runAgent(“请介绍一下你自己。”); })();运行npx ts-node src/index.ts你会看到控制台打印出整个工作流的执行步骤和中间状态最终输出AI的回答。第一个问题触发了搜索分支第二个问题则直接跳过了搜索。4. 高级特性与实战技巧掌握了基础构建之后我们来看看LangGraph TypeScript版的一些高级特性和能让你事半功倍的实战技巧。4.1 子图Subgraph模块化与复用当你的工作流变得非常复杂时把所有节点都放在一个主图里会难以维护。子图允许你将一部分节点和边打包成一个独立的、可复用的“模块”这个模块在主图中看起来就像一个普通的节点。典型场景一个“研究助手”工作流其中“搜集资料”这个步骤本身可能又包含“学术搜索”、“新闻搜索”、“社交媒体搜索”等多个并行或串行的子任务。我们可以把“搜集资料”做成一个子图。import { StateGraph, START, END } from “langchain/langgraph”; // 假设我们已定义 ResearchState 和若干节点函数 // 1. 先创建一个“资料搜集”子图 const researchSubgraph new StateGraphResearchState() .addNode(“search_academic”, searchAcademicNode) .addNode(“search_news”, searchNewsNode) .addNode(“aggregate_results”, aggregateResultsNode) .addEdge(START, “search_academic”) .addEdge(START, “search_news”) // 并行执行 .addConditionalEdges(“search_academic”, routeToAggregate) .addConditionalEdges(“search_news”, routeToAggregate) .addEdge(“aggregate_results”, END); const compiledResearch researchSubgraph.compile(); // 2. 在主图中将这个编译好的子图作为一个节点添加 const mainWorkflow new StateGraphMainState() .addNode(“understand_task”, understandTaskNode) .addNode(“conduct_research”, compiledResearch) // 关键添加子图作为节点 .addNode(“write_report”, writeReportNode) .addEdge(START, “understand_task”) .addEdge(“understand_task”, “conduct_research”) .addEdge(“conduct_research”, “write_report”) .addEdge(“write_report”, END);优势封装性主图逻辑更清晰。可复用同一个子图可以在不同主图中使用。独立测试子图可以单独编译和测试。4.2 持久化与长期记忆CheckpointerLangGraph一个强大的特性是支持检查点Checkpointer。它可以将工作流的完整状态State持久化到数据库如内存、Redis、PostgreSQL等并允许从任意一个保存的检查点恢复执行。这是实现“长期对话”、“异步任务”、“人类审核中断后继续”等场景的基石。TypeScript版通过MemorySaver提供了一个简单的内存检查点实现。对于生产环境你需要实现自己的BaseCheckpointSaver接口连接到你的数据库。import { MemorySaver } from “langchain/langgraph”; // 创建内存检查点管理器 const memory new MemorySaver(); // 在编译图时传入 checkpointer const app workflow.compile({ checkpointer: memory, }); // 运行图时可以指定一个线程IDthread_id用于标识这次会话或任务 const config { configurable: { thread_id: “user_123_session_1” } }; const initialState { messages: [new HumanMessage(“Hi”)] }; // 第一次调用会创建检查点 const stream1 app.stream(initialState, config); // ... 处理 stream1 // 模拟一段时间后第二次调用如用户发送下一条消息 // 注意这里传入的初始状态会被忽略因为会从检查点恢复 const newMessageState { messages: [new HumanMessage(“What’s next?”)] }; // 使用相同的 thread_idLangGraph会加载上次保存的状态并在此基础上追加新消息然后继续执行 const stream2 app.stream(newMessageState, config);核心机制Checkpointer不仅保存State还保存了图的执行位置。恢复时它能知道上次执行到了哪个节点然后从那里继续。这对于多轮对话机器人、需要长时间运行并可能被中断的自动化流程如客服工单处理来说是必不可少的特性。4.3 与LangChain Agent的深度集成LangGraph的Node可以无缝集成LangChain的Agent。你可以把一个完整的Agent它本身能调用工具、思考作为一个Node嵌入到你的图中。这样你就拥有了“宏流程”图控制“微流程”Agent的能力。import { createReactAgent } from “langchain/langgraph/prebuilt”; import { ChatOpenAI } from “langchain/openai”; import { tool } from “langchain/core/tools”; // 1. 定义一些工具 const calculatorTool tool( ({ a, b }: { a: number; b: number }) a b, { name: “add”, description: “Adds two numbers”, schema: ... } ); const tools [calculatorTool]; // 2. 创建一个React Agent const llm new ChatOpenAI({ temperature: 0 }); const agent createReactAgent({ llm, tools, // 可以自定义Agent的提示词等 }); // 3. 在你的图状态中可能需要一个字段来存储Agent的中间步骤 interface StateWithAgent { messages: BaseMessage[]; agentSteps?: any[]; // 用于存储Agent的思考过程 finalOutput?: string; } // 4. 创建一个Node这个Node内部运行的是LangChain Agent async function runAgentNode(state: StateWithAgent): PromisePartialStateWithAgent { // 将最新的用户消息交给Agent处理 const response await agent.invoke({ messages: state.messages, }); // response 里包含了Agent生成的消息和可能的中间步骤 return { messages: [...state.messages, response.messages[response.messages.length - 1]], agentSteps: response.steps, // 保存步骤供后续节点分析 }; } // 5. 将这个Node加入到你的主图中这种模式非常强大比如你可以设计一个图第一个节点用简单规则做意图分类如果是简单查询直接走快速回答节点如果是复杂任务则路由到“Agent节点”让这个拥有强大工具调用能力的Agent去自由发挥Agent执行完后再进入一个“结果审核节点”进行检查和格式化。5. 常见问题、调试技巧与性能优化在实际开发中你肯定会遇到各种问题。下面是我踩过的一些坑和总结的经验。5.1 常见问题速查表问题现象可能原因解决方案编译错误Property ‘messages’ is missing…State接口没有定义messages: BaseMessage[]字段。确保State接口中必须包含messages: BaseMessage[]。这是LangGraph的强制要求。运行时错误Cannot read property ‘xxx’ of undefined在Node中访问了State中可能为undefined的字段没有做安全判断。1. 在State定义中使用可选属性?。2. 在Node代码中访问前先检查if (state.someField) { … }。流程没有按预期分支条件边conditionalEdge的路由函数逻辑有误或者返回的节点名与addNode时注册的名称不匹配。1. 在路由函数中多加console.log打印判断逻辑和返回值。2. 检查节点名拼写确保完全一致区分大小写。Stream输出看不到某个节点的更新1. 该节点可能因为条件边被跳过。2. 该节点执行出错但错误被吞掉。3.stream()迭代时只打印了部分状态字段。1. 检查条件边逻辑。2. 在Node函数内部用try-catch包裹并console.error错误。3. 在stream循环中打印完整的chunk对象看看。TypeScript类型报错说State类型不兼容在addNode或addEdge时传入的函数参数类型与StateGraphMyState中定义的泛型不匹配。确保所有节点函数、条件边函数的第一个参数类型都是MyState或它的只读版本。使用统一的State接口导入。检查点Checkpointer没有生效1. 编译图时没有传入checkpointer配置。2. 每次调用stream/invoke时没有传入相同的configurable.thread_id。1. 确认compile({ checkpointer: mySaver })。2. 确保恢复执行时使用了与创建时相同的thread_id。5.2 调试技巧充分利用stream()方法app.stream()返回的异步生成器会产出每个节点执行后的完整状态快照。这是最好的调试工具。你可以像示例中那样遍历并打印每个节点的状态变化清晰地看到数据是如何流动的。为节点函数添加详细日志在每个节点函数的开始和结束以及关键分支处使用console.log输出当前State的相关字段。给日志加上节点名前缀方便过滤。简化复现当遇到复杂问题时尝试创建一个最小的、可复现的示例图。只保留能触发问题的最少节点和边这能帮你快速定位是逻辑错误还是环境配置问题。使用TypeScript严格模式在tsconfig.json中开启“strict”: true。这能帮助你在编码阶段就捕获许多潜在的类型错误比如访问未定义的属性。5.3 性能优化考量State设计要精简State会在每个节点间被序列化/反序列化尤其是在使用远程Checkpointer时。避免在State中存储过大的对象如图片Buffer、长文本。可以考虑只存储引用ID在节点内部再去查询详细数据。节点的幂等性与副作用理想情况下节点函数应该是幂等的给定相同输入产生相同输出。对于有副作用的操作如发送邮件、写入数据库要做好错误处理和重试机制并考虑这些操作失败时如何更新State来反映失败状态。并行执行LangGraph支持并行节点。通过addEdge(START, [“node_a”, “node_b”])可以让node_a和node_b同时开始执行。但要注意它们之间的数据依赖关系只有不依赖彼此输出的节点才适合并行。LLM调用优化LLM调用通常是瓶颈。考虑缓存对相似的LLM提示词和参数进行结果缓存。批处理如果多个节点都需要调用LLM看是否能合并请求。模型选择在不需要很强推理能力的节点如分类、提取使用更小、更快的模型。6. 从Demo到生产架构与部署思考当你用LangGraph做出了一个效果不错的原型后如何将它变成一个稳定、可维护的生产级服务状态存储Checkpointer将内存检查点 (MemorySaver) 替换为持久化存储。实现一个PostgresCheckpointSaver或RedisCheckpointSaver。你需要考虑状态数据的序列化通常用JSON和版本兼容性。错误处理与重试在图层面增加错误处理节点。可以设计一个全局的“错误处理”节点其他节点在抛出特定错误时通过条件边路由到这里进行统一的重试、降级或通知人工。可观测性在生产环境中光靠console.log是不够的。需要将节点的执行时间、输入输出、LLM的Token使用情况等指标发送到监控系统如Prometheus、Datadog。可以在每个节点函数外层包裹一个装饰器来统一收集指标。图的版本管理你的工作流逻辑图的结构可能会迭代。你需要一种机制来管理不同版本的图并能将旧的检查点状态迁移到新版本的图上执行。这通常需要自定义的迁移脚本。API封装将你的LangGraph应用封装成一个标准的Web API如使用Fastify、Express或NestJS。API端点接收用户输入创建或加载对应的thread_id调用图并返回结果。处理好异步、长轮询或WebSocket连接以支持流式响应。测试策略单元测试单独测试每个节点函数模拟输入State断言输出State。集成测试测试整个编译好的图用固定的输入验证输出是否符合预期。端到端测试模拟真实用户对话流测试整个系统。LangGraph TypeScript版为我们提供了一个极其强大的抽象来管理复杂的、有状态的LLM工作流。它强迫我们以“状态流”和“图结构”的方式思考问题这本身就能让系统设计变得更清晰。从简单的线性链到复杂的、带循环和分支的智能体它都能优雅地胜任。
