向量化系统完整设计文档

📋 文档信息

版本: v1.0.0
创建日期: 2025-12-12
最后更新: 2025-12-12
状态: Phase 1 已完成，Phase 2-5 待实施

一、系统概述

1.1 目标

构建一个统一、可扩展的向量化系统，支持：

✅ 多种文档格式的向量化
✅ 批量处理和增量更新
✅ 多数据源接入
✅ 质量控制和监控
✅ 与现有RAG系统无缝集成

1.2 核心能力

能力	说明	优先级	状态
单文档向量化	支持单个文档的向量化处理	P0	✅ Phase 1 完成
批量向量化	支持大批量文档的并行处理	P0	✅ Phase 1 完成
增量更新	支持文档变更的增量同步	P2	⏳ Phase 3 计划
多数据源	支持API、数据库、文件系统等	P3	⏳ Phase 4 计划
任务调度	异步任务队列和调度管理	P1	⏳ Phase 2 计划
质量控制	向量质量评估和优化	P1	⏳ Phase 2 计划
监控指标	系统性能和质量监控	P4	⏳ Phase 5 计划

二、系统架构

2.1 整体架构

┌─────────────────────────────────────────────────────────────────┐
│                          应用层                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ REST API     │  │ 定时任务      │  │ 事件监听      │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────────┐
│                          服务层                                  │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │          VectorizationService (核心服务)                  │  │
│  └──────────────────────────────────────────────────────────┘  │
│                              │                                   │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Scheduler    │  │ Pipeline     │  │ Monitor      │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────────┐
│                          执行层                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Loader       │  │ Splitter     │  │ Embedder     │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘
                              │
┌─────────────────────────────────────────────────────────────────┐
│                          存储层                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐          │
│  │ Milvus       │  │ MySQL        │  │ Redis        │          │
│  │ (向量数据)    │  │ (元数据)      │  │ (缓存)        │          │
│  └──────────────┘  └──────────────┘  └──────────────┘          │
└─────────────────────────────────────────────────────────────────┘

2.2 模块结构

pigx-ai-biz/src/main/java/com/pig4cloud/pigx/ai/rag/
├── vector/                        # 向量化模块
│   ├── controller/                # REST API
│   │   └── VectorizationController.java
│   ├── service/                   # 服务层
│   │   ├── VectorizationService.java
│   │   └── impl/
│   │       └── VectorizationServiceImpl.java
│   ├── domain/                    # 领域模型
│   │   ├── config/
│   │   │   └── VectorizationConfig.java
│   │   ├── entity/
│   │   │   └── VectorizationTaskEntity.java
│   │   ├── dto/
│   │   │   ├── VectorizationResult.java
│   │   │   └── VectorizationMetrics.java
│   │   └── type/
│   │       ├── TaskType.java
│   │       └── TaskStatus.java
│   ├── mapper/                    # 数据访问
│   │   └── VectorizationTaskMapper.java
│   ├── pipeline/                  # 处理流水线
│   │   ├── VectorizationPipeline.java
│   │   └── DocumentVectorizationPipeline.java
│   ├── scheduler/                 # 任务调度（Phase 2）
│   │   ├── VectorizationScheduler.java
│   │   └── BatchJobExecutor.java
│   ├── monitor/                   # 监控（Phase 5）
│   │   └── VectorizationMonitor.java
│   └── executor/                  # 执行器（已有）
│       └── ...

三、核心组件设计

3.1 VectorizationService（向量化服务）

职责: 统一的向量化入口，协调整个向量化流程

核心接口:

java

public interface VectorizationService {

    // ==================== Phase 1: 基础向量化 ====================

    /**
     * 单文档向量化
     * @param documentId 文档ID
     * @param config 向量化配置（可选，null则使用默认配置）
     * @return 向量化结果
     */
    VectorizationResult vectorizeDocument(Long documentId, VectorizationConfig config);

    /**
     * 批量文档向量化
     * @param documentIds 文档ID列表
     * @param config 向量化配置
     * @return 批量向量化结果
     */
    VectorizationTaskEntity batchVectorize(List<Long> documentIds, VectorizationConfig config);

