向量化系统阶段三实施规划——增量更新

📋 概述

Phase 3 实现智能增量更新机制，避免全量重复向量化，提升系统效率：

✨ 文档变更检测: 自动识别新增、修改、删除的文档
✨ 智能增量同步: 只处理变更的文档，节省资源
✨ 定时自动同步: 配置化的定时增量同步策略
✨ 版本管理: 文档版本追踪和回滚能力
✨ 冲突解决: 处理并发修改和向量更新冲突

🎯 核心目标

业务目标

目标	说明	指标
减少重复处理	只向量化变更的文档	处理量降低 70%+
提升同步速度	增量同步比全量快	速度提升 5-10倍
降低成本	减少 Embedding API 调用	成本降低 60%+
实时性	文档变更后快速同步	延迟 < 5分钟

技术目标

实现文档版本管理机制
支持多种变更检测策略
提供灵活的同步调度配置
保证数据一致性

🏗️ 架构设计

整体架构

┌─────────────────────────────────────────────────────────────┐
│                      文档变更来源                            │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │ 用户编辑文档  │  │ 批量导入文档  │  │ API创建文档   │      │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘      │
└─────────┼──────────────────┼──────────────────┼─────────────┘
          │                  │                  │
          └──────────────────┼──────────────────┘
                             ▼
┌─────────────────────────────────────────────────────────────┐
│              DocumentChangeDetector (变更检测器)             │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  检测策略:                                            │  │
│  │  • 基于时间戳 (updateTime > lastSyncTime)           │  │
│  │  • 基于版本号 (version > lastSyncVersion)           │  │
│  │  • 基于哈希值 (contentHash != lastHash)             │  │
│  │  • 基于事件监听 (实时监听数据库变更)                 │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────┬────────────────────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────┐
│           IncrementalSyncService (增量同步服务)              │
│  ┌──────────────────────────────────────────────────────┐  │
│  │  同步流程:                                            │  │
│  │  1. 检测变更文档                                      │  │
│  │  2. 分类处理 (新增/修改/删除)                        │  │
│  │  3. 执行增量向量化                                    │  │
│  │  4. 更新同步记录                                      │  │
│  │  5. 记录版本信息                                      │  │
│  └──────────────────────────────────────────────────────┘  │
└────────────────────────────┬────────────────────────────────┘
                             │
         ┌───────────────────┼───────────────────┐
         ▼                   ▼                   ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 新增文档         │ │ 修改文档         │ │ 删除文档         │
│ → 向量化        │ │ → 删除旧向量    │ │ → 删除向量      │
│   → 存储        │ │   → 重新向量化  │ │   → 更新索引    │
└─────────────────┘ └─────────────────┘ └─────────────────┘
                             │
                             ▼
┌─────────────────────────────────────────────────────────────┐
│              SyncHistoryService (同步历史服务)               │
│  记录每次同步的: 时间、文档数、耗时、状态、版本号            │
└─────────────────────────────────────────────────────────────┘

📦 核心组件设计

1. DocumentChangeDetector - 变更检测器

职责: 检测文档的新增、修改、删除

检测策略:

策略1: 基于时间戳（简单高效，推荐）

检测逻辑:
1. 查询数据集的最后同步时间 lastSyncTime
2. 查询 updateTime > lastSyncTime 的文档
3. 返回变更文档列表

优点: 实现简单，性能好
缺点: 依赖时间戳准确性

策略2: 基于版本号（精确可靠）

检测逻辑:
1. 为每个文档维护版本号（version字段）
2. 记录数据集最后同步的版本号 lastSyncVersion
3. 查询 version > lastSyncVersion 的文档

优点: 精确可靠，支持版本回溯
缺点: 需要维护版本号字段

策略3: 基于内容哈希（最精确）

检测逻辑:
1. 计算文档内容的 MD5/SHA256 哈希值
2. 比较当前哈希值与上次同步时的哈希值
3. 哈希值不同则认为内容已变更

优点: 最精确，能检测细微变更
缺点: 计算开销大，需要存储哈希值

策略4: 基于事件监听（实时性最好）

实现方式:
1. 监听 kb_document 表的 INSERT/UPDATE/DELETE 事件
2. 通过 MyBatis-Plus 拦截器或数据库触发器
3. 实时触发增量向量化

优点: 实时性最好，无需轮询
缺点: 实现复杂，对数据库有要求

