中文资讯掘金 AI Wed, 15 Jul 2026 正文 ok 1428 words

RAG 入库实战：一份原始资料，如何变成可检索的知识

英文标题：RAG 入库实战：一份原始资料，如何变成可检索的知识

RAG 最常见的用途，是让大模型回答它原本不知道，或者不能只靠自身记忆准确回答的问题。企业内部文档、产品手册、课程资料、小说原文、项目代码说明，都可以成为 RAG 的外部知识来源。这些内容不一定存在

打开原文相关 4 · 机会 1

3.2 KB

589 B

99.0 KB

37.8 KB

16.7 KB

中文

RAG 最常见的用途，是让大模型回答它原本不知道，或者不能只靠自身记忆准确回答的问题。

企业内部文档、产品手册、课程资料、小说原文、项目代码说明，都可以成为 RAG 的外部知识来源。这些内容不一定存在于模型的训练数据中，即使存在，也可能不完整、不准确或者不够新。

所以，我们会先把自己的资料整理成知识库。用户提问时，系统从知识库中找出相关内容，再把这些内容交给大模型生成答案。

听起来很简单：把资料切一切，生成向量，再放进向量数据库不就行了吗？

真正实现时，问题往往就出在这一步。内容被切得太碎、来源信息丢失、原有结构没有保留，或者重复入库后新旧数据混在一起，都会直接影响后面的检索效果。到了这个阶段，再怎么调整 prompt，也很难弥补入库时已经丢失的信息。

以《逆天邪神.txt》为例，一份原始资料会依次经历读取、整理、切分和向量化，最后写入数据库中。

其他的资料也是一样的，只是采用的 loader 不一样。

RAG 入库做了什么

一个常见的 RAG 流程是：用户提出问题，系统从向量数据库里检索相关片段，再让大模型根据这些片段组织答案。

这很容易让人把注意力都放在后半段：换一个更强的模型、调整 prompt、增加召回数量，或者尝试一种新的检索策略。

但如果入库时已经把一段关键情节从中间切断，模型就看不到完整语义；如果 chunk 没有携带章节信息，系统即使命中了原文，也无法告诉用户出处；如果旧数据没有清理或覆盖，同一个问题还可能召回多个版本的内容。

模型只能使用检索回来的材料。材料从一开始就不完整，后面再聪明也补不回来。

因此，入库不是直接执行 txt -> vector，而是经过下面这条链路：

最重要的顺序：先理解文档，再切分文档，最后才考虑数据库。

保留文档结构

程序拿到 txt 后，首先用 LangChain 的 TextLoader 把它转换成统一的 Document：

import { fileURLToPath } from "node:url";
import { Document } from "@langchain/core/documents";
import { TextLoader } from "@langchain/classic/document_loaders/fs/text";


const BOOK_ID = "1";


const BOOK_NAME = "逆天邪神";

export async function loadRawDocuments(fileUrl) {
  
  const filePath = fileURLToPath(fileUrl);
  const docs = await new TextLoader(filePath).load();

  return docs.map(
    (doc) =>
      new Document({
        pageContent: doc.pageContent,
        metadata: {
          ...doc.metadata,
          
          source: doc.metadata?.source ?? filePath,
          book_id: BOOK_ID,
          book_name: BOOK_NAME,
        },
      }),
  );
}

pageContent 保存正文，metadata 保存书名、来源和业务标识。Loader 到这里就应该停下，它的任务只是把外部数据源变成统一格式，而不是决定文本应该怎样切。

为什么不立刻把整本书交给文本切分器？

因为小说不是一块没有结构的长文本。它有序章、章节、番外，章节之前还可能有书名和作者信息。如果直接按固定长度切分，正文仍然存在，但“这段内容属于哪一章”这个关系消失了。

一旦这个关系丢失，后面想展示出处、补充相邻上下文，或者按原文顺序重排检索结果，就只能重新猜。

更稳妥的做法是先识别章节边界，把整本书拆成章节级 Document：

const CHAPTER_PATTERN = /(^|\n)(序章|第\d+章[^\n]*)/g;

export function parseOneDocumentIntoChapters(doc) {
  
  const text = doc.pageContent.replace(/\r\n/g, "\n").trim();

  
  const matches = Array.from(text.matchAll(CHAPTER_PATTERN));

  
  if (matches.length === 0) {
    return [createChapterDocument()];
  }

  
  return matches.map((match, index) => {
    
    const start = match.index + match[1].length;
    const end = matches[index + 1]?.index ?? text.length;

    
    return createChapterDocument({
      sourceDoc: doc,
      
      chapterOrder: index,
      
      chapterNum: parseChapterNumber(match[2], index),
      chapterTitle: match[2].trim(),
      chapterText: text.slice(start, end).trim(),
    });
  });
}

