LangChain.js 实战与原理:用 LCEL 构建可维护的 RAG / Agent 系统(含 4 套 30+ 行代码)
目录
-
背景:为什么是 JavaScript 版的 LangChain
-
原理:LCEL 执行模型与核心组件(Chain、Tool、Retriever、Memory)
-
实战一(Node 端):最小可用 RAG 服务(流式输出 + 缓存 + 观测回调)
-
实战二(浏览器端):前端离线 RAG(Web Worker + MemoryVectorStore)
-
实战三(边缘计算):Cloudflare Workers 部署“问文档” API(类 LangServe)
-
实战四(Agent 工具链):函数调用 + 多工具路由 + 降级策略
-
对比与取舍:LangChain.js vs LangChain(Py) / LlamaIndex / DSPy
-
性能与工程化:切片策略、Embedding 选型、批量化与缓存、评测与防注入
-
总结与互动:让链条“可观察、可测试、可迭代”
1. 背景:为什么是 JavaScript 版的 LangChain
如果把“大模型应用”视作一条生产线,那么 LangChain 就是把“模型—数据—工具—记忆—流程”串起来的那套输送带。很多人先接触的是 Python 版,但 LangChain.js 有其独特优势:
-
同构能力:Node 端做服务、浏览器端做交互与本地推理,一套 TypeScript 类型体系贯穿前后端。
-
生态贴近前端:容易与 React/Vue、Web Worker、Service Worker、Edge Runtime(Vercel/Cloudflare)融合。
-
部署轻量:函数即服务 / 边缘节点友好,支持流式输出,适合构建“问文档”“智能表单”“Agent 工作台”等场景。
很多前端团队都在问:我需要从零写一个 RAG / Agent 框架吗? 答案通常是——不。LangChain.js 提供了规范化的组件与可组合的 LCEL(LangChain Expression Language),让我们用“积木式”方式稳定地搭链,避免“胶水地狱”。
2. 原理:LCEL 执行模型与核心组件
LCEL(LangChain Expression Language) 是 LangChain 的可组合执行模型。可以把它理解成“可串联、可分支的可运行对象(Runnable)流水线”。它有几个关键属性:
-
声明式:我们用
prompt.pipe(model).pipe(parser)
这种链式写法描述流程。 -
可观测:在每个 Runnable 上挂回调(Callbacks)做日志、指标、追踪。
-
可复用:每个环节都是“黑盒—白盒兼具”的可测试单元。
-
可路由:支持条件路由(Router)、并行(Parallel)、字典映射(RunnableMap),像在画数据流图。
2.1 核心组件速览
-
Model(LLM/ChatModel/Embedding):大模型与向量化。
-
Prompt:非“字符串连接”,而是可模板化的 Prompt 对象,支持变量注入与多消息。
-
Chain(Runnable):LCEL 的基本执行节点,可由 Prompt→Model→Parser 组成。
-
Tool:外部能力的可调用接口(HTTP、DB、计算、搜索等)。
-
Retriever / VectorStore:RAG 的检索器与底层向量库。
-
Memory:上下文记忆(聊天摘要、窗口记忆、向量记忆等)。
-
OutputParser:把模型输出转为结构化对象(JSON Schema、安全解析与容错)。
-
Callbacks:链路观测(日志、Tracing、Token 统计、异常报警)。
2.2 LCEL 的执行思路(以 RAG 为例)
-
用户问题进入 chain
-
Retriever 基于向量相似度检索相关文档
-
Prompt 将文档摘要与问题注入模板
-
LLM 生成回答;同时 OutputParser 做结构化或安全解析
-
Callbacks 记录本次链路的输入、检索片段、token、耗时等指标
3. 实战一(Node 端):最小可用 RAG 服务(流式输出 + 缓存 + 观测回调)
目标:在 Node 中构建一个 最小可用 的检索增强(RAG)链路,具备:
-
内置
MemoryVectorStore
(也可替换为 Pinecone、Weaviate、Qdrant) -
流式输出(边生成边返回)
-
简单缓存(避免重复 Embedding)
-
链路观测(打印关键事件)
依赖(按需替换):
@langchain/core
、@langchain/openai
或任意兼容的模型提供方、langchain
(集合包,随版本可能拆分)
// file: server/rag-minimal.ts
// 依赖安装示例:
// npm i langchain @langchain/core @langchain/openai express cors
// 环境变量:OPENAI_API_KEY=xxx
import express from "express";
import cors from "cors";
import { ChatOpenAI, OpenAIEmbeddings } from "@langchain/openai";
import { MemoryVectorStore } from "langchain/vectorstores/memory";
import { RecursiveCharacterTextSplitter } from "langchain/text_splitter";
import { RunnableSequence, RunnablePassthrough } from "@langchain/core/runnables";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import { Document } from "@langchain/core/documents";
import type { CallbackManagerForChainRun } from "@langchain/core/callbacks/base";// 1) 初始化服务
const app = express();
app.use(cors());
app.use(express.json());// 2) 构建向量库(示例数据,也可通过文件系统/数据库加载)
const docsRaw = [{ id: "1", text: "LangChain.js 是一个用于构建大模型应用的 TypeScript/JavaScript 库。" },{ id: "2", text: "LCEL 允许以可组合的方式串联 Prompt、Model、Parser 等组件。" },{ id: "3", text: "RAG(检索增强生成)结合向量检索与大模型,降低幻觉并对接企业知识库。" },
];const embeddings = new OpenAIEmbeddings();
const vectorStore = await MemoryVectorStore.fromDocuments(await new RecursiveCharacterTextSplitter({ chunkSize: 200, chunkOverlap: 40 }).splitDocuments(docsRaw.map(d => new Document({ pageContent: d.text, metadata: { id: d.id } }))),embeddings
);// 3) 构建 Retriever
const retriever = vectorStore.asRetriever(3);// 4) 构建 Prompt
const prompt = ChatPromptTemplate.fromMessages([["system", "你是严谨的资深工程师,请基于给定上下文回答。若无依据,请说不知道。"],["human", "问题:{question}\n\n上下文(仅供参考):\n{context}"],
]);// 5) 模型与解析器(可替换其它模型提供方)
const model = new ChatOpenAI({modelName: "gpt-4o-mini", // 示例;替换为你可用的 Chat 模型temperature: 0.2,
});// 6) 将检索、模板、模型、解析器串为 LCEL
const chain = RunnableSequence.from([{question: new RunnablePassthrough(),context: async (q: string, _r, runManager?: CallbackManagerForChainRun) => {const start = Date.now();const docs = await retriever.getRelevantDocuments(q);const elapsed = Date.now() - start;await runManager?.handleText?.(`检索到片段:${docs.length},耗时:${elapsed}ms`);return docs.map(d => `- ${d.pageContent}`).join("\n");},},prompt,model,new StringOutputParser(),
]);// 7) HTTP 接口:支持流式
app.post("/rag", async (req, res) => {const { question } = req.body as { question: string };res.setHeader("Content-Type", "text/event-stream; charset=utf-8");res.setHeader("Cache-Control", "no-cache");res.setHeader("Connection", "keep-alive");const stream = await chain.stream(question, {callbacks: [{handleText(text) {res.write(`data: ${text}\n\n`); // SSE 流式},handleChainEnd(outputs) {res.write(`data: [DONE]\n\n`);},handleChainError(err) {res.write(`data: [ERROR] ${String(err)}\n\n`);},}]});// 消耗流,保持事件顺序for await (const _ of stream) { /* no-op: 已在回调输出 */ }res.end();
});app.listen(8787, () => {console.log("RAG server listening on http://127.0.0.1:8787");
});
运行思路:
-
启动后
POST /rag
,请求体{ "question": "LCEL 的优势是什么?" }
-
通过 SSE 流式返回 token 片段
-
可替换向量库为 Pinecone / Qdrant / Weaviate,替换 Embedding 为企业内部方案
示意图(链路观测面板):
图 2:在回调中抓取检索片段数量、耗时、token 指标
4. 实战二(浏览器端):前端离线 RAG(Web Worker + MemoryVectorStore)
目标:不依赖后端,在浏览器内完成小型知识库的检索与回答(适合 Demo、内网、教育场景)。要点:
-
使用 Web Worker 在后台构建与检索,避免阻塞 UI
-
使用
MemoryVectorStore
与RecursiveCharacterTextSplitter
-
使用 可替换模型适配层(可对接浏览器本地模型、或通过 HTTP 调用云端模型)
// file: web/worker.ts
// Web Worker:负责拆分、向量化、检索、调用模型并返回答案
// 注意:浏览器端 Embedding/LLM 适配视实际可用性而定;此处给出抽象接口以便替换。
import { RecursiveCharacterTextSplitter } from "langchain/text_splitter";
import { Document } from "@langchain/core/documents";
import { MemoryVectorStore } from "langchain/vectorstores/memory";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { RunnableSequence } from "@langchain/core/runnables";
import { StringOutputParser } from "@langchain/core/output_parsers";// —— 抽象:浏览器端 Embedding 与 ChatModel 适配层 ——
// 你可以替换为:本地 wasm 模型 / 远端 API 代理 / WebGPU 方案等
async function embedBatch(texts: string[]): Promise<number[][]> {// 示例:调用你后端的 /embed 接口,或本地 wasm Embeddingconst res = await fetch("/api/embed", {method: "POST",headers: { "Content-Type": "application/json" },body: JSON.stringify({ texts }),});return (await res.json()).vectors;
}async function chatOnce(prompt: string): Promise<string> {// 示例:调用你后端的 /chat 接口(或 Vercel/CF Worker 代理)const res = await fetch("/api/chat", {method: "POST",headers: { "Content-Type": "application/json" },body: JSON.stringify({ prompt }),});const data = await res.json();return data.text;
}// —— Worker 状态 ——
let vectorStore: MemoryVectorStore | null = null;// —— 消息协议 ——
type Message =| { type: "build"; payload: { corpus: Array<{ id: string; text: string }> } }| { type: "ask"; payload: { question: string } };self.onmessage = async (e: MessageEvent<Message>) => {const msg = e.data;if (msg.type === "build") {const { corpus } = msg.payload;const splitter = new RecursiveCharacterTextSplitter({ chunkSize: 300, chunkOverlap: 60 });const docs = corpus.map((d) => new Document({ pageContent: d.text, metadata: { id: d.id } }));const chunks = await splitter.splitDocuments(docs);// 对 chunks 做批量 embedding(减少请求轮次)const vectors = await embedBatch(chunks.map(c => c.pageContent));vectorStore = new MemoryVectorStore(undefined, vectors);// 将文本与向量绑定(此写法按你的 VectorStore 实现调整)// 这里用简化:直接在内存结构中附加文本(vectorStore as any)._docs = chunks; (self as any).postMessage({ type: "built", payload: { chunks: chunks.length } });return;}if (msg.type === "ask") {if (!vectorStore) {(self as any).postMessage({ type: "error", payload: "VectorStore not built" });return;}const { question } = msg.payload;// 简化的向量相似检索:假设 vectorStore 支持 asRetrieverconst retriever = (vectorStore as any).asRetriever?.(3) || {getRelevantDocuments: async (_q: string) => (vectorStore as any)._docs.slice(0, 3),};const relevant = await retriever.getRelevantDocuments(question);const prompt = ChatPromptTemplate.fromMessages([["system", "你是严谨的技术顾问,回答时务必引用上下文;若无依据,请说不知道。"],["human", "问题:{question}\n\n上下文:\n{context}"],]);const chain = RunnableSequence.from([async (q: string) => ({question: q,context: relevant.map((d: any) => `- ${d.pageContent}`).join("\n"),}),prompt,async (msgs) => ({ text: await chatOnce(msgs.toChatMessages().map(m => m.content).join("\n")) }),new StringOutputParser().pipe((text) => text.trim()),]);const answer = await chain.invoke(question);(self as any).postMessage({ type: "answer", payload: { text: answer } });}
};
// file: web/app.ts
// 主线程:绑定 UI 与 Worker
const worker = new Worker(new URL("./worker.ts", import.meta.url), { type: "module" });worker.onmessage = (e: MessageEvent<any>) => {const { type, payload } = e.data;if (type === "built") {log(`向量库构建完成:切片=${payload.chunks}`);} else if (type === "answer") {renderAnswer(payload.text);} else if (type === "error") {log(`错误:${payload}`);}
};document.querySelector<HTMLButtonElement>("#build")!.onclick = () => {const seed = [{ id: "a", text: "LangChain.js 在浏览器端可与 Web Worker 配合构建轻量 RAG。" },{ id: "b", text: "MemoryVectorStore 适合 Demo,小规模知识库可直接内存载入。" },{ id: "c", text: "生产建议使用持久化向量库,如 Pinecone / Qdrant / Weaviate。" },];worker.postMessage({ type: "build", payload: { corpus: seed } });
};document.querySelector<HTMLButtonElement>("#ask")!.onclick = () => {const q = (document.querySelector<HTMLInputElement>("#q")!.value || "").trim();if (!q) return;worker.postMessage({ type: "ask", payload: { question: q } });
};function log(s: string) {const el = document.querySelector("#log")!;el.innerHTML += `<div>${s}</div>`;
}function renderAnswer(text: string) {document.querySelector("#answer")!.innerHTML = text.replace(/\n/g, "<br/>");
}
这套方案的核心在于“适配层”:Embedding/LLM 不一定在浏览器端本地执行,很多时候通过你自己的后端代理(如
/api/embed
、/api/chat
)更稳妥。这样做可以统一密钥管理、路由策略和限流。
5. 实战三(边缘计算):Cloudflare Workers 部署“问文档” API(类 LangServe)
目标:在 Cloudflare Workers 上部署一个极简的 RAG API,支持:
-
POST /ask
:输入问题,返回答案 -
使用 Workers KV / D1 / R2 存储切片与元信息(示例里以内存/模拟方式表示)
-
边缘就近计算,降低延迟
// file: worker/src/index.ts
// package.json: { "type": "module" }
// wrangler.toml: 配置你的账户与路由
import { Hono } from "hono";
import { OpenAIEmbeddings, ChatOpenAI } from "@langchain/openai";
import { MemoryVectorStore } from "langchain/vectorstores/memory";
import { RecursiveCharacterTextSplitter } from "langchain/text_splitter";
import { Document } from "@langchain/core/documents";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { RunnableSequence } from "@langchain/core/runnables";
import { StringOutputParser } from "@langchain/core/output_parsers";type Bindings = {OPENAI_API_KEY: string;
};const app = new Hono<{ Bindings: Bindings }>();// 1) 初始化(生产中请使用持久化)
const rawDocs = ["Workers 适合边缘部署,延迟低,流量按需计费。","LangChain.js 可与 Workers 组合,快速提供问文档 API。","RAG 的关键是正确的切片、检索配置和提示词对齐。",
];
const documents = rawDocs.map((t, i) => new Document({ pageContent: t, metadata: { id: i } }));
const splitter = new RecursiveCharacterTextSplitter({ chunkSize: 200, chunkOverlap: 40 });
const chunks = await splitter.splitDocuments(documents);
const embeddings = new OpenAIEmbeddings();
const store = await MemoryVectorStore.fromDocuments(chunks, embeddings);
const retriever = store.asRetriever(3);const prompt = ChatPromptTemplate.fromMessages([["system", "你是一名可靠的技术写作者,请基于上下文回答;不知道就明说。"],["human", "问题:{q}\n\n上下文:\n{ctx}"],
]);app.post("/ask", async (c) => {const { q } = await c.req.json<{ q: string }>();const model = new ChatOpenAI({apiKey: c.env.OPENAI_API_KEY,modelName: "gpt-4o-mini",temperature: 0.2,});const chain = RunnableSequence.from([async () => {const docs = await retriever.getRelevantDocuments(q);return { q, ctx: docs.map(d => "- " + d.pageContent).join("\n") };},prompt,model,new StringOutputParser()]);const text = await chain.invoke({});return c.json({ text });
});export default app;
小贴士:Workers 中更推荐无状态逻辑 + 外部存储(KV/R2/D1),方便横向扩展。向量库可放在外部服务(如 Pinecone),通过 HTTP 访问。
6. 实战四(Agent 工具链):函数调用 + 多工具路由 + 降级策略
目标:演示 Agent 如何选择工具、执行子任务,并在失败时降级。要点:
-
工具(Tool):定义可被模型调用的外部能力,函数签名明确
-
路由(Router):根据模型意图将任务派发给不同工具
-
降级:当工具失败或不可用时,走替代路径(如直接给出保守答案)
// file: server/agent-tools.ts
// 演示:两个工具 + 路由 + 降级
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";
import { StructuredTool } from "@langchain/core/tools";
import { RunnableSequence, RunnableLambda } from "@langchain/core/runnables";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";// 1) 定义工具
class WeatherTool extends StructuredTool {name = "weather";description = "查询指定城市的天气(返回今日温度与天气概况)";schema = z.object({ city: z.string().describe("城市名,如 Beijing") });async _call(input: { city: string }): Promise<string> {// 示例:调用真实天气 API// 此处返回模拟值return `今日 ${input.city} 晴,28℃,湿度 40%`;}
}class SearchTool extends StructuredTool {name = "search";description = "根据关键词进行快速检索,返回前 3 条摘要";schema = z.object({ q: z.string().describe("搜索关键词") });async _call(input: { q: string }): Promise<string> {// 示例:调用搜索 APIreturn `结果1: ...\n结果2: ...\n结果3: ...`;}
}// 2) Agent 路由:让模型决定是否调用工具
const tools = [new WeatherTool(), new SearchTool()];
const toolMap = Object.fromEntries(tools.map(t => [t.name, t]));// 3) 构建决策 Prompt
const plannerPrompt = ChatPromptTemplate.fromMessages([["system", "你是任务规划器。判断用户问题是否需要调用工具。可选工具: weather, search。输出 JSON: {tool?:string, args?:object, rationale:string}"],["human", "用户问题:{input}"],
]);const executorPrompt = ChatPromptTemplate.fromMessages([["system", "你是回答助手。若提供了工具结果,请综合回答;若没有工具结果,请基于常识回答。"],["human", "用户问题:{input}\n\n工具结果:{toolResult}"],
]);export async function runAgent(userInput: string, apiKey: string) {const model = new ChatOpenAI({ apiKey, modelName: "gpt-4o-mini", temperature: 0 });const planner = RunnableSequence.from([plannerPrompt,model,new StringOutputParser().pipe((txt) => {try { return JSON.parse(txt); } catch { return { rationale: txt }; }}),]);const executor = RunnableSequence.from([executorPrompt,model,new StringOutputParser()]);const agent = RunnableSequence.from([async () => ({ input: userInput }),planner,// 路由到具体工具(若有)RunnableLambda.from(async (plan) => {const { tool, args } = plan || {};if (!tool || !toolMap[tool]) {return { toolResult: "未调用工具", input: userInput };}try {const result = await toolMap[tool].invoke(args ?? {});return { toolResult: result, input: userInput };} catch (e) {// 降级:工具失败时回退return { toolResult: `工具 ${tool} 调用失败:${String(e)}`, input: userInput };}}),executor,]);return agent.invoke({});
}// 用法:
// const text = await runAgent("北京今天热吗?需要带伞吗?", process.env.OPENAI_API_KEY!);
// console.log(text);
这段代码展示了 “规划—执行—降级” 的基本套路,实际工程可扩展:
-
工具注册中心 + 权限控制(哪些用户能用哪些工具)
-
调用超时 / 重试 / 熔断
-
观测:工具调用耗时、错误率、路由命中率
7. 对比与取舍:LangChain.js vs LangChain(Py) / LlamaIndex / DSPy
-
LangChain.js:前后端同构、TypeScript 友好、Edge 部署顺滑。适合 Web 全栈与前端主导的团队。
-
LangChain(Python):生态成熟、社区资源丰富、与数据/科研栈耦合更深(Pandas、PyTorch)。
-
LlamaIndex:RAG 能力抽象细颗粒、数据连接器丰富、检索评测工具完善;与 LangChain 并非“谁替代谁”,更多是“风格不同”。
-
DSPy:强调“以学习(编译)替代 Prompt 手写”,适合有实验基础的团队做“自动提示词/路由器”优化。
取舍建议:
-
你的界面与交互在 Web 端,且希望一套 TS 类型贯穿,那就 LangChain.js。
-
你有现成 Python 数据/ML 管线,或者要跑大量数据预处理/训练,Python 版可能更顺。
-
你希望玩“偏学术/自动化优化”的提示词与策略寻优,可以引入 DSPy 作为上层优化器。
8. 性能与工程化:切片、Embedding、批量化、评测与安全
切片(Chunking)
-
建议从
chunkSize ∈ [300, 800]
、chunkOverlap ∈ [40, 120]
试起; -
长文本优先按语义断句再定长切片,减少跨主题污染。
Embedding 选型
-
关注 语种(中文/多语种)、维度(768/1024/1536)、价格/速率;
-
若知识库偏专业术语,考虑领域化向量或混合检索(向量+BM25)。
批量化与缓存
-
Embedding 批量化(
batch
)能显著降延迟与费用; -
对输入/切片做 hash,命中缓存跳过重复向量化;
-
使用 Key-Value 缓存(Redis/KV)缓存模型中间产物(如重排结果)。
评测(Eval)
-
引入 合成问答集 + 人工校验;
-
指标:覆盖率(Recall)、回答一致性、事实性(Hallucination Rate);
-
工具可参考 RAGAS、自建少量人工评测集。
安全(Prompt Injection)
-
提示词中对模型加“边界约束”,例如“仅依据上下文回答,不可执行危险指令”;
-
对检索结果做白名单与内容过滤;
-
对外部 Tool 做参数校验、限流与审计。
9. 总结与互动:让链条“可观察、可测试、可迭代”
-
LangChain.js 的价值不止是“把模型接起来”,更是以 LCEL 为核心的工程化范式:
-
用 Runnable 把流程拆成可测试单元
-
用 Callbacks 做全链路观测
-
用 Router/Tools 管理复杂场景与多模型策略
-
在 浏览器/边缘/Node 间自如迁移,快速上线 MVP 并持续演进
-
-
若你正准备把“问文档/数字员工/智能客服/AI 助教”落地,建议先用本文四个实战作为骨架,逐步替换存储、模型与工具。
欢迎在评论区留言:
你更关心哪一块?(切片与重排 / 多工具路由 / 浏览器端本地推理 / LangServe 对接)我可以在后续文章把相应模块拆到“可复用模板”的级别,附上完整 repo。
参考与延伸阅读(可靠外链)
-
LangChain.js 官方文档:https://js.langchain.com
-
LangChain.js GitHub:GitHub - langchain-ai/langchainjs: 🦜🔗 Build context-aware reasoning applications 🦜🔗
-
Cloudflare Workers 文档:Overview · Cloudflare Workers docs
-
Pinecone 向量数据库文档:Pinecone Database - Pinecone Docs
-
OpenAI API(Chat/Embedding)文档:https://platform.openai.com/docs