    /**
     * 数据集向量化
     * @param datasetId 数据集ID
     * @return 向量化任务
     */
    VectorizationTaskEntity vectorizeDataset(Long datasetId);

    /**
     * 删除文档向量
     * @param documentId 文档ID
     * @return 是否成功
     */
    boolean deleteVectors(Long documentId);

    /**
     * 重建数据集索引
     * @param datasetId 数据集ID
     * @return 向量化任务
     */
    VectorizationTaskEntity rebuildIndex(Long datasetId);

    /**
     * 查询任务状态
     * @param taskId 任务ID
     * @return 任务实体
     */
    VectorizationTaskEntity getTaskStatus(String taskId);

    // ==================== Phase 2: 异步任务调度 ====================

    /**
     * 异步向量化（提交到任务队列）
     * @param documentIds 文档ID列表
     * @param config 配置
     * @param priority 优先级
     * @return 任务ID
     */
    String asyncVectorize(List<Long> documentIds, VectorizationConfig config, Integer priority);

    /**
     * 取消任务
     * @param taskId 任务ID
     * @return 是否成功
     */
    boolean cancelTask(String taskId);

    /**
     * 重试失败任务
     * @param taskId 任务ID
     * @return 新任务ID
     */
    String retryTask(String taskId);

    // ==================== Phase 3: 增量更新 ====================

    /**
     * 增量更新文档
     * @param documentId 文档ID
     * @param updatedContent 更新后的内容
     * @return 向量化结果
     */
    VectorizationResult incrementalUpdate(Long documentId, String updatedContent);

    /**
     * 增量同步数据集
     * @param datasetId 数据集ID
     * @param since 起始时间
     * @return 同步任务
     */
    VectorizationTaskEntity incrementalSync(Long datasetId, LocalDateTime since);

    // ==================== Phase 4: 多数据源 ====================

    /**
     * 从数据源向量化
     * @param dataSourceConfig 数据源配置
     * @return 向量化任务
     */
    VectorizationTaskEntity vectorizeDataSource(DataSourceConfig dataSourceConfig);

    /**
     * 文本直接向量化
     * @param text 文本内容
     * @param metadata 元数据
     * @param config 配置
     * @return 向量化结果
     */
    VectorizationResult vectorizeText(String text, Map<String, Object> metadata, VectorizationConfig config);

    // ==================== Phase 5: 监控指标 ====================

    /**
     * 获取数据集向量化指标
     * @param datasetId 数据集ID
     * @return 指标数据
     */
    VectorizationMetrics getMetrics(Long datasetId);

    /**
     * 获取系统整体指标
     * @return 系统指标
     */
    VectorizationMetrics getSystemMetrics();
}

3.2 VectorizationPipeline（处理流水线）

职责: 定义向量化处理的标准流程

java

public interface VectorizationPipeline {

    /**
     * 执行向量化流水线
     * @param input 输入数据
     * @param config 配置
     * @return 处理结果
     */
    VectorizationResult execute(PipelineInput input, VectorizationConfig config);

    /**
     * 流水线步骤:
     * 1. 加载文档（DocumentLoader）
     * 2. 文本分块（DocumentSplitter）
     * 3. 向量生成（EmbeddingGenerator）
     * 4. 质量检查（QualityChecker）
     * 5. 存储向量（VectorStore）
     * 6. 更新索引（IndexUpdater）
     */
}

实现类: DocumentVectorizationPipeline

java

@Service
public class DocumentVectorizationPipeline implements VectorizationPipeline {

    @Override
    public VectorizationResult execute(PipelineInput input, VectorizationConfig config) {
        // 1. 加载文档
        Document document = documentLoader.load(input.getDocumentId());

        // 2. 文本分块
        List<TextSegment> chunks = documentSplitter.split(
            document,
            config.getChunkSize(),
            config.getChunkOverlap()
        );

        // 3. 向量生成（批量）
        List<Embedding> embeddings = embeddingGenerator.generateBatch(
            chunks,
            config.getBatchSize()
        );

        // 4. 质量检查
        if (config.getEnableQualityCheck()) {
            chunks = qualityChecker.filter(chunks, embeddings);
        }

        // 5. 存储向量
        vectorStore.add(embeddings, chunks, input.getMetadata());

        // 6. 更新索引
        indexUpdater.update(input.getDatasetId());

        return VectorizationResult.success(chunks.size());
    }
}