正则表达式并不是这里的重点。不同小说的标题格式不同，真实项目里当然需要调整匹配规则。真正值得保留的是这个原则：先按内容本身的业务结构分层，再按模型和数据库需要的长度切片。

就比如，每个章节都会继承原始文档的 metadata，并补上自己的章节信息：

{
  book_id: "1",
  book_name: "逆天邪神",
  source: "逆天邪神.txt",
  chapter_num: 1,
  chapter_title: "第1章 云澈、萧澈",
  chapter_order: 0
}

这些字段会继续跟随章节进入下一步。文本会越切越小，但它来自哪里，不应该随着切分一起丢掉。

控制切片粒度

章节保住了结构，却仍然太长，不适合直接生成一个向量。

用户通常只问一个人物、一次对话或一段情节。如果一整章只有一个向量，语义会被大量无关内容稀释；命中后把整章交给模型，也会浪费上下文窗口。

但 chunk 也不是越小越好。切得过小，一句话可能被拦腰分开，人物和行为落入两个片段，检索到其中一个仍然无法回答问题。

这里采用 RecursiveCharacterTextSplitter，让切分器优先寻找段落、换行等自然边界，同时保留少量重叠：

import { Document } from "@langchain/core/documents";
import { RecursiveCharacterTextSplitter } from "@langchain/textsplitters";


const CHUNK_SIZE_BYTES = 900;


const CHUNK_OVERLAP_BYTES = 90;

const textSplitter = new RecursiveCharacterTextSplitter({
  chunkSize: CHUNK_SIZE_BYTES,
  chunkOverlap: CHUNK_OVERLAP_BYTES,
  
  lengthFunction: (text) => Buffer.byteLength(text, "utf8"),
});

export async function splitChapterDocuments(chapterDocuments) {
  const chunks = await textSplitter.splitDocuments(chapterDocuments);

  return chunks.map(
    (doc, chunkOrder) =>
      new Document({
        pageContent: doc.pageContent.trim(),
        
        metadata: { ...doc.metadata, chunk_order: chunkOrder },
      }),
  );
}

这段配置里有两个容易忽略的细节。

第一个是 overlap 。这里保留 90 字节的重叠，不是为了制造重复内容，而是给切分边界留一小段缓冲。答案刚好跨过边界时，相邻 chunk 仍有机会保留完整语义。

第二个是长度单位 。JavaScript 的 string.length 和数据库字段限制不是一回事，中文字符在 UTF-8 编码下通常占多个字节。Milvus 的 VarChar 长度限制按字节计算，因此 splitter 也必须使用 Buffer.byteLength()。否则程序以为片段没有超长，写入时却可能被数据库拒绝。

当然，900 和 90 也不是适用于所有资料的标准答案。它们只是当前 schema 和文本类型下的起点。真正需要观察的是：片段是否保留完整语义、是否超过存储限制，以及实际问题能否召回足够的上下文。

设计 Milvus Schema

chunk 准备好之后，就需要把它们保存到数据库中，这里使用向量数据库 Milvus。

如果只从“向量检索”四个字理解向量数据库，很容易只设计 id + vector 两个字段。但一个向量本身无法直接展示给用户，也无法说明它来自哪里。检索命中后，系统真正要使用的是向量背后的原文和上下文。

当前 collection 保存的是下面这些信息：

字段	用途
`id`	标识一个稳定的知识片段
`vector`	进行相似度检索
`content`	命中后交给大模型的原文
`book_id`、`book_name`	区分和过滤不同书籍
`chapter_num`	标记章节来源
`index`	恢复片段在全书中的顺序

collection 的核心 schema 可以概括为：

import {
  MilvusClient,
  DataType,
  MetricType,
  IndexType,
} from "@zilliz/milvus2-sdk-node";


const VECTOR_DIM = 1024;


const COLLECTION_NAME = "story";

const client = new MilvusClient({
  address: process.env.MILVUS_ADDRESS,
});

await client.connectPromise;

await client.createCollection({
  collection_name: COLLECTION_NAME,
  fields: [
    
    {
      name: "id",
      data_type: DataType.VarChar,
      is_primary_key: true,
      max_length: 100,
    },
    
    { name: "vector", data_type: DataType.FloatVector, dim: VECTOR_DIM },
    
    { 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: 1000 },
  ],
});


await client.createIndex({
  collection_name: COLLECTION_NAME,
  field_name: "vector",
  metric_type: MetricType.COSINE,
  index_type: IndexType.IVF_FLAT,
  params: { nlist: 1024 },
});

向量字段的 dim 必须与 embedding 模型的实际输出维度完全一致。这里使用 1024 维，所以模型配置和 Milvus schema 都共享同一个 VECTOR_DIM，避免两处配置悄悄发生偏差。