推荐方案: 策略1（时间戳） + 策略2（版本号） 组合使用

2. IncrementalSyncService - 增量同步服务

职责: 执行增量同步逻辑

核心方法:

incrementalSync(datasetId, since):
    1. 检测变更文档
        - 新增文档: status=NEW or createTime > since
        - 修改文档: updateTime > since && id IN (已向量化文档)
        - 删除文档: status=DELETED or 物理删除

    2. 分类处理
        新增文档:
            → 直接向量化
            → 记录版本信息

        修改文档:
            → 删除旧向量 (根据documentId)
            → 重新向量化
            → 更新版本信息

        删除文档:
            → 删除向量 (根据documentId)
            → 删除版本记录

    3. 批量处理
        - 使用 Phase 2 的异步任务队列
        - 按优先级处理

    4. 更新同步记录
        - 记录本次同步的时间点
        - 记录处理的文档数量
        - 记录同步状态和耗时

    5. 返回同步结果
        - 新增/修改/删除文档数
        - 成功/失败统计
        - 耗时信息

3. IncrementalSyncScheduler - 定时同步调度器

职责: 定时触发增量同步任务

调度策略:

yaml

# 数据集级别的增量同步配置
dataset_sync_config:
  dataset_id: 1
  sync_enabled: true
  sync_interval: 60            # 同步间隔（分钟）
  sync_cron: "0 */5 * * * ?"  # Cron表达式（可选）
  priority: 5                  # 任务优先级
  auto_start: true             # 是否自动启动

实现方式:

方式1: @Scheduled 定时任务
  - 使用 Spring @Scheduled 注解
  - 定时扫描所有启用增量同步的数据集
  - 提交增量同步任务

方式2: Quartz/XXL-Job
  - 使用项目现有的定时任务框架
  - 为每个数据集创建独立的定时任务
  - 支持动态创建/删除任务

推荐: 方式2（Quartz），更灵活可控

4. DocumentVersionService - 文档版本服务

职责: 管理文档的版本信息

版本记录表设计:

sql