3.3 VectorizationScheduler（任务调度器）- Phase 2

职责: 管理异步向量化任务的调度和执行

java

@Service
public class VectorizationScheduler {

    private final PriorityBlockingQueue<VectorizationTask> taskQueue;
    private final ThreadPoolExecutor executor;

    /**
     * 提交任务到队列
     */
    public String submitTask(VectorizationTask task) {
        taskQueue.offer(task);
        return task.getTaskId();
    }

    /**
     * 执行任务
     */
    @Scheduled(fixedDelay = 1000)
    public void processTasks() {
        VectorizationTask task = taskQueue.poll();
        if (task != null) {
            executor.execute(() -> executeTask(task));
        }
    }

    /**
     * 任务执行逻辑
     */
    private void executeTask(VectorizationTask task) {
        // 更新状态为 RUNNING
        // 执行向量化
        // 处理失败重试
        // 更新最终状态
    }
}

四、数据模型设计

4.1 VectorizationConfig（配置模型）

java

@Data
@Builder
public class VectorizationConfig {

    // ==================== 向量模型配置 ====================

    /**
     * Embedding模型名称
     * 示例: "text-embedding-ada-002", "text-embedding-3-large"
     */
    private String embeddingModel = "text-embedding-ada-002";

    /**
     * 向量维度
     * ada-002: 1536, text-embedding-3-large: 3072
     */
    private Integer embeddingDimension = 1536;

    // ==================== 分块配置 ====================

    /**
     * 分块大小（token数）
     */
    private Integer chunkSize = 512;

    /**
     * 重叠大小（token数）
     */
    private Integer chunkOverlap = 50;

    /**
     * 分块器类型
     * RECURSIVE: 递归分块
     * SENTENCE: 句子分块
     * PARAGRAPH: 段落分块
     */
    private String splitterType = "RECURSIVE";

    // ==================== 批处理配置 ====================

    /**
     * 批处理大小
     */
    private Integer batchSize = 10;

    /**
     * 最大并发数
     */
    private Integer maxConcurrency = 5;

    // ==================== 质量控制 ====================

    /**
     * 最小分块长度
     */
    private Double minChunkLength = 50.0;

    /**
     * 最大分块长度
     */
    private Double maxChunkLength = 2000.0;

    /**
     * 启用质量检查
     */
    private Boolean enableQualityCheck = true;

    /**
     * 质量分数阈值（0-1）
     */
    private Double qualityThreshold = 0.6;

    // ==================== 增量配置（Phase 3）====================

    /**
     * 启用增量更新
     */
    private Boolean enableIncremental = true;

    /**
     * 同步间隔（分钟）
     */
    private Integer syncIntervalMinutes = 60;

    // ==================== 存储配置 ====================

    /**
     * 索引名称
     */
    private String indexName;

    /**
     * 自定义元数据
     */
    private Map<String, Object> customMetadata;

    /**
     * 从数据集创建配置
     */
    public static VectorizationConfig fromDataset(KbDatasetEntity dataset) {
        return VectorizationConfig.builder()
            .embeddingModel(dataset.getEmbeddingModel())
            .chunkSize(dataset.getChunkMaxTokenSize())
            .chunkOverlap(dataset.getChunkOverlapTokenSize())
            .indexName("npu")  // 从数据集获取或使用默认值
            .build();
    }
}

4.2 VectorizationTaskEntity（任务实体）

java

@Data
@TableName("kb_vectorization_task")
public class VectorizationTaskEntity {

    @TableId(type = IdType.ASSIGN_ID)
    private Long id;

    /**
     * 任务ID（UUID）
     */
    private String taskId;

    /**
     * 任务名称
     */
    private String name;

    /**
     * 任务类型
     */
    private TaskType taskType;

    /**
     * 任务状态
     */
    private TaskStatus status;

