dify实现原理分析-上传文件创建知识库总体流程

本文介绍了dify在上传文件构建知识库的总体流程。dify提供三种方式来创建构建知识库，本文主要介绍了文件上传的总体流程，索引的构建实现流程在后面的文章中进行分析。

一铭

5857人浏览 · 2025-01-06 07:25:35

一铭 · 2025-01-06 07:25:35 发布

dify实现原理分析-上传文件创建知识库

概述

本文介绍上传文件到知识库的处理逻辑。本文主要介绍文件上传的主要实现流程。

知识库文件上传API介绍

文件上传API

/datasets/{dataset_id}/document/create-by-text

API的处理类设置

api.add_resource(
    DocumentAddByTextApi,
    "/datasets/<uuid:dataset_id>/document/create_by_text",
    "/datasets/<uuid:dataset_id>/document/create-by-text",
)

文本上传的处理实现：DocumentAddByTextApi.post

class DocumentAddByTextApi(DatasetApiResource):
    """Resource for documents."""
    @cloud_edition_billing_resource_check("vector_space", "dataset")
    @cloud_edition_billing_resource_check("documents", "dataset")
    def post(self, tenant_id, dataset_id):
        # 获取用户请求参数，解析参数，验证参数的合法性
        # 进一步检查参数的合法性
        DocumentService.document_create_args_validate(args)
        # 启动文档分割任务，(1)分割文档构建嵌入向量并保存到向量库中；(2)提取关键词，并保存到数据库中。
        DocumentService.save_document_with_dataset_id(...)
        # ...

post函数的主要实现逻辑

参数验证：

使用 reqparse.RequestParser() 定义并验证请求中的参数，如 name, text, process_rule, 等等。
检查必要的参数是否为非空，并进行必要的数据类型转换。

在数据库中验证数据集是否存在

查询指定的 dataset 对象，确保该数据集存在。如果不存在则抛出错误。

注：在上传文档时，需要先创建一个数据集。上传文件后，首先要检查数据集是否存在。

索引技术类型验证：

确保提供的或默认的索引技术类型是有效的。如果不有效则抛出错误。

上传文件并添加数据源信息：

将传入的文本内容通过 FileService.upload_text() 方法上传到文件系统，dify支持多种文件存储系统，比如：aliyun_oss,local等，并在配置数据库中添加一条文件上传信息的记录。
创建一个数据源字典，指定文件类型及其信息列表。这里的数据源类型被设置为：“upload_file”。

参数验证和保存文档：

调用 DocumentService.document_create_args_validate(args) 验证所有传入参数的有效性。
调用 DocumentService.save_document_with_dataset_id() 方法将文档保存到数据库中，并启动索引构建任务，构建文档索引。该函数是文档上传过程并对文档进行处理的核心。

异常处理：

捕获并处理可能的 ProviderTokenNotInitError 异常，抛出自定义的 ProviderNotInitializeError。

返回结果：

将成功创建的文档和相关的批处理信息以 JSON 格式返回，并附带 HTTP 状态码 200。

post函数功能小结

提供一个 REST API 接口，允许通过上传文件的文本内容创建新的知识库（Dataset）。文件保存到存储库（可设置，比如：S3等）中。
通过Celery任务队列，启动异步索引构建任务，来把文档切分成小块(chunk)，并把分块构建成嵌入向量，保存到向量数据库中。
把文档和分块信息（元数据信息）保存到数据库中，抽取文本的关键词，并把关键词保存到数据库中。
确保输入数据的有效性和一致性，防止错误输入导致的数据损坏或业务逻辑混乱。
实现与数据库的交互，对新创建的文档进行持久化存储。

save_document_with_dataset_id函数

总体实现逻辑分析

从以上过程可以看出，其主要功能是在save_document_with_dataset_id函数中完成。现在，我们来看一下该函数的实现逻辑。

从以上流程图可以基本上看出该函数的实现逻辑。

参数检查和限制验证：检查上传文档数量是否超出了限制；
dify提供了三种构建知识库的方式，会根据不同的方式来处理文件，然后把处理完成的文档保存到Document列表中。
- 上传文件：upload_file
- notion导入：notion_import
- 站点爬虫：website_crawl
数据集配置更新：根据上传的方式和文档来更新数据集的配置信息，包括数据集的分组绑定信息
文档处理逻辑：分别按以上三种方式来构建插入数据库的文档上传记录也就是Document对象，对于上传文件这种方式，会对每个文件创建一个Document对象，该对象用来保存一个文档的相关信息和数据。
构建文档索引：通过异步的方式来构建文档中数据的索引。在构建索引的过程中，会对文档中的内容进行切分。

具体代码实现分析

上传文件创建知识库的主要逻辑在函数：save_document_with_dataset_id 中实现。该函数实现了保存文档到数据集的逻辑。

    @staticmethod
    def save_document_with_dataset_id(
        dataset: Dataset,
        document_data: dict,
        account: Account | Any,
        dataset_process_rule: Optional[DatasetProcessRule] = None,
        created_from: str = "web",
    ):

1.参数检查和限制验证：检查文档数量是否超出了限制

    features = FeatureService.get_features(current_user.current_tenant_id)
    if features.billing.enabled:
        # 检查文档数量限制
        # 根据不同数据源类型(upload_file/notion_import/website_crawl)计算文档数
        # 验证是否超过批量上传限制

2.数据集配置更新

# 更新数据集的数据源类型
if not dataset.data_source_type:
    dataset.data_source_type = document_data["data_source"]["type"]

# 设置索引技术
if not dataset.indexing_technique:
    # 配置高质量索引的相关设置
    # 设置嵌入模型和检索模型

3.文档处理逻辑

根据是否存在原始文档ID，分两种情况：

（1）更新已有文档

if document_data.get("original_document_id"):
    document = DocumentService.update_document_with_dataset_id(...)

（2）创建新文档

else:
    # 生成批次号
    batch = time.strftime("%Y%m%d%H%M%S") + str(random.randint(100000, 999999))
    
    # 保存处理规则
    if not dataset_process_rule:
        # 创建自定义或自动处理规则

4.根据数据源类型处理文档

使用 Redis 锁确保并发安全：

（1）数据源为"上传文件"的处理

if document_data["data_source"]["type"] == "upload_file":
    # 处理文件列表
    # 检查文档是否已经存在，更新数据库中该文档上传信息的记录
    # 创建新文档记录:创建一个Document对象来记录文档上传的信息，然后并保存到数据库中
    # 处理上传文档
    DocumentService.build_document(...)

（2）Notion 导入

elif document_data["data_source"]["type"] == "notion_import":
    # 处理 Notion 页面
    # 检查现有页面
    # 创建新文档记录
    # 清理未选择的文档

（3）网站爬取

elif document_data["data_source"]["type"] == "website_crawl":
    # 处理 URL 列表
    # 创建文档记录

5.索引构建任务

启动索引构建任务，通过Celery的dataset队列中来触发索引构建流程。

# 触发文档索引构建任务，任务异步执行
if document_ids:
    document_indexing_task.delay(dataset.id, document_ids)
# 触发重复文档索引任务
if duplicate_document_ids:
    duplicate_document_indexing_task.delay(dataset.id, duplicate_document_ids)