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): Promise<Partial<AgentState>> { // 从消息历史中取出最新的一条,假设是用户输入 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还是C,C执行完后可能又跳回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/openai@langchain/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): Promise<string> { // 模拟网络延迟 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 ): Promise<Partial<AssistantState>> { 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 ): Promise<Partial<AssistantState>> { // 只有需要搜索时才执行 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 ): Promise<Partial<AssistantState>> { 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 ): Promise<Partial<AssistantState>> { 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 StateGraph<AssistantState>(); // 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 StateGraph<ResearchState>() .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 StateGraph<MainState>() .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 持久化与长期记忆(Checkpointer)
LangGraph一个强大的特性是支持检查点(Checkpointer)。它可以将工作流的完整状态(State)持久化到数据库(如内存、Redis、PostgreSQL等),并允许从任意一个保存的检查点恢复执行。这是实现“长期对话”、“异步任务”、“人类审核中断后继续”等场景的基石。
TypeScript版通过MemorySaver提供了一个简单的内存检查点实现。对于生产环境,你需要实现自己的BaseCheckpointSaver接口,连接到你的数据库。
import { MemorySaver } from “@langchain/langgraph”; // 创建内存检查点管理器 const memory = new MemorySaver(); // 在编译图时传入 checkpointer const app = workflow.compile({ checkpointer: memory, }); // 运行图时,可以指定一个线程ID(thread_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_id,LangGraph会加载上次保存的状态,并在此基础上追加新消息,然后继续执行 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): Promise<Partial<StateWithAgent>> { // 将最新的用户消息交给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时,传入的函数参数类型与StateGraph<MyState>中定义的泛型不匹配。 | 确保所有节点函数、条件边函数的第一个参数类型都是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工作流。它强迫我们以“状态流”和“图结构”的方式思考问题,这本身就能让系统设计变得更清晰。从简单的线性链到复杂的、带循环和分支的智能体,它都能优雅地胜任。