CREATE TABLE `kb_document_version` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `document_id` bigint NOT NULL COMMENT '文档ID',
  `dataset_id` bigint NOT NULL COMMENT '数据集ID',
  `version` int NOT NULL COMMENT '版本号',
  `content_hash` varchar(64) COMMENT '内容哈希（MD5/SHA256）',
  `chunk_count` int DEFAULT 0 COMMENT '分段数量',
  `vectorized_at` datetime COMMENT '向量化时间',
  `vectorization_config` text COMMENT '向量化配置（JSON）',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_doc_version` (`document_id`, `version`),
  KEY `idx_dataset_id` (`dataset_id`),
  KEY `idx_vectorized_at` (`vectorized_at`)
) ENGINE=InnoDB COMMENT='文档版本记录表';

核心功能:

记录版本: 每次向量化后记录版本信息
版本查询: 查询文档的历史版本
版本对比: 对比两个版本的差异
版本回滚: 恢复到指定版本的向量（可选）

5. SyncHistoryService - 同步历史服务

职责: 记录每次增量同步的历史

同步历史表设计:

sql

CREATE TABLE `kb_sync_history` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `dataset_id` bigint NOT NULL COMMENT '数据集ID',
  `sync_type` varchar(32) NOT NULL COMMENT '同步类型: FULL/INCREMENTAL',
  `task_id` varchar(64) COMMENT '关联的任务ID',
  `start_time` datetime NOT NULL COMMENT '开始时间',
  `end_time` datetime COMMENT '结束时间',
  `duration_ms` bigint COMMENT '耗时（毫秒）',
  `new_documents` int DEFAULT 0 COMMENT '新增文档数',
  `updated_documents` int DEFAULT 0 COMMENT '修改文档数',
  `deleted_documents` int DEFAULT 0 COMMENT '删除文档数',
  `success_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_dataset_id` (`dataset_id`),
  KEY `idx_start_time` (`start_time`)
) ENGINE=InnoDB COMMENT='同步历史记录表';

🔄 实施方案

Phase 3.1: 基础增量检测（Week 1-2）

目标: 实现基于时间戳的简单增量检测

任务清单:

✅ 创建数据库表: kb_document_version, kb_sync_history
✅ 实现 DocumentChangeDetector（时间戳策略）
✅ 实现 IncrementalSyncService 基础逻辑
✅ 实现增量同步 API
✅ 测试验证

核心SQL:

sql

-- 检测变更文档（基于时间戳）
SELECT id, name, update_time
FROM kb_document
WHERE dataset_id = #{datasetId}
  AND update_time > #{lastSyncTime}
  AND status != 'DELETED'
ORDER BY update_time ASC;

-- 记录同步历史
INSERT INTO kb_sync_history (dataset_id, sync_type, start_time, ...)
VALUES (?, 'INCREMENTAL', NOW(), ...);

Phase 3.2: 版本管理（Week 3）

目标: 实现文档版本追踪

任务清单:

✅ 为 kb_document 表添加 version 字段
✅ 实现 DocumentVersionService
✅ 每次向量化后记录版本信息
✅ 实现版本查询和对比 API

版本号策略:

自增版本号: 每次修改 version++
初始版本: version = 1
乐观锁: UPDATE ... SET version = version + 1 WHERE id = ? AND version = ?

Phase 3.3: 定时增量同步（Week 4）

目标: 实现自动定时增量同步

任务清单:

✅ 实现 IncrementalSyncScheduler
✅ 为数据集添加增量同步配置
✅ 集成到现有定时任务框架（XXL-Job）
✅ 实现任务的启动/停止/修改配置

定时任务配置:

java

// 数据集配置表添加字段
ALTER TABLE kb_dataset
ADD COLUMN `incremental_sync_enabled` tinyint(1) DEFAULT 0 COMMENT '是否启用增量同步',
ADD COLUMN `sync_interval_minutes` int DEFAULT 60 COMMENT '同步间隔（分钟）',
ADD COLUMN `last_sync_time` datetime COMMENT '最后同步时间';

Phase 3.4: 事件监听（Week 5，可选）

目标: 实现实时变更监听

实现方案:

方案1: MyBatis-Plus 拦截器

java

@Intercepts({
    @Signature(type = Executor.class, method = "update", args = {MappedStatement.class, Object.class})
})
public class DocumentChangeInterceptor implements Interceptor {
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        // 检测到文档表的变更
        // 发送变更事件到消息队列
        // 触发增量向量化
    }
}

方案2: Spring Event

java

// 发布事件
applicationEventPublisher.publishEvent(new DocumentChangedEvent(documentId));

// 监听事件
@EventListener
public void onDocumentChanged(DocumentChangedEvent event) {
    // 触发增量向量化
    incrementalSyncService.syncDocument(event.getDocumentId());
}

推荐: 方案2（Spring Event），实现简单，对现有代码侵入小

📊 接口设计

1. 增量同步API

POST /ai/vectorization/sync/incremental

请求参数:
{
  "datasetId": 1,
  "since": "2025-12-01T00:00:00",  // 可选，不传则使用上次同步时间
  "force": false                   // 是否强制全量同步
}

响应:
{
  "code": 0,
  "data": {
    "taskId": "xxx",
    "datasetId": 1,
    "syncType": "INCREMENTAL",
    "newDocuments": 5,
    "updatedDocuments": 3,
    "deletedDocuments": 1,
    "status": "RUNNING"
  }
}

2. 查询同步历史

GET /ai/vectorization/sync/history

请求参数:
- datasetId: 数据集ID
- page: 页码
- size: 每页数量

响应: 同步历史列表

3. 获取文档版本历史

GET /ai/vectorization/document/{documentId}/versions

响应: 文档的所有版本记录

4. 配置数据集增量同步

POST /ai/vectorization/dataset/{datasetId}/sync-config

请求参数:
{
  "enabled": true,
  "intervalMinutes": 60,
  "priority": 5,
  "autoStart": true
}

⚙️ 配置示例

application.yml

yaml

vectorization:
  incremental:
    # 变更检测策略
    detection-strategy: TIMESTAMP  # TIMESTAMP/VERSION/HASH/EVENT

    # 定时同步配置
    scheduler:
      enabled: true
      default-interval: 60         # 默认同步间隔（分钟）
      check-interval: 60000        # 检查间隔（毫秒）

    # 版本管理
    version:
      enabled: true
      keep-history: true           # 是否保留历史版本
      max-versions: 10             # 最多保留版本数

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

🎯 性能优化

1. 批量检测优化

sql

-- 使用索引加速查询
CREATE INDEX idx_dataset_update ON kb_document(dataset_id, update_time);

-- 分页查询避免大结果集
SELECT id FROM kb_document
WHERE dataset_id = ? AND update_time > ?
ORDER BY update_time
LIMIT 100 OFFSET 0;

2. 并行处理

将变更文档分批，并行提交到任务队列:
- 每批100个文档
- 优先级按变更时间排序
- 利用 Phase 2 的异步调度能力

3. 缓存优化

缓存最后同步时间和版本号:
- Redis Key: sync:dataset:{datasetId}:last_time
- TTL: 7天
- 减少数据库查询

⚠️ 注意事项

1. 数据一致性

问题: 并发修改导致版本冲突
方案: 使用乐观锁 + 版本号控制

2. 删除文档处理

软删除: 推荐方案，标记 status=DELETED，保留记录
物理删除: 需要在删除前触发向量删除

3. 同步失败处理

记录详细的失败原因
支持手动重试
失败文档单独标记

4. 性能影响

增量检测查询需要索引支持
版本表会持续增长，需要定期清理
实时监听方案对性能有一定影响

📈 监控指标

关键指标

指标	说明	目标值
增量检测耗时	检测变更文档的时间	< 1s
增量同步耗时	完成增量同步的时间	< 全量的20%
API调用减少率	相比全量的API调用减少	> 60%
同步准确率	检测到的变更准确性	> 99%

🔄 下一步：阶段四

Phase 4 将实现：

✨ 多数据源适配器框架
✨ API、数据库、文件系统等数据源支持
✨ 数据源配置管理
✨ 统一的数据获取接口

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

向量化系统阶段三实施规划——增量更新 ​

📋 概述 ​

🎯 核心目标 ​

业务目标 ​

技术目标 ​

🏗️ 架构设计 ​

整体架构 ​

📦 核心组件设计 ​

1. DocumentChangeDetector - 变更检测器 ​

策略1: 基于时间戳（简单高效，推荐） ​

策略2: 基于版本号（精确可靠） ​

策略3: 基于内容哈希（最精确） ​

策略4: 基于事件监听（实时性最好） ​

2. IncrementalSyncService - 增量同步服务 ​

3. IncrementalSyncScheduler - 定时同步调度器 ​

4. DocumentVersionService - 文档版本服务 ​

5. SyncHistoryService - 同步历史服务 ​

🔄 实施方案 ​

Phase 3.1: 基础增量检测（Week 1-2） ​

Phase 3.2: 版本管理（Week 3） ​

Phase 3.3: 定时增量同步（Week 4） ​

Phase 3.4: 事件监听（Week 5，可选） ​

📊 接口设计 ​

1. 增量同步API ​

2. 查询同步历史 ​

3. 获取文档版本历史 ​

4. 配置数据集增量同步 ​

⚙️ 配置示例 ​

application.yml ​

🎯 性能优化 ​

1. 批量检测优化 ​

2. 并行处理 ​

3. 缓存优化 ​

⚠️ 注意事项 ​

1. 数据一致性 ​

2. 删除文档处理 ​

3. 同步失败处理 ​

4. 性能影响 ​

📈 监控指标 ​

关键指标 ​

🔄 下一步：阶段四 ​

向量化系统阶段三实施规划——增量更新

📋 概述

🎯 核心目标

业务目标

技术目标

🏗️ 架构设计

整体架构

📦 核心组件设计

1. DocumentChangeDetector - 变更检测器

策略1: 基于时间戳（简单高效，推荐）

策略2: 基于版本号（精确可靠）

策略3: 基于内容哈希（最精确）

策略4: 基于事件监听（实时性最好）

2. IncrementalSyncService - 增量同步服务

3. IncrementalSyncScheduler - 定时同步调度器

4. DocumentVersionService - 文档版本服务

5. SyncHistoryService - 同步历史服务

🔄 实施方案

Phase 3.1: 基础增量检测（Week 1-2）

Phase 3.2: 版本管理（Week 3）

Phase 3.3: 定时增量同步（Week 4）

Phase 3.4: 事件监听（Week 5，可选）

📊 接口设计

1. 增量同步API

2. 查询同步历史

3. 获取文档版本历史

4. 配置数据集增量同步

⚙️ 配置示例

application.yml

🎯 性能优化

1. 批量检测优化

2. 并行处理

3. 缓存优化

⚠️ 注意事项

1. 数据一致性

2. 删除文档处理

3. 同步失败处理

4. 性能影响

📈 监控指标

关键指标

🔄 下一步：阶段四