从手写 API 调用升级到工程化 AI 应用框架。LangChain.js 解决怎么调用模型LangGraph 解决怎么编排 AI 流程。第一章直接用 fetch 调用 API 没问题但项目一复杂就会遇到几个痛点每次都要手写消息格式、处理流式解析、管理对话历史多步骤的 AI 流程先分析、再检索、再生成要自己串联换个模型要改好几处代码复杂的条件分支逻辑写起来乱LangChain.js 解决前两个问题LangGraph 解决后两个。原始写法fetch → 解析 → 手动拼接历史 → 再 fetch → ... LangChain.jsmodel.invoke() → 链式组合 → 自动管理历史 LangGraph节点 边 条件路由 → 状态机驱动的 AI 流程3.2 LangChain.js 核心用法3.2.1 模型初始化import { ChatOpenAI } from langchain/openai const model new ChatOpenAI({ model: deepseek-chat, apiKey: process.env.DEEPSEEK_API_KEY, configuration: { baseURL: https://api.deepseek.com/v1 }, temperature: 0.7, })ChatOpenAI是 LangChain.js 对 OpenAI 兼容接口的封装。DeepSeek、Qwen、通义千问只需改model和baseURL代码其他地方不用动。3.2.2 四种调用方式import { HumanMessage, SystemMessage, AIMessage } from langchain/core/messages // 1. invoke一次性调用返回 AIMessage 对象 const res await model.invoke([ new SystemMessage(你是 Vue3 技术专家), new HumanMessage(ref 和 reactive 的区别), ]) console.log(res.content) // 字符串内容 console.log(res._getType()) // ai // 2. stream流式调用返回 AsyncIterable const stream await model.stream([new HumanMessage(解释 Vue3 响应式原理)]) for await (const chunk of stream) { process.stdout.write(chunk.content) // 逐 token 输出 } // 3. 并发调用比 batch 更灵活推荐用这个 const questions [defineProps 怎么用, useEffect 和 useLayoutEffect 区别] const responses await Promise.all( questions.map(q model.invoke([new HumanMessage(q)])) ) // 4. 临时修改配置不影响原模型 const preciseModel model.bind({ temperature: 0 }) const creativeModel model.bind({ temperature: 1.2 })3.2.3 多轮对话LangChain.js 的消息类型本身就是多轮对话的载体把历史追加进去就行const history [] async function chat(userInput) { history.push(new HumanMessage(userInput)) const res await model.invoke([ new SystemMessage(你是前端开发导师记住学生的学习进度。), ...history, // 把完整历史传入 ]) history.push(new AIMessage(res.content)) // AI 回复也存入历史 return res.content } await chat(我是前端新手刚学完 HTML 和 CSS) await chat(我想学 JavaScript从哪里开始) const r await chat(我之前说过我的基础是什么来着) // 模型能记住3.2.3 多轮对话LangChain.js 的消息类型本身就是多轮对话的载体把历史追加进去就行const history [] async function chat(userInput) { history.push(new HumanMessage(userInput)) const res await model.invoke([ new SystemMessage(你是前端开发导师记住学生的学习进度。), ...history, // 把完整历史传入 ]) history.push(new AIMessage(res.content)) // AI 回复也存入历史 return res.content } await chat(我是前端新手刚学完 HTML 和 CSS) await chat(我想学 JavaScript从哪里开始) const r await chat(我之前说过我的基础是什么来着) // 模型能记住3.3 ChatPromptTemplate 提示词模板3.3.1 基础用法import { ChatPromptTemplate } from langchain/core/prompts const prompt ChatPromptTemplate.fromMessages([ [system, 你是一位{role}擅长{skill}。], [human, {question}], ]) // formatMessages填入变量 → 得到可直接传给模型的 messages 数组 const messages await prompt.formatMessages({ role: 资深前端架构师, skill: Vue3 和性能优化, question: 大型 Vue3 项目应该怎么做状态管理, }) const res await model.invoke(messages)3.3.2 模板复用同一个模板不同的变量——适合批量审查、翻译、格式化等任务const reviewPrompt ChatPromptTemplate.fromMessages([ [system, 你是{lang}代码审查专家检查{aspects} 输出 JSON{ score: number, issues: string[], suggestions: string[] }], [human, 审查\n{lang}\n{code}\n], ]) // 复用同一模板并发审查不同代码 const [vueResult, reactResult] await Promise.all([ model.invoke(await reviewPrompt.formatMessages({ lang: Vue3, aspects: 内存泄漏、生命周期管理, code: vueCode, })), model.invoke(await reviewPrompt.formatMessages({ lang: React, aspects: 性能问题、Hook 使用规范, code: reactCode, })), ])3.3.3 partial 预填变量partial()预先填入部分变量生成一个新模板。适合不同模块共用基础人设但每次的问题不同const basePrompt ChatPromptTemplate.fromMessages([ [system, 你是{company}的{role}用{tone}的语气回答。], [human, {question}], ]) // 预填固定的部分生成专属模板 const csPrompt basePrompt.partial({ company: 极速购电商平台, role: 客服助手, tone: 热情友好, }) const techPrompt basePrompt.partial({ company: 极速购电商平台, role: 技术支持工程师, tone: 专业严谨, }) // 使用时只需填剩余变量 const r1 await model.invoke(await csPrompt.formatMessages({ question: 我的订单什么时候发货 })) const r2 await model.invoke(await techPrompt.formatMessages({ question: 为什么接口返回 401 }))3.4 LCEL 链式调用LCELLangChain Expression Language用.pipe()把多个步骤串联成链每一步是一个可组合的 Runnable。3.4.1 最简单的链import { StringOutputParser } from langchain/core/output_parsers // prompt → model → 字符串解析器 const chain prompt.pipe(model).pipe(new StringOutputParser()) // invoke 传入的是 prompt 的变量 const result await chain.invoke({ question: Teleport 组件有什么用 }) console.log(typeof result) // string不是 AIMessage 对象StringOutputParser把AIMessage转成纯字符串后续步骤不用再.content取值。3.4.2 顺序链多步处理import { RunnableSequence } from langchain/core/runnables const parser new StringOutputParser() // 第一步分析需求提取功能点 const analyzeChain ChatPromptTemplate.fromMessages([ [system, 你是需求分析师提取核心功能点每点一行不超过 5 个。], [human, 需求{requirement}], ]).pipe(model).pipe(parser) // 第二步根据功能点生成组件列表 const componentChain ChatPromptTemplate.fromMessages([ [system, 你是 Vue3 架构师根据功能点列出需要的组件格式组件名作用。], [human, 功能点{features}], ]).pipe(model).pipe(parser) // RunnableSequence前一步输出自动传入下一步 const pipeline RunnableSequence.from([ { features: analyzeChain, // analyzeChain 的结果 → features requirement: (input) input.requirement, }, componentChain, ]) const result await pipeline.invoke({ requirement: 电商后台商品管理、订单管理、数据统计看板, })3.4.3 并行链同时执行多个任务import { RunnableParallel } from langchain/core/runnables const makeChain (systemPrompt) ChatPromptTemplate.fromMessages([ [system, systemPrompt], [human, {topic}], ]).pipe(model).pipe(new StringOutputParser()) // 三条链同时执行结果合并成一个对象 const parallelChains RunnableParallel.from({ pros: makeChain(列出这个方案的 3 个优点每点一行), cons: makeChain(列出这个方案的 3 个缺点每点一行), alternatives: makeChain(列出 2-3 个替代方案简短说明各自适用场景), }) const result await parallelChains.invoke({ topic: 用 Pinia 做 Vue3 全局状态管理 }) console.log(result.pros) // 优点 console.log(result.cons) // 缺点 console.log(result.alternatives) // 替代方案3.4.4 链的健壮性配置// 失败自动重试 const reliableModel model.withRetry({ stopAfterAttempt: 3, onFailedAttempt: (err) console.log(重试中... ${err.message}), }) // 主模型失败时切到备用模型 const modelWithFallback model.withFallbacks([backupModel]) // 链上的 stream 和普通调用用法一样 const chain prompt.pipe(model).pipe(new StringOutputParser()) const stream await chain.stream({ question: Vite 比 Webpack 快在哪 }) for await (const chunk of stream) { process.stdout.write(chunk) }3.5 会话记忆管理3.5.1 RunnableWithMessageHistoryLangChain.js 内置的记忆管理方案自动把历史注入到链里不用手动维护history数组import { InMemoryChatMessageHistory } from langchain/core/chat_history import { RunnableWithMessageHistory } from langchain/core/runnables import { MessagesPlaceholder } from langchain/core/prompts // 多个 session 的历史存储生产换 Redis 或数据库 const sessionHistories {} function getHistory(sessionId) { if (!sessionHistories[sessionId]) { sessionHistories[sessionId] new InMemoryChatMessageHistory() } return sessionHistories[sessionId] } const prompt ChatPromptTemplate.fromMessages([ [system, 你是 Vue3 技术导师记住每位学生的学习进度。], new MessagesPlaceholder(history), // 历史消息自动注入到这里 [human, {input}], ]) const chain prompt.pipe(model).pipe(new StringOutputParser()) const chainWithMemory new RunnableWithMessageHistory({ runnable: chain, getMessageHistory: getHistory, inputMessagesKey: input, historyMessagesKey: history, }) // 每次调用传 sessionId 区分不同用户互相隔离 const aliceConfig { configurable: { sessionId: alice } } const bobConfig { configurable: { sessionId: bob } } await chainWithMemory.invoke({ input: 我刚开始学 Vue3 }, aliceConfig) const r await chainWithMemory.invoke({ input: 我上次说我在学什么 }, aliceConfig) // r 里模型能正确回答在学 Vue33.5.2 滑动窗口防止上下文超长对话轮次多了历史记录会超过模型的上下文限制。滑动窗口是最简单的应对方案class SlidingWindowChat { constructor({ systemPrompt, maxTokens 3000 }) { this.systemPrompt systemPrompt this.history [] this.maxTokens maxTokens } // 粗略估算 token 数中文 0.6 token/字 estimateTokens(messages) { return messages.reduce( (sum, m) sum Math.ceil((m.content?.length ?? 0) * 0.6), 0 ) } // 超出限制时删除最早的一对消息user assistant 成对删 trimToFit() { while ( this.history.length 2 this.estimateTokens(this.history) this.maxTokens ) { this.history.splice(0, 2) } } async chat(userInput) { this.history.push(new HumanMessage(userInput)) this.trimToFit() const res await model.invoke([ new SystemMessage(this.systemPrompt), ...this.history, ]) this.history.push(new AIMessage(res.content)) return res.content } } const chat new SlidingWindowChat({ systemPrompt: 你是前端助手, maxTokens: 2000 })3.6 LangGraph 核心概念LangGraph 把 AI 流程建模成有向图节点Node执行具体工作的函数调用模型、查数据库、调用 API边Edge节点之间的连接决定执行顺序状态State贯穿整个图的共享数据每个节点读取并更新条件边Conditional Edge根据当前状态动态决定走哪个节点3.6.1 StateGraph 基础import { StateGraph, END, START, Annotation, messagesStateReducer } from langchain/langgraph // 第一步定义状态结构 const GraphState Annotation.Root({ // messages 使用内置 reducer新消息追加到数组末尾 messages: Annotation({ reducer: messagesStateReducer, default: () [], }), // 替换型每次更新直接覆盖旧值 intent: Annotation({ reducer: (_, newVal) newVal, default: () , }), // 累加型自定义 reducer新值追加到数组 logs: Annotation({ reducer: (existing, newVal) [...existing, ...newVal], default: () [], }), }) // 第二步定义节点函数 // 接收完整的 state返回需要更新的字段只写变化的不变的不用写 async function chatNode(state) { const res await model.invoke([ new SystemMessage(你是前端助手), ...state.messages, ]) return { messages: [res] } // messages reducer 会把 res 追加进去 } // 第三步构建图 const graph new StateGraph(GraphState) .addNode(chat, chatNode) .addEdge(START, chat) .addEdge(chat, END) .compile() // 第四步运行 const result await graph.invoke({ messages: [new HumanMessage(Vue3 的 Teleport 是什么)], }) const lastMsg result.messages[result.messages.length - 1] console.log(lastMsg.content)3.6.2 多节点顺序图把复杂任务拆成多个节点每个节点专注一件事const State Annotation.Root({ userInput: Annotation({ reducer: (_, n) n, default: () }), analysis: Annotation({ reducer: (_, n) n, default: () }), solution: Annotation({ reducer: (_, n) n, default: () }), codeExample: Annotation({ reducer: (_, n) n, default: () }), }) // 节点1分析问题类型 async function analyzeNode(state) { const res await model.invoke([ new SystemMessage(判断问题类型性能/逻辑/语法/架构一句话输出。), new HumanMessage(state.userInput), ]) return { analysis: res.content } } // 节点2给出解决思路 async function solutionNode(state) { const res await model.invoke([ new SystemMessage(给出简洁解决思路不超过 3 步。), new HumanMessage(问题${state.userInput}\n类型${state.analysis}), ]) return { solution: res.content } } // 节点3生成代码示例 async function codeNode(state) { const res await model.invoke([ new SystemMessage(根据解决方案写代码示例15行以内。), new HumanMessage(state.solution), ]) return { codeExample: res.content } } const graph new StateGraph(State) .addNode(analyze, analyzeNode) .addNode(solution, solutionNode) .addNode(code, codeNode) .addEdge(START, analyze) .addEdge(analyze, solution) .addEdge(solution, code) .addEdge(code, END) .compile() const result await graph.invoke({ userInput: Vue3 列表渲染 1000 条数据时页面卡顿, })3.6.3 条件路由动态决定流程条件路由是 LangGraph 最核心的特性——让 AI 自己决定流程走向import { z } from zod const State Annotation.Root({ messages: Annotation({ reducer: messagesStateReducer, default: () [] }), questionType: Annotation({ reducer: (_, n) n, default: () }), answer: Annotation({ reducer: (_, n) n, default: () }), }) // 意图分类节点 const ClassifySchema z.object({ type: z.enum([code_help, concept, resource]), }) async function classifyNode(state) { const lastMsg state.messages[state.messages.length - 1] const classifyModel model.withStructuredOutput(ClassifySchema) const result await classifyModel.invoke([ new SystemMessage(判断前端问题的类型 - code_help需要写/调试代码 - concept解释概念或原理 - resource推荐学习资料), new HumanMessage(lastMsg.content), ]) return { questionType: result.type } } // 三个处理节点各有专注方向 async function codeHelpNode(state) { /* 生成代码解决方案 */ } async function conceptNode(state) { /* 解释概念举例子 */ } async function resourceNode(state) { /* 推荐学习资源 */ } // 路由函数根据 state 返回下一个节点的名称 function routeQuestion(state) { const map { code_help: code_help, concept: concept, resource: resource } return map[state.questionType] ?? concept } const graph new StateGraph(State) .addNode(classify, classifyNode) .addNode(code_help, codeHelpNode) .addNode(concept, conceptNode) .addNode(resource, resourceNode) .addEdge(START, classify) // addConditionalEdgesclassify 执行完后调用 routeQuestion根据返回值跳转 .addConditionalEdges(classify, routeQuestion, { code_help: code_help, concept: concept, resource: resource, }) .addEdge(code_help, END) .addEdge(concept, END) .addEdge(resource, END) .compile() // 测试 const r await graph.invoke({ messages: [new HumanMessage(Vue3 中 v-for 和 v-if 同时使用时哪个优先级更高)], }) console.log(路由到, r.questionType) // concept console.log(回答, r.answer)3.6.4 带循环的图生成 → 检查 → 修正LangGraph 支持循环节点可以指回之前的节点实现自我修正的模式const ReviewState Annotation.Root({ requirement: Annotation({ reducer: (_, n) n, default: () }), code: Annotation({ reducer: (_, n) n, default: () }), review: Annotation({ reducer: (_, n) n, default: () }), attempts: Annotation({ reducer: (_, n) n, default: () 0 }), passed: Annotation({ reducer: (_, n) n, default: () false }), }) async function generateCodeNode(state) { const res await model.invoke([ new SystemMessage(你是 Vue3 工程师生成符合要求的组件代码。), new HumanMessage(需求${state.requirement} ${state.review ? 上次审查问题${state.review}请修正。 : }), ]) return { code: res.content, attempts: state.attempts 1 } } const ReviewSchema z.object({ passed: z.boolean(), issues: z.array(z.string()), }) async function reviewCodeNode(state) { const reviewModel model.withStructuredOutput(ReviewSchema) const result await reviewModel.invoke([ new SystemMessage(审查 Vue3 代码检查内存泄漏、响应式使用、类型安全。), new HumanMessage(state.code), ]) return { review: result.issues.join(; ), passed: result.passed } } // 路由函数通过了结束没通过且没超次数就重新生成 function routeReview(state) { if (state.passed) return end if (state.attempts 3) return end // 最多重试 3 次 return regenerate } const reviewGraph new StateGraph(ReviewState) .addNode(generate, generateCodeNode) .addNode(review, reviewCodeNode) .addEdge(START, generate) .addEdge(generate, review) .addConditionalEdges(review, routeReview, { end: END, regenerate: generate, // 指回 generate形成循环 }) .compile() const result await reviewGraph.invoke({ requirement: 带 loading 和错误处理的数据获取 composable, }) console.log(生成了 ${result.attempts} 次最终通过${result.passed})3.7 完整项目Vue3 LangGraph 聊天应用把本章内容串起来后端用 LangGraph 管理对话流程前端用 Vue3 实现多会话聊天界面。后端架构Express ├── POST /api/sessions 创建会话返回 sessionId ├── GET /api/sessions/:id/history 获取会话历史 ├── DELETE /api/sessions/:id 删除会话 ├── POST /api/chat 普通聊天非流式 └── POST /api/chat/stream 流式聊天SSE LangGraph管理对话状态和流程 InMemoryChatMessageHistory每个 session 独立的历史记录 MapsessionId, session内存存储生产换 Redis关键代码带记忆的对话图import { StateGraph, END, START, Annotation, messagesStateReducer } from langchain/langgraph const ChatState Annotation.Root({ messages: Annotation({ reducer: messagesStateReducer, default: () [] }), systemPrompt: Annotation({ reducer: (_, n) n, default: () 你是前端助手 }), }) async function chatNode(state) { const res await model.invoke([ new SystemMessage(state.systemPrompt), ...state.messages, ]) return { messages: [res] } } const chatGraph new StateGraph(ChatState) .addNode(chat, chatNode) .addEdge(START, chat) .addEdge(chat, END) .compile()关键代码流式 SSE 接口app.post(/api/chat/stream, async (req, res) { const { sessionId, message, systemPrompt } req.body const session getOrCreateSession(sessionId) const history await session.history.getMessages() res.setHeader(Content-Type, text/event-stream) res.setHeader(Cache-Control, no-cache) res.setHeader(Connection, keep-alive) let fullReply // streamEvents 比 stream 更细粒度可以区分 token / tool_call / chain_end 等事件 for await (const event of chatGraph.streamEvents( { messages: [...history, new HumanMessage(message)], systemPrompt }, { version: v2 } )) { if ( event.event on_chat_model_stream event.data?.chunk?.content ) { fullReply event.data.chunk.content res.write(event: token\ndata: ${JSON.stringify({ token: event.data.chunk.content })}\n\n) } } // 流结束后保存历史 await session.history.addMessage(new HumanMessage(message)) await session.history.addMessage(new AIMessage(fullReply)) res.write(event: done\ndata: {}\n\n) res.end() })关键代码Vue3 流式接收// composables/useChat.js import { ref, reactive } from vue export function useChat(apiBase http://localhost:3000) { const messages ref([]) const loading ref(false) const streaming ref(false) const streamContent ref() async function send(message, sessionId) { if (!message.trim() || loading.value) return loading.value true streaming.value true streamContent.value // 先把用户消息推到界面 messages.value.push({ role: user, content: message, time: new Date() }) const res await fetch(${apiBase}/api/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ sessionId, message }), }) const reader res.body.getReader() const decoder new TextDecoder() let buffer while (true) { const { value, done } await reader.read() if (done) break buffer decoder.decode(value, { stream: true }) const lines buffer.split(\n) buffer lines.pop() // 保留未完整的行 for (const line of lines) { if (line.startsWith(data: )) { try { const data JSON.parse(line.slice(6)) if (data.token) streamContent.value data.token } catch {} } if (line event: done) { // 流式结束把临时内容转为正式消息 messages.value.push({ role: assistant, content: streamContent.value, time: new Date(), }) streaming.value false streamContent.value } } } loading.value false } return { messages, loading, streaming, streamContent, send } }!-- 在组件里使用 -- template div div v-formsg in messages :keymsg.time :classmsg.role {{ msg.content }} /div !-- 正在流式输出的临时内容 -- div v-ifstreaming classassistant {{ streamContent }}span classcursor / /div textarea v-modelinput keydown.ctrl.enterhandleSend / button clickhandleSend :disabledloading发送/button /div /template script setup import { ref } from vue import { useChat } from /composables/useChat const { messages, loading, streaming, streamContent, send } useChat() const input ref() const sessionId ref(null) async function handleSend() { if (!input.value.trim()) return await send(input.value, sessionId.value) input.value } /script3.8 选型参考场景推荐方案单次 AI 调用model.invoke()直接用多步线性处理LCEL 顺序链RunnableSequence多任务并行RunnableParallel或Promise.all需要会话记忆RunnableWithMessageHistory需要条件分支LangGraphaddConditionalEdges需要循环修正LangGraph节点指回前面的节点复杂 AgentLangGraph第七章详细讲3.9 本章小结LangChain.js 的核心是Runnable接口所有组件模型、模板、解析器都可以用.pipe()任意组合ChatPromptTemplate把提示词参数化partial()预填部分变量实现模板复用LCEL 的顺序链和并行链解决了多步 AI 流程的编排问题代码比手写逻辑清晰很多会话记忆用RunnableWithMessageHistory不同sessionId自动隔离生产环境把InMemoryChatMessageHistory换成 Redis 实现即可LangGraph 的核心三要素State状态、Node节点、Edge边条件边实现动态路由循环边实现自我修正streamEvents比stream更细粒度可以区分 token 输出和工具调用等不同事件前端拿来推送进度更准确