    /**
     * 数据集ID
     */
    private Long datasetId;

    /**
     * 文档ID列表（JSON数组）
     */
    private String documentIds;

    /**
     * 配置（JSON）
     */
    private String config;

    /**
     * 总文档数
     */
    private Integer totalDocuments;

    /**
     * 已处理文档数
     */
    private Integer processedDocuments;

    /**
     * 失败文档数
     */
    private Integer failedDocuments;

    /**
     * 总分段数
     */
    private Integer totalChunks;

    /**
     * 已处理分段数
     */
    private Integer processedChunks;

    /**
     * 进度百分比（0-100）
     */
    @TableField(exist = false)
    private Double progress;

    /**
     * 优先级（Phase 2）
     */
    private Integer priority;

    /**
     * 重试次数（Phase 2）
     */
    private Integer retryCount;

    /**
     * 最大重试次数（Phase 2）
     */
    private Integer maxRetries;

    /**
     * 错误信息
     */
    private String errorMessage;

    /**
     * 耗时（毫秒）
     */
    private Long durationMs;

    /**
     * 开始时间
     */
    private LocalDateTime startedAt;

    /**
     * 完成时间
     */
    private LocalDateTime completedAt;

    /**
     * 创建人
     */
    @TableField(fill = FieldFill.INSERT)
    private String createBy;

    /**
     * 创建时间
     */
    @TableField(fill = FieldFill.INSERT)
    private LocalDateTime createTime;

    /**
     * 更新人
     */
    @TableField(fill = FieldFill.INSERT_UPDATE)
    private String updateBy;

    /**
     * 更新时间
     */
    @TableField(fill = FieldFill.INSERT_UPDATE)
    private LocalDateTime updateTime;

    /**
     * 计算进度
     */
    public Double getProgress() {
        if (totalDocuments == null || totalDocuments == 0) {
            return 0.0;
        }
        return (processedDocuments * 100.0) / totalDocuments;
    }

    /**
     * 是否已完成
     */
    public boolean isCompleted() {
        return status == TaskStatus.SUCCESS
            || status == TaskStatus.FAILED
            || status == TaskStatus.CANCELLED;
    }
}

4.3 任务类型枚举

java

public enum TaskType {
    /**
     * 全量向量化
     */
    FULL("全量向量化"),

    /**
     * 增量更新（Phase 3）
     */
    INCREMENTAL("增量更新"),

    /**
     * 重建索引
     */
    REBUILD("重建索引"),

    /**
     * 单文档处理
     */
    SINGLE_DOC("单文档处理");

    private final String description;
}

4.4 任务状态枚举

java

public enum TaskStatus {
    /**
     * 待处理
     */
    PENDING("待处理"),

    /**
     * 运行中
     */
    RUNNING("运行中"),

    /**
     * 成功
     */
    SUCCESS("成功"),

    /**
     * 失败
     */
    FAILED("失败"),

    /**
     * 已取消（Phase 2）
     */
    CANCELLED("已取消");

    private final String description;
}

五、RAG系统集成

5.1 集成点1: RagContentRetrieverService

java

@Service
public class RagContentRetrieverService {

    private final VectorizationService vectorizationService;
    private final VectorIndexService vectorIndexService;

    public Map<String, Object> retrieve(String query, Long datasetId) {
        // 确保数据集已向量化
        vectorIndexService.ensureIndexReady(datasetId);

        // 执行检索
        List<Content> contents = retriever.retrieve(Query.from(query));

        return Map.of(RagState.RETRIEVED_CONTENTS_KEY, contents);
    }
}

5.2 集成点2: DatasetService

java

@Service
public class DatasetService {

    private final VectorizationService vectorizationService;
    private final TaskScheduler taskScheduler;

    /**
     * 创建数据集时自动向量化
     */
    public void createDataset(Dataset dataset, List<Long> documentIds) {
        // 保存数据集
        datasetRepository.save(dataset);

        // 自动触发向量化
        VectorizationTaskEntity task = vectorizationService.batchVectorize(
            documentIds,
            VectorizationConfig.fromDataset(dataset)
        );

        // Phase 2: 异步执行
        // taskScheduler.schedule(task);
    }
}

