向量化系统阶段四实施规划——多数据源支持

📋 概述

Phase 4 实现统一的多数据源接入框架，支持从多种数据源获取数据并向量化：

✨ 数据源适配器框架: 统一的数据源接入接口
✨ 多种数据源支持: API、数据库、文件系统、对象存储、Web爬虫
✨ 配置化管理: 灵活的数据源配置和管理
✨ 定时拉取: 支持定时从数据源同步数据
✨ 数据转换: 统一的数据格式转换机制

🎯 核心目标

业务目标

目标	说明	优先级
扩展数据来源	不限于现有文档系统	P0
自动化数据接入	减少人工导入工作	P1
统一数据格式	标准化的数据处理流程	P0
灵活配置	支持多种数据源组合	P1

支持的数据源

数据源	说明	优先级	难度
REST API	从外部API获取数据	P0	低
MySQL数据库	直接从数据库表读取	P0	低
MinIO/S3	对象存储中的文件	P0	中
本地文件系统	NAS/共享文件夹	P1	低
PostgreSQL	PG数据库支持	P1	低
MongoDB	NoSQL数据库	P2	中
Elasticsearch	从ES索引读取	P2	中
Web爬虫	爬取网页内容	P3	高
SFTP/FTP	远程文件传输	P3	中

🏗️ 架构设计

整体架构

┌──────────────────────────────────────────────────────────────┐
│                        数据源层                               │
│  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐          │
│  │ API  │  │ MySQL│  │ MinIO│  │ File │  │ Web  │  ...     │
│  └───┬──┘  └───┬──┘  └───┬──┘  └───┬──┘  └───┬──┘          │
└──────┼─────────┼─────────┼─────────┼─────────┼──────────────┘
       │         │         │         │         │
       └─────────┼─────────┼─────────┼─────────┘
                 ▼         ▼         ▼         ▼
┌──────────────────────────────────────────────────────────────┐
│                  DataSourceAdapterFactory                     │
│              (数据源适配器工厂 - 策略模式)                     │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  根据数据源类型创建对应的适配器                         │  │
│  │  • API → ApiDataSourceAdapter                          │  │
│  │  • MYSQL → MysqlDataSourceAdapter                      │  │
│  │  •MINIO → MinioDataSourceAdapter                      │  │
│  └────────────────────────────────────────────────────────┘  │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│              DataSourceAdapter (适配器接口)                   │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  统一接口:                                              │  │
│  │  • connect()         - 连接数据源                       │  │
│  │  • fetchDocuments()  - 获取文档列表                     │  │
│  │  • fetchContent()    - 获取文档内容                     │  │
│  │  • supportsIncremental() - 是否支持增量                │  │
│  │  • close()           - 关闭连接                         │  │
│  └────────────────────────────────────────────────────────┘  │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│                 DataTransformer (数据转换器)                  │
│  ┌────────────────────────────────────────────────────────┐  │
│  │  将各种格式的数据转换为统一的 Document 对象             │  │
│  │  • JSON → Document                                     │  │
│  │  • Database Row → Document                             │  │
│  │  • File → Document                                     │  │
│  │  • HTML → Document                                     │  │
│  └────────────────────────────────────────────────────────┘  │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│            VectorizationService (向量化服务)                  │
│              调用 Phase 1-3 的向量化能力                      │
└──────────────────────────────────────────────────────────────┘

📦 核心组件设计

1. DataSourceAdapter - 数据源适配器接口

统一接口定义:

java

/**
 * 数据源适配器接口
 * 所有数据源适配器必须实现此接口
 */
public interface DataSourceAdapter {

    /**
     * 数据源类型
     */
    DataSourceType getType();

    /**
     * 连接到数据源
     * @param config 数据源配置
     * @return 是否连接成功
     */
    boolean connect(DataSourceConfig config);

    /**
     * 获取文档列表（分页）
     * @param offset 偏移量
     * @param limit 数量
     * @return 文档列表
     */
    List<SourceDocument> fetchDocuments(int offset, int limit);

    /**
     * 获取单个文档的完整内容
     * @param documentId 文档标识
     * @return 文档内容
     */
    String fetchContent(String documentId);

