手把手教你用 Milvus + LangChain 搭建《天龙八部》RAG 知识库

RAG = 检索 + 生成，向量数据库是桥梁，LangChain 是胶水，而你的业务数据才是灵魂。

你是否曾幻想过，像问 ChatGPT 一样问一本小说：“段誉会什么武功？”然后它不仅能给出答案，还能告诉你这句话出自第几章？这就是 RAG（检索增强生成）的典型应用——把静态文档变成可交互的知识库。

本文将带你从零开始，用Milvus+LangChain搭建一个“天龙八部问答机器人”。全文实战优先，不但给出可运行的代码，还会逐函数解析关键 API，并用流程图帮你理清整个流程。读完你不仅会跑，还会调优，顺带带走几个可以直接复用的设计模式。

一、项目全景：我们到底在做什么？

RAG 的核心三步：

1. 离线建库：将电子书加载、分块、向量化，存入向量数据库。
2. 在线检索：用户提问时，将问题向量化，从数据库中检索最相关的几个片段。
3. 生成答案：将检索到的片段拼接成上下文，喂给大模型，生成最终回答。

整个项目包含两个主要脚本：

• main.mjs：负责建库（加载 EPUB、分块、生成向量、插入 Milvus）。
• rag.mjs：负责问答（检索 + LLM 生成）。

下面我们按照流程图的顺序逐一拆解。

二、整体流程（先看全景图）

上半部分是离线建库，下半部分是在线问答。我们按顺序展开。

三、准备工作：环境与依赖

3.1 安装依赖

npm install @zilliz/milvus2-sdk-node @langchain/openai @langchain/community @langchain/textsplitters dotenv

3.2 环境变量（.env）

MILVUS_ADDRESS=https://your-instance.region.zillizcloud.com:19530 MILVUS_TOKEN=your-api-token OPENAI_API_KEY=sk-xxx OPENAI_BASE_URL=https://api.openai.com/v1 # 代理可替换 EMBEDDINGS_MODEL_NAME=text-embedding-3-small MODEL_NAME=gpt-4o-mini

3.3 初始化核心客户端

import "dotenv/config"; import { MilvusClient, DataType, MetricType, IndexType } from '@zilliz/milvus2-sdk-node'; import { OpenAIEmbeddings, ChatOpenAI } from '@langchain/openai'; import { EPubLoader } from '@langchain/community/document_loaders/fs/epub'; import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters"; const COLLECTION_NAME = 'ebook'; const VECTOR_DIM = 1024; const CHUNK_SIZE = 500; const EPUB_FILE = './天龙八部.epub'; // Milvus 客户端（注意地址处理） const client = new MilvusClient({ address: process.env.MILVUS_ADDRESS.replace(/^https?:///, ''), token: process.env.MILVUS_TOKEN, ssl: true }); // Embedding 模型（指定维度） const embeddings = new OpenAIEmbeddings({ apiKey: process.env.OPENAI_API_KEY, model: process.env.EMBEDDINGS_MODEL_NAME, configuration: { baseURL: process.env.OPENAI_BASE_URL }, dimensions: VECTOR_DIM }); // LLM（用于问答） const model = new ChatOpenAI({ temperature: 0.1, model: process.env.MODEL_NAME, apiKey: process.env.OPENAI_API_KEY, configuration: { baseURL: process.env.OPENAI_BASE_URL } });

注意：text-embedding-3-small默认输出 1536 维，我们通过dimensions: 1024截断，减少存储和计算开销，同时保持语义质量。

四、离线建库（核心函数详解）

4.1 集合管理：ensureCollection

在插入数据前，必须确保 Milvus 集合存在，并已创建索引且加载到内存。这个函数承担了“守门员”的角色。