如果产品需要在答案中展示章节标题，还可以继续保存 chapter_title。schema 不是为了把现有 metadata 原样复制一遍，而是要反过来问：命中片段后，我需要怎样过滤、排序、展示和排查它？

生成稳定主键

第一次入库成功并不难。更麻烦的是第二次。

调试 RAG 时，我们很可能重新调整 chunk 大小、修正章节解析，甚至更换 embedding 模型。如果每次都只做 insert，旧数据和新数据就会同时存在。同一段原文被召回多次，看上去像检索很有把握，实际只是知识库重复了。

这次使用 book_id + chunk_order 生成固定宽度的主键：

const CHUNK_ID_WIDTH = 6;

function buildChunkId(bookId, chunkOrder) {
  
  return `${bookId}_${String(chunkOrder).padStart(CHUNK_ID_WIDTH, "0")}`;
}

章节号更符合人的阅读习惯，却不适合直接当主键。小说里可能有序章、番外、重复章节号或格式异常的标题；全书 chunk 顺序反而更容易形成确定的标识。

写入时再使用 upsert：主键不存在就插入，已经存在就更新。

await client.upsert({
  collection_name: COLLECTION_NAME,
  data: rows,
});

这样至少能保证相同主键不会因为脚本重跑而无限累积。需要注意的是，如果新的切片策略让 chunk 总数变少，旧版本末尾那些不再出现的主键不会自动消失。生产环境还需要增加版本字段，或者在整本书重建前按 book_id 清理旧数据。

这也是入库系统和一次性导入脚本的区别：前者必须知道数据下一次更新时会发生什么。

从文本到向量

切分完成后，每个 chunk 仍然是一个 Document：正文保存在 pageContent，书籍和章节信息保存在 metadata。写入 Milvus 之前，需要调用 embedding 模型，把 pageContent 转换成一组数字。

import { OpenAIEmbeddings } from "@langchain/openai";


const VECTOR_DIM = 1024;

const embeddings = new OpenAIEmbeddings({
  model: process.env.AI_EMBEDDING_MODEL,
  dimensions: VECTOR_DIM,
  apiKey: process.env.AI_EMBEDDING_KEY,
  configuration: {
    baseURL: process.env.AI_EMBEDDING_BASE_URL,
  },
});

async function buildMilvusRow(doc) {
  const chunkOrder = Number(doc.metadata.chunk_order);
  const bookId = String(doc.metadata.book_id);
  const id = buildChunkId(bookId, chunkOrder);

  
  const vector = await embeddings.embedQuery(doc.pageContent);

  
  if (vector.length !== VECTOR_DIM) {
    throw new Error(`向量维度错误：期望 ${VECTOR_DIM}，实际 ${vector.length}`);
  }

  return {
    id,
    vector,
    
    content: doc.pageContent,
    
    book_id: bookId,
    book_name: String(doc.metadata.book_name),
    chapter_num: Number(doc.metadata.chapter_num),
    index: chunkOrder,
  };
}

embedQuery() 的输入是一段字符串，返回值是 number[]。同一个 chunk 会以两种形式保存：vector 表示它的语义，用于相似度检索；content 保留原文，用于生成答案。书籍、章节和顺序信息则从 metadata 映射到 Milvus 的普通字段中。

控制并发与重试

一本小说可能产生几千个 chunk。每个 chunk 都要请求 embedding 服务，随后还要写入 Milvus。

最直接的写法是：

await Promise.all(
  chunks.map((chunk) => embeddings.embedQuery(chunk.pageContent)),
);

它很短，但会把几千个请求几乎同时发出去。模型服务通常会限制单位时间内的请求数量，请求过多时，后面的请求可能直接失败。即使只有一条失败，我们也很难确认哪些片段已经处理，哪些片段还需要重新处理。

实际实现采用了三层保护：

同一时间最多生成 4 个片段的向量，处理完再继续取下一批；
每 32 个知识片段写入一次 Milvus，避免一次提交的数据太多；
请求偶尔失败时自动再试几次，仍然失败才停止脚本。

每个批次都会依次生成向量、写入 Milvus，再检查这一批是否全部写入成功：

const EMBEDDING_CONCURRENCY = 4;


const UPSERT_BATCH_SIZE = 32;


export async function upsertAllDocuments(chunkDocuments) {
  let totalWritten = 0;

  
  for (
    let start = 0;
    start < chunkDocuments.length;
    start += UPSERT_BATCH_SIZE
  ) {
    const batch = chunkDocuments.slice(start, start + UPSERT_BATCH_SIZE);

    
    const rows = await mapWithConcurrency(
      batch,
      EMBEDDING_CONCURRENCY,
      buildMilvusRow,
    );

    
    const result = await withRetry("写入 Milvus", () =>
      client.upsert({ collection_name: COLLECTION_NAME, data: rows }),
    );

    
    assertMutationSuccess(result, "写入 Milvus");

    totalWritten += rows.length;
  }

  
  await client.flush({ collection_names: [COLLECTION_NAME] });
  return totalWritten;
}