    /**
     * 是否支持增量同步
     */
    boolean supportsIncremental();

    /**
     * 增量同步（获取指定时间后的文档）
     * @param since 起始时间
     * @return 变更的文档列表
     */
    List<SourceDocument> fetchIncremental(LocalDateTime since);

    /**
     * 测试连接
     */
    boolean testConnection();

    /**
     * 关闭连接
     */
    void close();
}

数据源配置基类:

java

@Data
public class DataSourceConfig {
    /**
     * 数据源类型
     */
    private DataSourceType type;

    /**
     * 数据源名称
     */
    private String name;

    /**
     * 是否启用
     */
    private Boolean enabled = true;

    /**
     * 定时拉取配置
     */
    private ScheduleConfig schedule;

    /**
     * 具体配置（JSON，由各适配器自行解析）
     */
    private Map<String, Object> config;
}

2. 各数据源适配器实现

2.1 ApiDataSourceAdapter - REST API适配器

配置示例:

json

{
  "type": "API",
  "name": "外部知识库API",
  "config": {
    "baseUrl": "https://api.example.com",
    "listEndpoint": "/documents",
    "contentEndpoint": "/documents/{id}/content",
    "authentication": {
      "type": "Bearer",
      "token": "xxx"
    },
    "headers": {
      "X-API-Key": "xxx"
    },
    "pagination": {
      "type": "OFFSET",  // OFFSET/CURSOR/PAGE
      "pageSize": 100
    },
    "mapping": {
      "idField": "doc_id",
      "titleField": "title",
      "contentField": "content",
      "timeField": "updated_at"
    }
  }
}

核心逻辑:

1. 调用 listEndpoint 获取文档列表
2. 根据 pagination 配置实现分页
3. 对每个文档调用 contentEndpoint 获取完整内容
4. 根据 mapping 配置提取字段
5. 转换为统一的 Document 对象

2.2 MysqlDataSourceAdapter - MySQL适配器

配置示例:

json

{
  "type": "MYSQL",
  "name": "业务数据库",
  "config": {
    "host": "localhost",
    "port": 3306,
    "database": "business_db",
    "username": "readonly_user",
    "password": "xxx",
    "table": "articles",
    "columns": {
      "id": "id",
      "title": "title",
      "content": "content",
      "updateTime": "update_time"
    },
    "where": "status = 'published'",
    "incrementalColumn": "update_time"
  }
}

核心逻辑:

sql

-- 全量查询
SELECT id, title, content, update_time
FROM articles
WHERE status = 'published'
LIMIT 100 OFFSET 0;

-- 增量查询
SELECT id, title, content, update_time
FROM articles
WHERE status = 'published'
  AND update_time > '2025-12-01 00:00:00'
ORDER BY update_time ASC;

2.3 MinioDataSourceAdapter - MinIO/S3适配器

配置示例:

json

{
  "type": "MINIO",
  "name": "MinIO对象存储",
  "config": {
    "endpoint": "http://minio:9000",
    "accessKey": "xxx",
    "secretKey": "xxx",
    "bucket": "documents",
    "prefix": "knowledge/",
    "fileTypes": ["pdf", "docx", "txt", "md"],
    "recursive": true
  }
}

核心逻辑:

1. 连接到MinIO
2. 列出指定bucket和prefix下的文件
3. 根据fileTypes过滤文件
4. 下载文件内容
5. 根据文件类型解析（PDF/Word/Text）
6. 转换为Document对象

2.4 FileSystemDataSourceAdapter - 本地文件系统适配器

配置示例:

json

{
  "type": "FILE_SYSTEM",
  "name": "共享文件夹",
  "config": {
    "basePath": "/data/documents",
    "patterns": ["*.pdf", "*.docx", "*.txt"],
    "recursive": true,
    "encoding": "UTF-8"
  }
}

2.5 WebCrawlerAdapter - Web爬虫适配器（可选）

配置示例:

json

