1. 项目概述为什么是 LangGraph TypeScript如果你最近在折腾大语言模型应用尤其是想构建一个能处理复杂、多步骤任务的智能体那你大概率已经听过 LangChain 的大名。但当你真正上手想把一个“用户提问 - 思考 - 行动 - 再思考”的循环流程用代码清晰地表达出来时可能会发现用基础的 LangChain 链来拼接代码会变得有点“面条化”状态管理也让人头疼。这时候LangGraph 就该登场了。简单来说LangGraph 是 LangChain 生态中专门用于构建有状态、多环节工作流的框架。它把整个应用流程抽象成一个“图”节点是执行单元边是流转逻辑。这听起来有点学术但你可以把它想象成乐高说明书每个步骤节点该做什么很清楚箭头边指明了做完这一步接下来该干嘛而且整个模型可以记住之前拼过哪些部分状态。而 TypeScript 版本则是为前端全栈和 Node.js 开发者打开了一扇门让我们能用熟悉的 JavaScript/TypeScript 生态来构建同样强大的 AI 应用。我选择从 Python 转向探索 TS 版核心原因有几个一是团队技术栈统一前后端都用 TS 沟通成本低二是 Vercel 等平台对 Serverless 函数的 TS 支持极好部署方便三是 TS 的类型系统能在开发阶段就帮我们规避很多状态结构上的错误这对于 LangGraph 这种强依赖状态管理的框架来说简直是福音。这次我就带你从零开始拆解一个 LangGraph TypeScript 项目的核心构成并手把手实现一个能自动联网搜索并总结的智能体。2. 核心概念与架构拆解在动手写代码之前我们必须先理解 LangGraph 的几个核心概念。这就像学开车先要明白方向盘、油门、刹车的功能一样理解了它们你才能驾驭自如。2.1 三要素State、Node、EdgeLangGraph 的架构围绕三个核心要素展开它们共同定义了一个工作流。1. State状态这是整个工作流的“记忆中枢”和“共享白板”。所有节点都读取和修改这个状态对象。在 TypeScript 中我们通常用一个 TypeScript 接口或类型来定义它。比如对于一个问答智能体状态可能包括用户的问题、当前的思考、收集到的资料以及最终答案。// 定义状态的结构 interface AgentState { // 用户输入的问题 question: string; // 模型生成的思考过程 reasoning: string; // 从网络搜索得到的信息 searchResults: string[]; // 最终生成的答案 answer: string; // 表示工作流是否应该继续 shouldContinue: boolean; }注意定义状态接口是第一步也是最重要的一步。它直接决定了你的工作流能处理什么信息。务必在设计时考虑周全避免后期频繁修改。2. Node节点节点是实际执行工作的函数。每个节点接收当前的状态执行一些操作比如调用大模型、执行搜索然后返回更新后的状态。在 LangGraph TS 中一个节点就是一个普通的异步函数。// 一个简单的节点函数示例 async function searchNode(state: AgentState): PromisePartialAgentState { console.log(正在搜索: ${state.question}); // 模拟搜索实际中会调用 SerpAPI 或 Tavily 等工具 const results await mockWebSearch(state.question); // 返回要更新到总状态中的部分 return { searchResults: results, reasoning: 已找到 ${results.length} 条相关信息。 }; }关键点在于节点函数返回的是PartialAgentState即状态的一部分。LangGraph 会自动将其与原有状态合并。3. Edge边边决定了工作流的走向。它根据当前状态决定下一个执行哪个节点。这通常通过一个条件判断函数来实现也就是“路由逻辑”。LangGraph 提供了两种特殊的边END表示结束START表示开始。2.2 CompiledStateGraph工作流的引擎当你定义好节点和边之后需要将它们“编译”成一个可执行的工作流这就是CompiledStateGraph的实例。这个编译后的图对象拥有.stream()或.invoke()方法你可以传入初始状态然后它就会按照你定义的逻辑一步步执行下去。与 LangChain 的区别这里可能是很多人困惑的点。LangChain 是一个庞大的工具链集合提供了连接模型、向量库、工具的各种组件。你可以用 LangChain 的LCEL来声明式地组合链。而 LangGraph 更专注于有状态、带循环和条件分支的复杂工作流。你可以把 LangGraph 看作是 LangChain 生态中用于解决特定复杂流程控制问题的“高级模块”。在实践中一个 LangGraph 工作流的节点内部完全可以调用 LangChain 的链、工具或智能体来完成任务。两者是互补而非替代关系。3. 环境准备与项目初始化理论说再多不如动手。我们从一个干净的 TypeScript 项目开始搭建一个具备联网搜索能力的智能问答工作流。3.1 依赖安装与配置首先创建一个新目录并初始化项目mkdir langgraph-ts-agent cd langgraph-ts-agent npm init -y接着安装核心依赖。这里我们不仅需要langchain/langgraph还需要 LangChain 的核心包、OpenAI 的模型集成包这里以 OpenAI 为例以及一个用于模拟搜索的工具包。同时我们需要 TypeScript 和相关的开发依赖。npm install langchain/langgraph langchain/core langchain langchain/openai npm install -D typescript types/node ts-node nodemon然后初始化 TypeScript 配置npx tsc --init修改生成的tsconfig.json确保设置适合 Node.js 环境{ compilerOptions: { target: ES2022, module: commonjs, lib: [ES2022], outDir: ./dist, rootDir: ./src, strict: true, esModuleInterop: true, skipLibCheck: true, forceConsistentCasingInFileNames: true, resolveJsonModule: true }, include: [src/**/*], exclude: [node_modules] }最后在package.json中添加启动脚本scripts: { dev: nodemon --exec ts-node src/index.ts, build: tsc, start: node dist/index.js }3.2 状态与工具定义在src目录下我们开始编写核心代码。首先定义状态和工具。1. 定义状态接口 (src/types.ts):// 定义工作流的全局状态 export interface AgentState { // 原始输入 input: string; // 模型生成的思考 reasoning: string; // 收集到的信息 gatheredInfo: string[]; // 最终输出 finalOutput?: string; // 控制流标志 shouldSearch: boolean; shouldFinish: boolean; }2. 模拟一个搜索工具 (src/tools/mockSearch.ts):在实际项目中你会集成真实的搜索 API如 Tavily、SerpAPI。这里我们模拟一个export async function mockWebSearch(query: string): Promisestring[] { console.log([工具调用] 模拟搜索查询: ${query}); // 模拟网络延迟 await new Promise(resolve setTimeout(resolve, 500)); // 返回模拟结果 return [ 关于“${query}”的百科摘要这是一个模拟的搜索结果A提供了相关的基础概念。, 技术博客文章指出“${query}”的关键点在于其工作流设计。, 社区讨论显示用户常遇到状态管理问题解决方案是清晰定义状态接口。 ]; }实操心得即使在开发阶段也建议尽快接入真实的工具 API。模拟工具会掩盖网络延迟、API 格式错误、令牌限制等问题这些问题在后期集成时可能会集中爆发增加调试难度。4. 构建智能体工作流分步实现现在进入核心环节构建图。我们将创建一个经典的“思考-行动”循环智能体它先判断是否需要搜索需要则搜索并总结不需要则直接回答。4.1 创建节点函数节点是图的基本单元。我们创建四个节点路由判断、思考、搜索、总结。1. 路由节点 (src/nodes/routerNode.ts):这个节点检查状态决定下一步是搜索还是直接生成答案。import { AgentState } from ../types; export async function routerNode(state: AgentState): PromisePartialAgentState { console.log([路由节点] 正在分析问题...); const { input, gatheredInfo } state; // 简单的路由逻辑如果问题包含“最新”、“新闻”、“2024”等关键词或尚未搜集信息则触发搜索 const needsSearch /(最新|新闻|最近|202[4-9]|如何安装|步骤)/.test(input) || gatheredInfo.length 0; return { reasoning: 问题分析“${input}”。需要联网搜索吗${needsSearch ? 是 : 否}, shouldSearch: needsSearch, // 如果不需要搜索直接准备结束 shouldFinish: !needsSearch }; }2. 思考节点 (src/nodes/thinkNode.ts):这个节点调用大语言模型根据已有信息进行思考。这里我们需要初始化模型。import { ChatOpenAI } from langchain/openai; import { HumanMessage } from langchain/core/messages; import { AgentState } from ../types; // 初始化模型建议从环境变量读取 API Key const llm new ChatOpenAI({ modelName: gpt-3.5-turbo, temperature: 0.2, // 低温度使输出更稳定 }); export async function thinkNode(state: AgentState): PromisePartialAgentState { console.log([思考节点] 调用模型进行思考...); const { input, gatheredInfo, reasoning } state; const prompt 你是一个分析助手。 当前问题${input} 已有的思考记录${reasoning} 已搜集到的信息${gatheredInfo.join(\n)} 请基于以上内容对问题进行分析并生成一段简洁的思考过程。重点是指出已知信息和缺失信息。 思考过程; try { const response await llm.invoke([new HumanMessage(prompt)]); const newReasoning response.content as string; return { reasoning: ${reasoning}\n模型思考${newReasoning} }; } catch (error) { console.error(思考节点调用模型失败, error); return { reasoning: ${reasoning}\n模型思考失败将尝试直接搜索。, shouldSearch: true // 出错时转为搜索 }; } }注意事项模型调用是异步操作且可能失败。务必做好错误处理并考虑在失败时如何优雅地降级或改变工作流路径例如这里我们选择触发搜索。3. 搜索节点 (src/nodes/searchNode.ts):调用我们之前定义的模拟搜索工具。import { AgentState } from ../types; import { mockWebSearch } from ../tools/mockSearch; export async function searchNode(state: AgentState): PromisePartialAgentState { console.log([搜索节点] 开始执行网络搜索...); const { input } state; try { const results await mockWebSearch(input); return { gatheredInfo: results, reasoning: ${state.reasoning}\n已完成搜索获得 ${results.length} 条结果。 }; } catch (error) { console.error(搜索节点执行失败, error); return { reasoning: ${state.reasoning}\n搜索失败将使用已有信息继续。 }; } }4. 总结生成节点 (src/nodes/generateNode.ts):这是最后一个节点综合所有信息生成最终答案。import { ChatOpenAI } from langchain/openai; import { HumanMessage } from langchain/core/messages; import { AgentState } from ../types; const llm new ChatOpenAI({ modelName: gpt-3.5-turbo, temperature: 0.7 }); export async function generateNode(state: AgentState): PromisePartialAgentState { console.log([生成节点] 综合信息生成最终答案...); const { input, gatheredInfo, reasoning } state; const prompt 请扮演一个专业的助手回答用户的问题。 用户问题${input} 以下是相关的背景信息和思考过程 ${reasoning} 以下是搜集到的具体信息 ${gatheredInfo.join(\n---\n)} 请生成一个友好、准确、完整的最终答案。如果信息不足请诚实说明。 最终答案; try { const response await llm.invoke([new HumanMessage(prompt)]); return { finalOutput: response.content as string, shouldFinish: true // 生成答案后标志工作流结束 }; } catch (error) { console.error(生成节点调用模型失败, error); return { finalOutput: 抱歉生成答案时出现错误。, shouldFinish: true }; } }4.2 组装与编译图所有零件准备好了现在在src/graph.ts中把它们组装起来。import { StateGraph, START, END } from langchain/langgraph; import { AgentState } from ./types; import { routerNode } from ./nodes/routerNode; import { thinkNode } from ./nodes/thinkNode; import { searchNode } from ./nodes/searchNode; import { generateNode } from ./nodes/generateNode; // 1. 创建工作流图并传入我们定义的状态结构 const workflow new StateGraphAgentState({ // 状态通道定义对应我们的接口 channels: { input: { value: }, reasoning: { value: }, gatheredInfo: { value: [] }, finalOutput: { value: undefined }, shouldSearch: { value: false }, shouldFinish: { value: false }, }, }); // 2. 添加节点 workflow.addNode(router, routerNode); workflow.addNode(think, thinkNode); workflow.addNode(search, searchNode); workflow.addNode(generate, generateNode); // 3. 添加边定义流程 // 从开始到路由节点 workflow.addEdge(START, router); // 从路由节点出发的条件边 workflow.addConditionalEdges(router, async (state: AgentState) { // 路由节点的输出决定了下一步 if (state.shouldFinish) { // 如果不需要搜索直接去生成答案但通常我们会先思考一下 return think; } if (state.shouldSearch) { // 需要搜索则先去思考一下搜索什么 return think; } // 默认情况理论上不会走到这里 return think; }); // 从思考节点出发的条件边 workflow.addConditionalEdges(think, async (state: AgentState) { if (state.shouldSearch) { return search; // 需要搜索则去搜索 } else { return generate; // 不需要搜索直接生成答案 } }); // 搜索完成后进入生成节点 workflow.addEdge(search, generate); // 生成节点完成后结束工作流 workflow.addEdge(generate, END); // 4. 编译图得到可执行的工作流实例 export const agentGraph workflow.compile();这段代码构建了一个清晰的流程START - router - think - (search - generate 或 generate) - END。addConditionalEdges是关键它实现了基于状态的条件分支。4.3 运行与测试最后在src/index.ts中我们初始化状态并运行这个工作流。import { agentGraph } from ./graph; import { AgentState } from ./types; async function main() { console.log( LangGraph TypeScript 智能体启动 \n); // 测试用例1需要搜索的问题 const initialState1: AgentState { input: LangGraph 在 2024 年有什么最新进展, reasoning: , gatheredInfo: [], finalOutput: undefined, shouldSearch: false, shouldFinish: false, }; console.log(输入问题${initialState1.input}); console.log(开始执行工作流...\n); // 使用 .stream() 可以观察到每一步的中间状态适合调试 const stream agentGraph.stream(initialState1, { streamMode: values }); for await (const chunk of stream) { console.log(当前节点输出状态快照, { reasoning: chunk.reasoning?.slice(-100), // 只看最后一部分 gatheredInfoLength: chunk.gatheredInfo?.length, shouldSearch: chunk.shouldSearch, shouldFinish: chunk.shouldFinish, }); console.log(---); } // 获取最终状态 const finalState await agentGraph.invoke(initialState1); console.log(\n 工作流执行完毕 ); console.log(最终答案); console.log(finalState.finalOutput); console.log(\n完整思考过程); console.log(finalState.reasoning); // 测试用例2不需要搜索的简单问题 console.log(\n\n 测试用例2简单问题 ); const initialState2: AgentState { input: LangGraph 是什么, reasoning: , gatheredInfo: [LangGraph 是用于构建有状态多步骤工作流的框架。], // 假设已有信息 finalOutput: undefined, shouldSearch: false, shouldFinish: false, }; const finalState2 await agentGraph.invoke(initialState2); console.log(问题${initialState2.input}); console.log(答案${finalState2.finalOutput}); } main().catch(console.error);运行npm run dev你将看到工作流一步步执行打印出路由决策、思考内容、搜索动作最终生成答案。通过两个测试用例你能清晰地看到基于shouldSearch标志的不同执行路径。5. 高级特性与实战技巧掌握了基础构建后我们来看看如何让这个智能体更强大、更健壮。5.1 实现长期记忆与状态持久化上面的例子中状态只在单次调用内有效。在实际应用中如聊天机器人你需要让智能体记住之前的对话。这可以通过持久化状态来实现。思路在每次图执行完毕后将最终状态保存到数据库如 Redis、PostgreSQL。下次用户发起新对话时先加载历史状态将新问题作为input合并进去再运行图。// 伪代码示例带有记忆的调用 interface ChatSession { sessionId: string; state: AgentState; // 保存完整的 AgentState } async function runAgentWithMemory(sessionId: string, newInput: string) { // 1. 从数据库加载历史状态 const history await db.getSession(sessionId); const baseState: AgentState history?.state || getInitialState(); // 2. 将新输入叠加到历史状态上注意可能需要清空 finalOutput 等字段 const runState: AgentState { ...baseState, input: newInput, finalOutput: undefined, // 清除上一次的输出 shouldFinish: false, // 重置控制标志 }; // 3. 运行图 const newState await agentGraph.invoke(runState); // 4. 将更新后的状态保存回数据库 await db.saveSession(sessionId, { state: newState }); return newState.finalOutput; }实操心得状态持久化时要小心处理“控制标志”如shouldFinish和“临时字段”。最好在每次运行前将这些字段重置为初始值避免上一次的运行结果干扰本次逻辑。5.2 子图Subgraph与模块化对于复杂智能体你可以将一部分功能如一个完整的搜索-评估流程封装成一个子图。子图本身也是一个CompiledStateGraph可以作为主图的一个节点被调用。这极大地提升了代码的复用性和可维护性。// 假设我们有一个封装好的“研究”子图 import { researchSubgraph } from ./subgraphs/research; // 在主图中可以像添加普通节点一样添加子图 workflow.addNode(deepResearch, researchSubgraph.getEntryNode()); // 需要获取子图的入口节点 // 然后像普通节点一样连接边 workflow.addEdge(think, deepResearch); workflow.addConditionalEdges(deepResearch, (state) { // 根据子图执行后的状态决定下一步 return state.researchComplete ? generate : think; });子图内部的复杂性被隐藏起来主图逻辑得以保持清晰。5.3 流式输出与中断处理.stream()方法不仅用于调试更是实现流式响应的关键。对于需要长时间运行的工作流如深度研究你可以将中间状态如思考过程、搜索到的条目实时推送给前端。// 流式调用示例 const stream agentGraph.stream(initialState, { streamMode: messages }); // 或 values for await (const chunk of stream) { // chunk 可能包含节点名、输出值等信息 if (chunk.node think) { // 将思考过程片段发送给客户端 websocket.send(JSON.stringify({ type: thinking, content: chunk.output.reasoning })); } if (chunk.node generate) { // 将生成的答案片段发送给客户端 websocket.send(JSON.stringify({ type: answer_chunk, content: chunk.output.finalOutput })); } }关于compiledStateGraph.stream()如何终止这是一个常见问题。如果工作流进入了一个长循环或你想在用户取消时中断目前的 LangGraph TS 版本没有提供直接的abort()方法。一个实用的方案是使用AbortController。const controller new AbortController(); const signal controller.signal; // 在节点函数中检查中断信号 async function someLongRunningNode(state: AgentState): PromisePartialAgentState { if (signal.aborted) { console.log(任务被用户取消); return { shouldFinish: true }; // 提前结束 } // ... 正常逻辑 } // 用户取消时调用 controller.abort();另一种更全局的方法是在调用invoke或遍历stream的外部设置一个超时或中断检查点。6. 常见问题、调试技巧与性能优化在实际开发中你会遇到各种问题。这里记录了一些典型坑点和解决思路。6.1 状态更新不符合预期问题节点返回了数据但状态没有正确合并。排查检查返回类型确保节点函数返回的是PartialAgentState且属性名与状态接口完全一致。检查合并逻辑LangGraph 默认是浅合并。如果你的状态中有数组或对象并希望追加而非替换需要手动处理。// 错误这会替换整个数组 // return { gatheredInfo: [new item] }; // 正确合并新条目 return { gatheredInfo: [...state.gatheredInfo, new item] };使用日志在每个节点的开始和结束打印状态这是最直接的调试方式。6.2 工作流陷入死循环问题智能体在“思考-搜索-再思考”中无限循环。原因条件边addConditionalEdges的逻辑有缺陷或者状态中的控制标志如shouldFinish没有被正确设置。解决在路由逻辑中设置“最大循环次数”或“超时时间”。在状态中增加iterationCount字段每次循环递增并在条件边中检查是否超过阈值。确保generateNode等终止节点必须将shouldFinish设为true。6.3 性能瓶颈与优化并行节点执行LangGraph 默认是顺序执行。如果节点间没有依赖可以探索使用addNode时配置并行执行但这需要仔细设计状态划分避免竞争条件。模型调用优化这是主要的耗时和成本来源。缓存对相同的提示词进行缓存可以使用LangChain的InMemoryCache或集成RedisCache。批处理如果可能将多个独立的问题批量发送给模型 API。选择合适模型对于路由、分类等简单任务使用小模型如gpt-3.5-turbo对于最终生成再用大模型如GPT-4。减少状态大小避免在状态中存储过大的中间数据如完整的网页 HTML。只存储提炼后的文本或摘要。6.4 与现有框架集成与 FastAPI/Express 集成将编译好的agentGraph实例封装成一个 API 端点即可。注意处理好异步请求和错误返回。与 Dify 等低代码平台结合Dify 擅长工作流编排和前端界面。你可以将 LangGraph 构建的复杂智能体作为 Dify 的一个“自定义工具”或通过 API 调用来集成。LangGraph 负责核心的、需要复杂状态管理的推理逻辑Dify 负责提供对话界面、知识库检索等外围功能。两者可以很好地协同。7. 从入门到生产下一步规划当你完成了第一个可运行的 LangGraph TS 智能体后可以考虑以下方向来深化和产品化工具集成用真实的工具替换模拟工具。集成langchain/community中的工具如SerpAPI、TavilySearchResults、WikipediaAPIWrapper让智能体真正拥有“手和脚”。复杂路由实现更智能的路由逻辑例如基于大语言模型来判断下一步行动而不仅仅是关键词匹配。可观测性加入详细的日志记录如Pino、Winston和监控如OpenTelemetry追踪每个节点的执行时间、令牌消耗和成功率。测试与评估为你的工作流编写单元测试和集成测试。使用LangSmithLangChain 官方的跟踪平台来调试和评估不同提示词、模型对最终结果的影响。部署将你的智能体部署为 Vercel Serverless Function、AWS Lambda 或 Docker 容器。注意管理环境变量尤其是 API Keys和冷启动问题。LangGraph TypeScript 将复杂的多步推理流程变得清晰、可维护。它要求你更结构化地思考 AI 应用这种思考方式本身比掌握任何一个具体 API 都更有价值。刚开始可能会觉得有些繁琐但当你需要修改一个复杂智能体的行为时你会庆幸自己的代码是建立在“图”这个清晰的概念之上而不是一团混乱的回调函数里。
