深入探索 FastGPT 数据集创建的原理与实现细节

本文将带你一步步理解 FastGPT 在创建数据集时的核心流程，并重点剖析“文件上传与预处理”这一步的详细实现逻辑。通过本文，你将掌握数据集创建的两个阶段、数据如何进入训练队列、以及背后的模型验证、数据清洗与事务处理机制。

数据集创建的总体流程

FastGPT 的文件上传过程分为两个阶段：

1. 第一阶段：文件上传与预处理 —— 将用户上传的文件解析成结构化的数据，并插入到 MongoDB 的训练队列（dataset_trainings 表）中。
2. 第二阶段：向量化处理或 QA 化处理 —— 通过监听 MongoDB 的新增记录，自动触发嵌入向量计算或问答对拆分。

下面的代码清晰地展示了这两个阶段的调用关系：

// 第一步：创建数据集并插入 mongodb 的数据表中
createCollectionAndInsertData
   -> pushDataListToTrainingQueue // 将数据集数据推送到训练队列

// 第二步：注册处理函数，并自动处理数据
startMongoWatch
   -> createDatasetTrainingMongoWatch // 监控 mongodb 的插入操作，触发文本处理任务
      -> generateQA(); # QA问答对的处理
      -> generateVector();  # 嵌入向量处理

核心要点： 第一步只是把数据“放”到 MongoDB 的训练队列，真正耗时的模型调用是在第二步才发生的。这种设计将上传与计算解耦，提升了系统的响应速度和稳定性。

数据集创建第一步的实现

这一步由 createCollectionAndInsertData 函数触发，核心逻辑封装在 pushDataListToTrainingQueue 函数中。整个处理分为三个子步骤：模型验证与配置、数据预处理、数据插入。

1. 模型验证与配置

• 检查推理模型（agentModel）和向量模型（vectorModel）是否有效 —— 通过 getLLMModel 和 getVectorModel 获取模型配置，如果返回空则直接拒绝，并抛出错误。
• 根据训练模式设置最大 Token 数和权重：
  • 当模式为 chunk（分块）时：最大 Token 数 = vectorModelData.maxToken * 1.5，模型为向量模型，权重为向量模型的权重。
  • 当模式为 qa 或 auto 时：最大 Token 数 = agentModelData.maxContext * 0.8，模型为推理模型，权重为 0。
• 如果训练模式不在预定义范围内，也会抛出错误。

注意： 权重（weight）在向量模式下用于后期排序，而在 QA 模式下设置为 0，表示该模式不关心向量相似度。

2. 数据预处理

• 调用 simpleText 函数清理问题和答案中的空白字符（去除首尾空格、多余换行等）。
• 对索引（index）数据同样进行文本清理。
• 过滤规则：
  • 过滤掉空问题或空答案的数据。
  • 使用 Set 去重，过滤重复内容（基于同一数据集的重复检测）。
  • 过滤超过最大 Token 限制的数据。
• 将过滤后的数据分为四类：success（成功）、overToken（超长）、repeat（重复）、error（格式错误）。

小提示： 分类统计可以帮助你快速定位上传数据中的问题，例如过多超长记录时可能需要调整模型的最大上下文或选用的分块策略。

3. 数据插入

• 批量插入：每批 200 条数据，使用 insertMany 写入 MongoDatasetTraining 集合。
• 容错处理：如果批量插入失败（例如某条数据违反了唯一索引），会记录失败的文档并尝试逐条插入，确保成功部分不丢失。
• 事务支持：函数支持在已开启的 session 中执行，如果没有传入 session，则会自动创建新的事务，以保证数据一致性。

常见问题： 为什么需要事务？因为数据集创建可能涉及多个表的写入（例如同时更新数据集计数），事务可以保证要么全部成功，要么全部回滚，避免出现脏数据。

主要实现源码分析

下面截取自 pushDataListToTrainingQueue 函数的几个关键部分，帮助你理解代码细节。