{
  "type": "WEB_CRAWLER",
  "name": "官网知识库爬虫",
  "config": {
    "startUrls": ["https://example.com/docs"],
    "urlPatterns": ["https://example.com/docs/*"],
    "maxDepth": 3,
    "maxPages": 1000,
    "selector": "article.content",
    "excludeSelectors": ["nav", "footer"],
    "respectRobotsTxt": true,
    "userAgent": "KnowledgeBot/1.0"
  }
}

3. DataSourceAdapterFactory - 适配器工厂

职责: 根据配置创建对应的数据源适配器

java

@Component
public class DataSourceAdapterFactory {

    // 注册所有适配器
    private final Map<DataSourceType, Class<? extends DataSourceAdapter>> adapters = Map.of(
        DataSourceType.API, ApiDataSourceAdapter.class,
        DataSourceType.MYSQL, MysqlDataSourceAdapter.class,
        DataSourceType.MINIO, MinioDataSourceAdapter.class,
        DataSourceType.FILE_SYSTEM, FileSystemDataSourceAdapter.class
    );

    /**
     * 创建适配器实例
     */
    public DataSourceAdapter createAdapter(DataSourceConfig config) {
        Class<? extends DataSourceAdapter> adapterClass = adapters.get(config.getType());

        if (adapterClass == null) {
            throw new IllegalArgumentException("不支持的数据源类型: " + config.getType());
        }

        // 使用Spring容器创建实例（支持依赖注入）
        DataSourceAdapter adapter = ApplicationContextHolder.getBean(adapterClass);
        adapter.connect(config);

        return adapter;
    }
}

4. DataSourceSyncService - 数据源同步服务

职责: 从数据源拉取数据并向量化

核心流程:

syncDataSource(dataSourceConfig):
    1. 创建数据源适配器
        adapter = factory.createAdapter(config)

    2. 测试连接
        if (!adapter.testConnection()) {
            throw exception
        }

    3. 拉取文档列表（分页）
        int offset = 0
        int limit = 100
        while (true) {
            documents = adapter.fetchDocuments(offset, limit)
            if (documents.isEmpty()) break

            // 批量处理
            processBatch(documents)

            offset += limit
        }

    4. 转换为标准Document
        for (sourceDoc : documents) {
            document = transformer.transform(sourceDoc)
            // 保存到 kb_document 表
            documentService.save(document)
        }

    5. 触发向量化
        vectorizationService.vectorizeDataset(datasetId)

    6. 记录同步历史
        saveSyncHistory(...)

5. DataTransformer - 数据转换器

职责: 将各种数据格式转换为统一的Document对象

java

public interface DataTransformer {

    /**
     * 转换为标准Document
     */
    KbDocumentEntity transform(SourceDocument source);

    /**
     * 支持的数据源类型
     */
    DataSourceType supportedType();
}

转换规则:

SourceDocument (来自数据源) → KbDocumentEntity (系统实体)

字段映射:
- sourceId → 保存到 metadata
- title → name
- content → 保存到MinIO，路径存到 filePath
- updateTime → updateTime
- metadata → 额外元数据保存到 metadata JSON字段

📊 数据库设计

数据源配置表

sql

CREATE TABLE `kb_data_source` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL COMMENT '数据源名称',
  `type` varchar(32) NOT NULL COMMENT '数据源类型: API/MYSQL/MINIO/FILE_SYSTEM',
  `dataset_id` bigint NOT NULL COMMENT '关联的数据集ID',
  `config` text NOT NULL COMMENT '数据源配置（JSON）',
  `enabled` tinyint(1) DEFAULT 1 COMMENT '是否启用',
  `sync_enabled` tinyint(1) DEFAULT 0 COMMENT '是否启用定时同步',
  `sync_cron` varchar(64) COMMENT '同步Cron表达式',
  `last_sync_time` datetime COMMENT '最后同步时间',
  `last_sync_status` varchar(32) COMMENT '最后同步状态',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_dataset_id` (`dataset_id`),
  KEY `idx_type` (`type`)
) ENGINE=InnoDB COMMENT='数据源配置表';

数据源同步历史表

sql

CREATE TABLE `kb_data_source_sync_history` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `data_source_id` bigint NOT NULL COMMENT '数据源ID',
  `start_time` datetime NOT NULL COMMENT '开始时间',
  `end_time` datetime COMMENT '结束时间',
  `duration_ms` bigint COMMENT '耗时（毫秒）',
  `fetched_count` int DEFAULT 0 COMMENT '获取文档数',
  `imported_count` int DEFAULT 0 COMMENT '导入文档数',
  `failed_count` int DEFAULT 0 COMMENT '失败文档数',
  `status` varchar(32) NOT NULL COMMENT '状态: SUCCESS/FAILED/PARTIAL',
  `error_message` text COMMENT '错误信息',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_data_source_id` (`data_source_id`),
  KEY `idx_start_time` (`start_time`)
) ENGINE=InnoDB COMMENT='数据源同步历史表';