这些措施不会让某个片段的检索结果变得更准确，但能避免入库停在半路，或者脚本显示成功、数据库里却缺少一部分内容。

串联入库流程

当读取、章节解析、切片和写入各自完成自己的职责后，整条链路应该非常短：

const BOOK_FILE_URL = new URL("./逆天邪神.txt", import.meta.url);


export async function loadAndProcessTxt() {
  
  const rawDocuments = await loadRawDocuments(BOOK_FILE_URL);

  
  const chapterDocuments = parseChapterDocuments(rawDocuments);

  
  const chunkDocuments = await splitChapterDocuments(chapterDocuments);

  
  const totalWritten = await upsertAllDocuments(chunkDocuments);

  
  const rowCount = await getCollectionRowCount();

  return {
    raw: rawDocuments.length,
    chapters: chapterDocuments.length,
    chunks: chunkDocuments.length,
    written: totalWritten,
    rowCount,
  };
}

那么执行上面的 loadAndProcessTxt 函数，就可以完成小说的入库。

用真实问题验证

数据写入 Milvus，只能证明入库脚本没有中断。知识片段能不能被正确找到、章节信息有没有保留、大模型能不能根据原文回答，还需要用一个真实问题验证。

可以随便问一个问题，比如以小说为例：

云澈拥有的第一部功法是什么？

验证过程分成两步：先从 Milvus 检索与问题最接近的原文片段，再把这些片段交给大模型生成答案。

import { ChatOpenAI, OpenAIEmbeddings } from "@langchain/openai";
import { Milvus } from "@langchain/community/vectorstores/milvus";


const COLLECTION_NAME = "story";


const VECTOR_DIM = 1024;


const TOP_K = 5;


const QUESTION = "云澈拥有的第一部功法是什么？";

const embeddings = new OpenAIEmbeddings({
  model: process.env.AI_EMBEDDING_MODEL,
  dimensions: VECTOR_DIM,
  apiKey: process.env.AI_EMBEDDING_KEY,
  configuration: {
    baseURL: process.env.AI_EMBEDDING_BASE_URL,
  },
});

const model = new ChatOpenAI({
  model: process.env.AI_MODEL,
  temperature: 0,
  apiKey: process.env.AI_KEY,
  configuration: {
    baseURL: process.env.AI_BASE_URL,
  },
});


const vectorStore = await Milvus.fromExistingCollection(embeddings, {
  collectionName: COLLECTION_NAME,
  url: process.env.MILVUS_ADDRESS,
  textField: "content",
  primaryField: "id",
  vectorField: "vector",
});


const results = await vectorStore.similaritySearchWithScore(QUESTION, TOP_K);

if (results.length === 0) {
  throw new Error("没有检索到相关片段，请检查 collection 和 embedding 配置");
}


const context = results
  .map(
    ([doc], index) => `
[片段 ${index + 1}]
章节：第 ${doc.metadata.chapter_num} 章
原文：${doc.pageContent}
`,
  )
  .join("\n");

const response = await model.invoke(`
你只能根据下面提供的小说原文回答问题。
如果原文中没有答案，请直接回答“根据现有片段无法确定”，不要补充或猜测。

${context}

问题：${QUESTION}
`);

console.log("模型回答：", response.content);


for (const [doc, score] of results) {
  console.log({
    score,
    chapter: doc.metadata.chapter_num,
    content: doc.pageContent,
  });
}

这里需要同时检查模型回答和检索原文。可以自己动手试试。

只有“检索到正确原文”和“答案能够被原文支持”同时成立，才能说明这批知识片段已经真正进入 RAG 链路。

总结

到这里，一份 txt 已经走完了完整的 RAG 入库链路：

Loader 读取原始文件，把不同数据源转换成统一的 Document。
章节解析保留原文结构，并把书籍、章节和顺序写入 metadata。
Splitter 在章节内部切出适合检索的 chunk，同时控制字节大小和上下文重叠。
Embedding 模型把 pageContent 转换成 vector，原文和来源信息继续保留。
稳定主键、分批 upsert、并发限制和重试，让大量知识片段能够重复、可靠地写入 Milvus。
真实问题经过向量检索后交给大模型回答，再用命中原文核对答案是否有依据。

小说只是这条链路中的一种数据源。换成产品手册、课程资料或企业文档时，需要调整的是文件读取方式和结构解析规则；从 chunk 切分、向量生成到检索验证，整体思路仍然成立。

一个可用的 RAG 知识库，不只是数据库里存在一批向量。每个片段还要保留原文、来源和顺序，能够被稳定更新，也能够通过真实问题验证。只有这样，原始资料才真正变成了大模型可以检索和使用的知识。

English / 原文