5.3 集成点3: ContentRetriever配置

java

@Service
public class RagContentRetrieverFactory {

    private final VectorConfigService vectorConfigService;

    /**
     * 通过appId配置检索参数
     */
    public ContentRetriever createRetriever(String appId, List<Long> datasetIds, RagConfig ragConfig) {
        // 获取向量化配置
        VectorizationConfig vectorConfig = vectorConfigService.getByAppId(appId);

        // 创建检索器时使用向量化配置
        return EmbeddingStoreContentRetriever.builder()
            .embeddingStore(kbMilvusEmbeddingStore)
            .embeddingModel(embeddingModelFactory.build(vectorConfig.getEmbeddingModel()))
            .maxResults(ragConfig.getMaxResults())
            .minScore(ragConfig.getMinScore())
            .build();
    }
}

六、接口设计

6.1 REST API

java

@RestController
@RequestMapping("/ai/vectorization")
@Tag(name = "向量化管理")
public class VectorizationController {

    // ==================== Phase 1: 基础API ====================

    /**
     * 单文档向量化
     */
    @PostMapping("/document/{id}")
    @Operation(summary = "向量化单个文档")
    public R<VectorizationResult> vectorizeDocument(@PathVariable Long id) {
        return R.ok(vectorizationService.vectorizeDocument(id, null));
    }

    /**
     * 批量向量化
     */
    @PostMapping("/batch")
    @Operation(summary = "批量向量化文档")
    public R<VectorizationTaskEntity> batchVectorize(@RequestBody BatchVectorizeRequest request) {
        VectorizationTaskEntity task = vectorizationService.batchVectorize(
            request.getDocumentIds(),
            request.getConfig()
        );
        return R.ok(task);
    }

    /**
     * 数据集向量化
     */
    @PostMapping("/dataset/{datasetId}")
    @Operation(summary = "向量化整个数据集")
    public R<VectorizationTaskEntity> vectorizeDataset(@PathVariable Long datasetId) {
        return R.ok(vectorizationService.vectorizeDataset(datasetId));
    }

    /**
     * 删除文档向量
     */
    @DeleteMapping("/document/{id}")
    @Operation(summary = "删除文档向量")
    public R<Boolean> deleteVectors(@PathVariable Long id) {
        return R.ok(vectorizationService.deleteVectors(id));
    }

    /**
     * 重建索引
     */
    @PostMapping("/index/rebuild")
    @Operation(summary = "重建数据集索引")
    public R<VectorizationTaskEntity> rebuildIndex(@RequestParam Long datasetId) {
        return R.ok(vectorizationService.rebuildIndex(datasetId));
    }

    /**
     * 查询任务状态
     */
    @GetMapping("/task/{taskId}")
    @Operation(summary = "查询任务状态")
    public R<VectorizationTaskEntity> getTaskStatus(@PathVariable String taskId) {
        return R.ok(vectorizationService.getTaskStatus(taskId));
    }

    /**
     * 分页查询任务列表
     */
    @GetMapping("/tasks")
    @Operation(summary = "分页查询任务列表")
    public R<IPage<VectorizationTaskEntity>> getTasks(
        @RequestParam(required = false) Long datasetId,
        @RequestParam(required = false) TaskStatus status,
        @RequestParam(defaultValue = "1") Integer page,
        @RequestParam(defaultValue = "10") Integer size
    ) {
        Page<VectorizationTaskEntity> pageParam = new Page<>(page, size);
        return R.ok(vectorizationService.getTasks(datasetId, status, pageParam));
    }

    /**
     * 获取数据集指标
     */
    @GetMapping("/metrics/dataset/{datasetId}")
    @Operation(summary = "获取数据集向量化指标")
    public R<VectorizationMetrics> getMetrics(@PathVariable Long datasetId) {
        return R.ok(vectorizationService.getMetrics(datasetId));
    }

    // ==================== Phase 2: 异步任务API ====================