📖 接口设计

1. 创建数据源

POST /ai/vectorization/data-source

请求:
{
  "name": "外部知识库API",
  "type": "API",
  "datasetId": 1,
  "config": { ... },
  "syncEnabled": true,
  "syncCron": "0 0 2 * * ?"
}

响应:
{
  "code": 0,
  "data": {
    "id": 1,
    "name": "外部知识库API",
    ...
  }
}

2. 测试数据源连接

POST /ai/vectorization/data-source/test

请求: DataSourceConfig

响应:
{
  "code": 0,
  "data": {
    "success": true,
    "message": "连接成功",
    "sampleDocuments": 5
  }
}

3. 手动同步数据源

POST /ai/vectorization/data-source/{id}/sync

响应: 同步任务信息

4. 查询同步历史

GET /ai/vectorization/data-source/{id}/sync-history

请求参数:
- page, size

响应: 同步历史列表

🔄 实施方案

Phase 4.1: 适配器框架（Week 1-2）

任务清单:

✅ 定义 DataSourceAdapter 接口
✅ 实现 DataSourceAdapterFactory
✅ 创建数据库表
✅ 实现 DataSourceSyncService 基础框架
✅ 实现 DataTransformer

Phase 4.2: API适配器（Week 3）

任务清单:

✅ 实现 ApiDataSourceAdapter
✅ 支持多种认证方式（Bearer, ApiKey, Basic）
✅ 支持多种分页策略（Offset, Cursor, Page）
✅ 测试验证

Phase 4.3: 数据库适配器（Week 4）

任务清单:

✅ 实现 MysqlDataSourceAdapter
✅ 实现 PostgresDataSourceAdapter
✅ 支持增量同步（基于时间字段）
✅ 连接池管理

Phase 4.4: 对象存储适配器（Week 5）

任务清单:

✅ 实现 MinioDataSourceAdapter
✅ 支持S3兼容存储
✅ 文件类型识别和解析
✅ 大文件处理优化

Phase 4.5: 定时同步（Week 6）

任务清单:

✅ 集成XXL-Job
✅ 为每个数据源创建定时任务
✅ 实现任务的启动/停止/修改

⚙️ 配置示例

application.yml

yaml

vectorization:
  data-source:
    # 连接池配置
    connection-pool:
      max-total: 10
      max-idle: 5
      min-idle: 2

    # 超时配置
    timeout:
      connect: 10000      # 连接超时（毫秒）
      read: 30000         # 读取超时
      write: 30000        # 写入超时

    # 批处理配置
    batch:
      size: 100           # 每批处理文档数
      parallel: true      # 是否并行处理

    # 重试配置
    retry:
      max-attempts: 3
      backoff-delay: 1000

⚠️ 注意事项

1. 安全性

凭证加密: 数据源密码使用加密存储
权限控制: 使用只读账号连接数据源
网络隔离: 生产环境使用专用网络

2. 性能

批量拉取: 使用分页批量获取，避免OOM
连接复用: 使用连接池，避免频繁创建连接
异步处理: 大数据量使用异步任务队列

3. 数据一致性

幂等性: 同一文档多次同步结果一致
去重: 根据 sourceId 去重
增量同步: 支持增量避免重复处理

4. 容错性

重试机制: 网络错误自动重试
部分成功: 允许部分文档失败，记录日志
降级策略: 外部数据源不可用时降级处理

📈 扩展性

新增数据源适配器步骤

