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）进一步提升答案质量。

---

七、避坑指南 & 常见问题

问题	解决方案
连接 Milvus 超时	检查地址、端口、SSL 设置，确保网络可达
集合加载失败（已加载）	用 getLoadState() 判断，或 try-catch 忽略
向量维度不匹配	确保 Embedding 模型 dimensions 与集合字段 dim 一致
检索结果不相关	增大 chunkSize 或 nprobe，或调整分块策略
插入速度慢	提高批量大小，或增加并发（但注意限流）
大模型回答幻觉	在 Prompt 中强调“只根据片段”，并降低 temperature

---

八、总结

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

最后，记住一句话：

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

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