模型验证和配置

export async function pushDataListToTrainingQueue({

  // 验证模型配置：检查agentModel和vectorModel的有效性，设置最大token数和权重。
  const { model, maxToken, weight } = await (async () => {
    const agentModelData = getLLMModel(agentModel);
    if (!agentModelData) {
      return Promise.reject(`File model ${agentModel} is inValid`);
    }
    const vectorModelData = getVectorModel(vectorModel);
    if (!vectorModelData) {
      return Promise.reject(`Vector model ${vectorModel} is inValid`);
    }
    
    if (trainingMode === TrainingModeEnum.chunk) {
      return {
        maxToken: vectorModelData.maxToken * 1.5,
        model: vectorModelData.model,
        weight: vectorModelData.weight
      };
    }

    if (trainingMode === TrainingModeEnum.qa || trainingMode === TrainingModeEnum.auto) {
      return {
        maxToken: agentModelData.maxContext * 0.8,
        model: agentModelData.model,
        weight: 0
      };
    }
  })();

过滤重复和超长内容

  // filter repeat or equal content
  const set = new Set();
  const filterResult: Record = {
    success: [],
    overToken: [],
    repeat: [],
    error: []
  };

数据清洗（去除空白字符）

  // format q and a, remove empty char
  data.forEach((item) => {
    item.q = simpleText(item.q);
    item.a = simpleText(item.a);

批量插入 MongoDB 训练表

  const batchSize = 200;
  const insertData = async (startIndex: number, session: ClientSession) => {
    const list = filterResult.success.slice(startIndex, startIndex + batchSize);

    if (list.length === 0) return;

    try {
      await MongoDatasetTraining.insertMany(
        list.map((item) => ({

常见问题（FAQ）

Q1：为什么数据集创建要分成两步？

A： 因为文件上传和预处理需要快速响应用户，而向量化或 QA 生成通常需要调用大模型，耗时较长。两步设计可以让用户立即看到上传成功，后台异步处理后续计算，提升用户体验，同时避免请求超时。

Q2：模型验证失败怎么办？

A： 函数会直接拒绝并返回错误信息（如 “File model xxx is inValid”）。你需要检查使用的模型 ID 是否在 FastGPT 的模型配置中已注册并可用。常见原因：模型名称拼写错误、模型未启用或 license 过期。

Q3：数据插入时 batchSize 为什么设定为 200？

A： 这个值是经验值，平衡了网络开销和 MongoDB 单次写入的性能。如果你的文档较大（例如每条包含长文本），可以适当调小（如 50）；如果文档较小，可以尝试调大（如 500）以获得更高吞吐。

Q4：如果批量插入失败，数据会丢失吗？

A： 不会丢失。代码会捕获异常并记录失败文档，然后逐条插入成功的数据。但你需要关注错误日志，手动处理那些始终插入失败的文档（可能是由于唯一键冲突或格式异常）。

小提示

• 提示一： 在调试或开发环境，可以开启 MongoDB 的监控日志，方便跟踪训练队列的写入情况。
• 提示二： 如果你希望提高数据处理的并行度，可以在第二步监听时设置多个消费者，但要小心 MongoDB Change Stream 的游标管理。
• 提示三： 数据预处理中的 simpleText 函数除了去除空白，还会对特殊字符做转义，确保问题/答案不会因为不可见字符导致模型处理异常。

总结

通过本文的分析，你应当清楚了 FastGPT 数据集创建的两阶段架构，以及第一步“文件上传与预处理”中模型验证、数据清洗、批量插入与事务处理的关键实现。正如我们所见，这一步仅将数据放入 MongoDB 的训练队列，真正的向量化或 QA 生成将在后续的监视器中触发。下一篇文章我们将深入分析第二步：MongoDB Change Stream 如何监听新增记录，并调用 generateQA 或 generateVector 完成最终的数据处理，敬请期待。