实现 DataSourceAdapter 接口
在 DataSourceAdapterFactory 注册
添加对应的 DataTransformer
编写单元测试
更新文档

示例：添加MongoDB适配器

java

@Component
public class MongoDataSourceAdapter implements DataSourceAdapter {

    @Override
    public DataSourceType getType() {
        return DataSourceType.MONGODB;
    }

    @Override
    public boolean connect(DataSourceConfig config) {
        // 连接逻辑
    }

    @Override
    public List<SourceDocument> fetchDocuments(int offset, int limit) {
        // 查询MongoDB集合
    }

    // ... 实现其他方法
}

🔄 下一步：阶段五

Phase 5 将实现：

✨ 详细的系统监控指标
✨ 可视化Dashboard
✨ 性能自动优化
✨ 告警和通知机制

最后更新: 2025-12-12 版本: Phase 4 规划 v1.0.0

向量化系统阶段四实施规划——多数据源支持 ​

📋 概述 ​

🎯 核心目标 ​

业务目标 ​

支持的数据源 ​

🏗️ 架构设计 ​

整体架构 ​

📦 核心组件设计 ​

1. DataSourceAdapter - 数据源适配器接口 ​

2. 各数据源适配器实现 ​

2.1 ApiDataSourceAdapter - REST API适配器 ​

2.2 MysqlDataSourceAdapter - MySQL适配器 ​

2.3 MinioDataSourceAdapter - MinIO/S3适配器 ​

2.4 FileSystemDataSourceAdapter - 本地文件系统适配器 ​

2.5 WebCrawlerAdapter - Web爬虫适配器（可选） ​

3. DataSourceAdapterFactory - 适配器工厂 ​

4. DataSourceSyncService - 数据源同步服务 ​

5. DataTransformer - 数据转换器 ​

📊 数据库设计 ​

数据源配置表 ​

数据源同步历史表 ​

📖 接口设计 ​

1. 创建数据源 ​

2. 测试数据源连接 ​

3. 手动同步数据源 ​

4. 查询同步历史 ​

🔄 实施方案 ​

Phase 4.1: 适配器框架（Week 1-2） ​

Phase 4.2: API适配器（Week 3） ​

Phase 4.3: 数据库适配器（Week 4） ​

Phase 4.4: 对象存储适配器（Week 5） ​

Phase 4.5: 定时同步（Week 6） ​

⚙️ 配置示例 ​

application.yml ​

⚠️ 注意事项 ​

1. 安全性 ​

2. 性能 ​

3. 数据一致性 ​

4. 容错性 ​

📈 扩展性 ​

新增数据源适配器步骤 ​

示例：添加MongoDB适配器 ​

🔄 下一步：阶段五 ​

向量化系统阶段四实施规划——多数据源支持

📋 概述

🎯 核心目标

业务目标

支持的数据源

🏗️ 架构设计

整体架构

📦 核心组件设计

1. DataSourceAdapter - 数据源适配器接口

2. 各数据源适配器实现

2.1 ApiDataSourceAdapter - REST API适配器

2.2 MysqlDataSourceAdapter - MySQL适配器

2.3 MinioDataSourceAdapter - MinIO/S3适配器

2.4 FileSystemDataSourceAdapter - 本地文件系统适配器

2.5 WebCrawlerAdapter - Web爬虫适配器（可选）

3. DataSourceAdapterFactory - 适配器工厂

4. DataSourceSyncService - 数据源同步服务

5. DataTransformer - 数据转换器

📊 数据库设计

数据源配置表

数据源同步历史表

📖 接口设计

1. 创建数据源

2. 测试数据源连接

3. 手动同步数据源

4. 查询同步历史

🔄 实施方案

Phase 4.1: 适配器框架（Week 1-2）

Phase 4.2: API适配器（Week 3）

Phase 4.3: 数据库适配器（Week 4）

Phase 4.4: 对象存储适配器（Week 5）

Phase 4.5: 定时同步（Week 6）

⚙️ 配置示例

application.yml

⚠️ 注意事项

1. 安全性

2. 性能

3. 数据一致性

4. 容错性

📈 扩展性

新增数据源适配器步骤

示例：添加MongoDB适配器

🔄 下一步：阶段五