    /**
     * 异步向量化
     */
    @PostMapping("/async")
    @Operation(summary = "异步向量化（提交到任务队列）")
    public R<String> asyncVectorize(@RequestBody AsyncVectorizeRequest request) {
        String taskId = vectorizationService.asyncVectorize(
            request.getDocumentIds(),
            request.getConfig(),
            request.getPriority()
        );
        return R.ok(taskId, "任务已提交");
    }

    /**
     * 取消任务
     */
    @PostMapping("/task/{taskId}/cancel")
    @Operation(summary = "取消正在执行的任务")
    public R<Boolean> cancelTask(@PathVariable String taskId) {
        return R.ok(vectorizationService.cancelTask(taskId));
    }

    /**
     * 重试失败任务
     */
    @PostMapping("/task/{taskId}/retry")
    @Operation(summary = "重试失败的任务")
    public R<String> retryTask(@PathVariable String taskId) {
        String newTaskId = vectorizationService.retryTask(taskId);
        return R.ok(newTaskId, "重试任务已创建");
    }

    /**
     * SSE推送任务进度（Phase 2）
     */
    @GetMapping(value = "/task/{taskId}/progress", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @Operation(summary = "实时获取任务进度（SSE）")
    public SseEmitter streamTaskProgress(@PathVariable String taskId) {
        return progressStreamService.createEmitter(taskId);
    }

    // ==================== Phase 3: 增量更新API ====================

    /**
     * 增量同步
     */
    @PostMapping("/sync/incremental")
    @Operation(summary = "增量同步数据集")
    public R<VectorizationTaskEntity> incrementalSync(
        @RequestParam Long datasetId,
        @RequestParam(required = false) LocalDateTime since
    ) {
        return R.ok(vectorizationService.incrementalSync(datasetId, since));
    }

    // ==================== Phase 5: 监控API ====================

    /**
     * 系统指标
     */
    @GetMapping("/metrics/system")
    @Operation(summary = "获取系统整体指标")
    public R<VectorizationMetrics> getSystemMetrics() {
        return R.ok(vectorizationService.getSystemMetrics());
    }
}

七、实现路线图

Phase 1: 基础向量化（P0）✅ 已完成

目标: 实现核心向量化能力

功能清单:

✅ VectorizationService核心接口
✅ DocumentVectorizationPipeline实现
✅ 与现有EmbeddingStoreIngestor集成
✅ 基本REST API
✅ 单文档向量化
✅ 批量文档向量化
✅ 数据集向量化
✅ 向量删除管理
✅ 任务状态追踪
✅ 数据库表结构

交付物:

✅ 代码实现
✅ 数据库SQL脚本
✅ API文档
✅ 部署指南（VECTORIZATION_PHASE1_README.md）

Phase 2: 异步任务调度（P1）⏳ 计划中

目标: 实现任务队列和异步处理

功能清单:

⏳ VectorizationScheduler实现
⏳ 优先级队列管理
⏳ 异步任务执行器
⏳ 任务取消和重试
⏳ 失败自动重试机制
⏳ SSE实时进度推送
⏳ 任务并发控制

技术选型:

任务队列: PriorityBlockingQueue + Redis（持久化）
线程池: ThreadPoolExecutor
进度推送: SseEmitter
重试: Spring Retry

核心类:

java

VectorizationScheduler       // 任务调度器
BatchJobExecutor            // 批量任务执行器
TaskProgressPublisher       // 进度推送器
RetryPolicy                 // 重试策略

Phase 3: 增量更新（P2）⏳ 计划中

目标: 支持文档变更的增量同步

功能清单:

⏳ 文档变更检测
⏳ 增量向量化逻辑
⏳ 定时增量同步
⏳ 增量同步调度器
⏳ 变更事件监听

技术方案:

变更检测: 基于文档更新时间
增量策略: 只处理变更的文档
定时同步: 使用Spring @Scheduled

核心类:

java

IncrementalUpdateScheduler   // 增量更新调度器
DocumentChangeDetector       // 文档变更检测器
IncrementalSyncService       // 增量同步服务

Phase 4: 多数据源（P3）⏳ 计划中

目标: 支持多种数据源接入

功能清单:

⏳ 数据源适配器框架
⏳ API数据源适配器
⏳ 数据库数据源适配器
⏳ 文件系统适配器
⏳ Web爬虫适配器
⏳ 数据源配置管理

支持的数据源:

数据源	说明	优先级
REST API	从外部API获取数据	P3
数据库	直接从数据库表读取	P3
文件系统	本地文件/NAS	P3
MinIO/S3	对象存储	P3
Web爬虫	网页抓取	P4

核心接口:

java

public interface DataSourceAdapter {
    List<Document> fetchDocuments(DataSourceConfig config);
    boolean supportsIncremental();
    List<Document> fetchIncremental(DataSourceConfig config, LocalDateTime since);
}

Phase 5: 监控优化（P4）⏳ 计划中

目标: 完善监控和性能优化

功能清单:

⏳ VectorizationMetrics详细指标
⏳ 性能监控Dashboard
⏳ 质量评分系统
⏳ 自动性能优化
⏳ 告警机制
⏳ 审计日志

监控指标:

java

public class VectorizationMetrics {
    // 数量指标
    private Long totalDocuments;      // 总文档数
    private Long totalChunks;         // 总分段数
    private Long totalVectors;        // 总向量数

