从手写 API 调用升级到工程化 AI 应用框架。LangChain.js 解决"怎么调用模型"，LangGraph 解决"怎么编排 AI 流程"。

第一章直接用 fetch 调用 API 没问题，但项目一复杂就会遇到几个痛点：

• 每次都要手写消息格式、处理流式解析、管理对话历史
• 多步骤的 AI 流程（先分析、再检索、再生成）要自己串联
• 换个模型要改好几处代码
• 复杂的条件分支逻辑写起来乱

LangChain.js 解决前两个问题，LangGraph 解决后两个。

原始写法：fetch → 解析 → 手动拼接历史 → 再 fetch → ... LangChain.js：model.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 链式调用

LCEL（LangChain 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 RunnableWithMessageHistory

LangChain.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 里模型能正确回答"在学 Vue3"

3.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') // addConditionalEdges：classify 执行完后调用 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 独立的历史记录 Map<sessionId, 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-for="msg in messages" :key="msg.time" :class="msg.role"> {{ msg.content }} </div> <!-- 正在流式输出的临时内容 --> <div v-if="streaming" class="assistant"> {{ streamContent }}<span class="cursor" /> </div> <textarea v-model="input" @keydown.ctrl.enter="handleSend" /> <button @click="handleSend" :disabled="loading">发送</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 = '' } </script>

3.8 选型参考

3.9 本章小结

• LangChain.js 的核心是Runnable接口，所有组件（模型、模板、解析器）都可以用.pipe()任意组合
• ChatPromptTemplate把提示词参数化，partial()预填部分变量实现模板复用
• LCEL 的顺序链和并行链解决了多步 AI 流程的编排问题，代码比手写逻辑清晰很多
• 会话记忆用RunnableWithMessageHistory，不同sessionId自动隔离，生产环境把InMemoryChatMessageHistory换成 Redis 实现即可
• LangGraph 的核心三要素：State（状态）、Node（节点）、Edge（边）；条件边实现动态路由，循环边实现自我修正
• streamEvents比stream更细粒度，可以区分 token 输出和工具调用等不同事件，前端拿来推送进度更准确