async function ensureCollection(bookId) { try { // 1. 检查集合是否已存在（幂等性） const hasCollection = await client.hasCollection({ collection_name: COLLECTION_NAME }); if (!hasCollection.value) { console.log('创建集合...'); // 2. 定义 Schema（字段结构） await client.createCollection({ collection_name: COLLECTION_NAME, fields: [ { name: 'id', data_type: DataType.VarChar, max_length: 100, is_primary_key: true }, { name: 'book_id', data_type: DataType.VarChar, max_length: 100 }, { name: 'book_name', data_type: DataType.VarChar, max_length: 200 }, { name: 'chapter_num', data_type: DataType.Int32 }, { name: 'index', data_type: DataType.Int32 }, { name: 'content', data_type: DataType.VarChar, max_length: 10000 }, { name: 'vector', data_type: DataType.FloatVector, dim: VECTOR_DIM } ] }); console.log('集合创建成功'); // 3. 创建向量索引（搜索加速的关键） console.log('创建索引...'); await client.createIndex({ collection_name: COLLECTION_NAME, field_name: 'vector', index_type: IndexType.IVF_FLAT, metric_type: MetricType.COSINE, params: { nlist: 1024 } }); console.log('索引创建成功'); } // 4. 加载集合到内存（搜索前必须） try { await client.loadCollection({ collection_name: COLLECTION_NAME }); console.log('集合加载成功'); } catch (err) { // 若已加载会报错，忽略即可 console.error('集合可能已加载'); } } catch (err) { console.error('创建集合时出错:', err); } }

API 解析：

• hasCollection：检查集合是否存在，返回{ value: boolean }。
• createCollection：定义字段，is_primary_key必须指定一个字段，我们使用字符串组合 ID 确保唯一性。
• createIndex：索引类型IVF_FLAT是倒排索引 + 扁平量化，适合中等规模数据；nlist是聚类个数，一般取sqrt(数据量)。metric_type选择COSINE，因为 OpenAI 的向量已归一化。
• loadCollection：将索引加载到内存，只有加载后才能搜索。重复加载会抛异常，所以我们用 try-catch 包裹。

为什么需要先创建索引再插入数据？因为 Milvus 支持动态建索引，但如果在插入后再创建，需要重建索引，会额外耗时。我们建议先建索引再插入，这样插入时就会自动构建增量索引。

4.2 流式加载与分块：loadAndProcessEPubStreaming

这个函数负责加载 EPUB、按章节拆分、分块、调用insertChunksBatch入库。我们采用“逐章处理”的策略，避免一次性将所有章节内容加载到内存。

async function loadAndProcessEPubStreaming(bookId) { try { console.log(`加载 EPUB: ${EPUB_FILE}`); const loader = new EPubLoader(EPUB_FILE, { splitChapters: true // 按目录自动拆分章节 }); const documents = await loader.load(); // 每个元素是一个 Document（一章节） console.log(`加载到 ${documents.length} 个章节`); const textSplitter = new RecursiveCharacterTextSplitter({ chunkSize: CHUNK_SIZE, chunkOverlap: 50 }); let totalInserted = 0; for (let chapterIndex = 0; chapterIndex < documents.length; chapterIndex++) { const chapter = documents[chapterIndex]; const chapterContent = chapter.pageContent; console.log(`处理第 ${chapterIndex + 1}/${documents.length} 章...`); // 分块 const chunks = await textSplitter.splitText(chapterContent); console.log(`拆分为 ${chunks.length} 个片段`); if (chunks.length === 0) continue; // 批量插入 const insertedCount = await insertChunksBatch( chunks, bookId, chapterIndex + 1 ); totalInserted += insertedCount; console.log(`已插入 ${totalInserted} 条记录`); } console.log(`总共插入 ${totalInserted} 条记录`); return totalInserted; } catch (err) { console.error('处理 EPUB 时出错:', err); } }

设计要点：

• EPubLoader的splitChapters: true会解析目录，每个章节生成一个独立的Document，pageContent存储纯文本。
• RecursiveCharacterTextSplitter默认分隔符为["\n\n", "\n", " ", ""]，它会递归尝试，保证每个块不超过chunkSize，同时chunkOverlap让边界信息得以保留。
• 我们逐章处理，处理完一章即释放该章节的引用，控制内存峰值。

为什么不用loadAndSplit？因为我们需要在插入时携带章节号，自定义 ID，所以手动遍历更灵活。

4.3 批量插入与向量化：insertChunksBatch

这个函数负责将一章内的所有 chunk 转为向量，并批量插入 Milvus。它是性能优化的核心。

async function insertChunksBatch(chunks, bookId, chapterNum) { try { if (chunks.length === 0) return 0; // 并发生成向量（注意并发数控制） const insertData = await Promise.all( chunks.map(async (chunk, chunkIndex) => { const vector = await embeddings.embedQuery(chunk); return { id: `${bookId}_${chapterNum}_${chunkIndex}`, book_id: bookId, book_name: '天龙八部', chapter_num: chapterNum, index: chunkIndex, content: chunk, vector: vector }; }) ); // 一次性插入整章 const insertResult = await client.insert({ collection_name: COLLECTION_NAME, data: insertData }); return Number(insertResult.insert_cnt) || 0; } catch (err) { console.error(`插入章节 ${chapterNum} 失败:`, err.message); throw err; // 让上层捕获 } }

API 解析：

• embedQuery：将单个文本转为向量，返回number[]。
• Promise.all：并发执行所有 chunk 的向量化，大幅缩短总时间。但如果一章有上百个 chunk，可能触发 API 限流，此时建议引入p-limit控制并发数（例如 10）。
• client.insert：一次性提交整章数据，减少网络往返。insert_cnt返回插入条数，我们转为数字返回。
• 错误处理：捕获异常后重新抛出，让上层（loadAndProcessEPubStreaming）知晓插入失败，从而停止处理或重试。

金句：向量化是 RAG 中最耗时的环节，善用并发和批量，别让 I/O 等你。

4.4 主函数（main）调度

const main = async () => { try { console.log('='.repeat(80)); console.log('电子书处理程序'); console.log('='.repeat(80)); await client.connectPromise; // 确保连接 const bookId = '1'; await ensureCollection(bookId); await loadAndProcessEPubStreaming(bookId); console.log('建库完成！'); } catch (err) { console.error('主流程出错:', err); } }; main();

解析：

• client.connectPromise是一个 Promise，在调用任何 Milvus 操作前自动连接，但我们显式等待它，确保连接成功。
• 流程简洁：先确保集合存在，再加载数据。
• 错误捕获后打印，避免进程崩溃。

五、在线问答（检索 + 生成）

这部分在rag.mjs中实现，核心是两个函数：检索和生成。

5.1 检索相关片段：retrieveRelevantContent

async function retrieveRelevantContent(question, k = 3) { try { const queryVector = await embeddings.embedQuery(question); const searchResult = await client.search({ collection_name: COLLECTION_NAME, vector: queryVector, limit: k, metric_type: MetricType.COSINE, output_fields: ['id', 'chapter_num', 'content'], // params: { nprobe: 16 } // 可调，提升召回 }); return searchResult.results; } catch (err) { console.error('检索失败:', err); return []; } }

API 解析：

• search：执行向量检索，limit控制返回条数，output_fields指定返回的标量字段。
• params.nprobe：IVF 索引专用，决定搜索时访问的聚类个数，默认 8，增大可提升召回但会降低速度。可调优。

5.2 组装 Prompt 并调用 LLM：answerEbookQuestion

async function answerEbookQuestion(question, k = 3) { const results = await retrieveRelevantContent(question, k); if (!results.length) return '未找到相关内容。'; const context = results.map((item, i) => ` [片段 ${i+1}] 章节：第 ${item.chapter_num} 章 内容：${item.content} `).join('\n----\n'); const prompt = `你是一个专业的《天龙八部》小说助手。请根据以下片段回答问题： ${context} 问题：${question} 要求： 1. 如果片段中有相关信息，请结合小说内容给出详细准确的回答。 2. 可以综合多个片段，提供完整的答案。 3. 如果片段中没有相关信息，请如实告知。 4. 回答要符合小说情节和人物设定。 5. 可以引用原文内容来支持回答。 AI 助手的回答：`; const response = await model.invoke(prompt); return response.content; }

设计要点：

• 将检索到的片段按章节和内容格式化，注入 Prompt。
• 明确要求 LLM “只根据片段回答”，降低幻觉风险。
• 设置temperature: 0.1让答案更确定。

六、项目亮点与技术最佳实践

1. 流式处理与内存优化
   逐章处理，避免整本书常驻内存；若数据量更大，可改用lazyLoad或流式读取。

2. 批量操作与并发控制
   按章批量插入，减少网络 I/O；Promise.all加速向量生成，同时预留限流接口。

3. 健壮的错误处理

   • 集合创建幂等（hasCollection）。
   • 加载集合时捕获“已加载”异常。
   • 插入失败抛出异常，上层可决策重试。

4. 可扩展的架构
   Loader、Splitter、Embedding、数据库均可替换，符合开闭原则。

5. 清晰的模块化
   每个函数职责单一，便于单元测试和维护。

6. 实践中的调优空间

   • 调整chunk_size和overlap。
   • 调整nlist/nprobe平衡速度与召回。
   • 可引入重排序（Rerank）进一步提升答案质量。

七、避坑指南 & 常见问题

八、总结

我们完整实现了一个 RAG 应用，从 EPUB 加载、分块、向量化、Milvus 建库、检索到 LLM 生成，代码总共不到 200 行。通过逐函数解析 API，你不仅知道“怎么写”，更明白了“为什么这么写”。

最后，记住一句话：

RAG 的精度不在模型，而在数据分块和检索策略；效果不好，先调 chunk 和 nprobe，别急着换大模型。

你可以把“天龙八部”换成技术文档、公司规章制度、法律条文……只需替换 Loader 即可。希望这篇文章能成为你 RAG 实战之路的一块基石。