    // 质量指标
    private Double avgQualityScore;   // 平均质量分数
    private Integer lowQualityCount;  // 低质量文档数

    // 性能指标
    private Long avgProcessTimeMs;    // 平均处理时间
    private Long totalProcessTimeMs;  // 总处理时间
    private Double throughput;        // 吞吐量（文档/秒）

    // 任务指标
    private Long successCount;        // 成功任务数
    private Long failureCount;        // 失败任务数
    private Double successRate;       // 成功率

    // 资源指标
    private Long storageUsedBytes;    // 存储使用量
    private Integer activeTaskCount;  // 活跃任务数
}

优化方向:

批处理优化
缓存策略
并发控制
索引优化
资源调度

八、数据库设计

8.1 向量化任务表

sql

CREATE TABLE `kb_vectorization_task` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `task_id` varchar(64) NOT NULL COMMENT '任务ID（UUID）',
  `name` varchar(255) DEFAULT NULL COMMENT '任务名称',
  `task_type` varchar(32) NOT NULL COMMENT '任务类型',
  `status` varchar(32) NOT NULL COMMENT '任务状态',
  `dataset_id` bigint DEFAULT NULL COMMENT '数据集ID',
  `document_ids` text COMMENT '文档ID列表（JSON）',
  `config` text COMMENT '配置（JSON）',
  `total_documents` int DEFAULT 0 COMMENT '总文档数',
  `processed_documents` int DEFAULT 0 COMMENT '已处理文档数',
  `failed_documents` int DEFAULT 0 COMMENT '失败文档数',
  `total_chunks` int DEFAULT 0 COMMENT '总分段数',
  `processed_chunks` int DEFAULT 0 COMMENT '已处理分段数',
  `priority` int DEFAULT 0 COMMENT '优先级',
  `retry_count` int DEFAULT 0 COMMENT '重试次数',
  `max_retries` int DEFAULT 3 COMMENT '最大重试次数',
  `error_message` text COMMENT '错误信息',
  `duration_ms` bigint DEFAULT NULL COMMENT '耗时（毫秒）',
  `started_at` datetime DEFAULT NULL COMMENT '开始时间',
  `completed_at` datetime DEFAULT NULL COMMENT '完成时间',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_task_id` (`task_id`),
  KEY `idx_dataset_id` (`dataset_id`),
  KEY `idx_status` (`status`),
  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='向量化任务表';

九、配置示例

9.1 默认配置

yaml

vectorization:
  # 向量模型
  embedding:
    model: text-embedding-ada-002
    dimension: 1536

  # 分块配置
  chunk:
    size: 512
    overlap: 50
    splitter-type: RECURSIVE

  # 批处理
  batch:
    size: 10
    max-concurrency: 5

  # 质量控制
  quality:
    enable-check: true
    min-chunk-length: 50
    max-chunk-length: 2000
    threshold: 0.6

  # 任务调度（Phase 2）
  scheduler:
    core-pool-size: 5
    max-pool-size: 10
    queue-capacity: 100
    retry:
      max-attempts: 3
      backoff-delay: 1000

十、最佳实践

10.1 配置建议

小文档（<1000字）: chunkSize=256, overlap=30
中等文档（1000-5000字）: chunkSize=512, overlap=50
大文档（>5000字）: chunkSize=1024, overlap=100

10.2 性能优化

合理设置 batchSize，避免API限流
使用 maxConcurrency 控制并发数
启用质量检查过滤低质量分段
大批量处理使用异步模式（Phase 2）

10.3 故障处理

失败任务自动重试（Phase 2）
记录详细错误信息便于排查
定期监控任务成功率
建立告警机制（Phase 5）

十一、参考文档

VECTORIZATION_PHASE1_README.md - Phase 1 部署指南
VECTORIZATION_PHASE2_GUIDE.md - Phase 2 实现指南（待创建）
VECTORIZATION_PHASE3_GUIDE.md - Phase 3 实现指南（待创建）
VECTORIZATION_PHASE4_GUIDE.md - Phase 4 实现指南（待创建）
VECTORIZATION_PHASE5_GUIDE.md - Phase 5 实现指南（待创建）

文档版本: v1.0.0 最后更新: 2025-12-12 维护者: 开发团队

向量化系统完整设计文档 ​

📋 文档信息 ​

一、系统概述 ​

1.1 目标 ​

1.2 核心能力 ​

二、系统架构 ​

2.1 整体架构 ​

2.2 模块结构 ​

三、核心组件设计 ​

3.1 VectorizationService（向量化服务） ​

3.2 VectorizationPipeline（处理流水线） ​

3.3 VectorizationScheduler（任务调度器）- Phase 2 ​

四、数据模型设计 ​

4.1 VectorizationConfig（配置模型） ​

4.2 VectorizationTaskEntity（任务实体） ​

4.3 任务类型枚举 ​

4.4 任务状态枚举 ​

五、RAG系统集成 ​

5.1 集成点1: RagContentRetrieverService ​

5.2 集成点2: DatasetService ​

5.3 集成点3: ContentRetriever配置 ​

六、接口设计 ​

6.1 REST API ​

七、实现路线图 ​

Phase 1: 基础向量化（P0）✅ 已完成 ​

Phase 2: 异步任务调度（P1）⏳ 计划中 ​

Phase 3: 增量更新（P2）⏳ 计划中 ​

Phase 4: 多数据源（P3）⏳ 计划中 ​

Phase 5: 监控优化（P4）⏳ 计划中 ​

八、数据库设计 ​

8.1 向量化任务表 ​

九、配置示例 ​

9.1 默认配置 ​

十、最佳实践 ​

10.1 配置建议 ​

10.2 性能优化 ​

10.3 故障处理 ​

十一、参考文档 ​

向量化系统完整设计文档

📋 文档信息

一、系统概述

1.1 目标

1.2 核心能力

二、系统架构

2.1 整体架构

2.2 模块结构

三、核心组件设计

3.1 VectorizationService（向量化服务）

3.2 VectorizationPipeline（处理流水线）

3.3 VectorizationScheduler（任务调度器）- Phase 2

四、数据模型设计

4.1 VectorizationConfig（配置模型）

4.2 VectorizationTaskEntity（任务实体）

4.3 任务类型枚举

4.4 任务状态枚举

五、RAG系统集成

5.1 集成点1: RagContentRetrieverService

5.2 集成点2: DatasetService

5.3 集成点3: ContentRetriever配置

六、接口设计

6.1 REST API

七、实现路线图

Phase 1: 基础向量化（P0）✅ 已完成

Phase 2: 异步任务调度（P1）⏳ 计划中

Phase 3: 增量更新（P2）⏳ 计划中

Phase 4: 多数据源（P3）⏳ 计划中

Phase 5: 监控优化（P4）⏳ 计划中

八、数据库设计

8.1 向量化任务表

九、配置示例

9.1 默认配置

十、最佳实践

10.1 配置建议

10.2 性能优化

10.3 故障处理

十一、参考文档