特斯拉级系统设计面试题（六）：PB 级自动驾驶路测数据标注系统 — 任务调度、质量控制与人机协同

1
┌─────────────────────────────────────────────────────────────┐
2
│                  数据接入层 (Vehicle → Cloud)                 │
3
│   - 数据回传 (HTTPS/MQTT)  - 切片 (clip)  - 元数据抽取       │
4
└────────────────────────┬────────────────────────────────────┘
5
                         │
6
┌────────────────────────▼────────────────────────────────────┐
7
│                  存储层 (分层)                                │
8
│  - 热数据 (NVMe, 30 天)  - 温数据 (HDD, 1 年)  - 冷数据 (S3) │
9
└────────────────────────┬────────────────────────────────────┘
10
                         │
11
┌────────────────────────▼────────────────────────────────────┐
12
│                  调度层                                       │
13
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐    │
14
│  │ 任务生成  │  │ 分配引擎  │  │ 质量控制  │  │ 模型预标  │    │
15
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘    │
16
│       └──────────────┴──────────────┴──────────────┘         │
17
└────────────────────────┬────────────────────────────────────┘
18
                         │
19
┌────────────────────────▼────────────────────────────────────┐
20
│                  标注工作台 (Web / Desktop App)               │
21
│  - 标注员  - 质检员  - 审核员  - 管理员                       │
22
└──────────────────────────────────────────────────────────────┘

1
-- 1. Clip 元数据表
2
CREATE TABLE data_clip (
3
    clip_id        VARCHAR(64)  NOT NULL,
4
    vehicle_id     VARCHAR(32)  NOT NULL,
5
    capture_time   DATETIME(3)  NOT NULL,
6
    duration_sec   INT          NOT NULL,
7
    camera_count   TINYINT      NOT NULL,
8
    lidar_frames   INT          NOT NULL,
9
    scene_tags     JSON         NULL COMMENT '场景标签: 高速/城市/雨天',
10
    difficulty     TINYINT      NOT NULL COMMENT '1-5 难度等级',
11
    priority       TINYINT      NOT NULL DEFAULT 3 COMMENT '1-5 优先级',
12
    status         VARCHAR(16)  NOT NULL COMMENT 'PENDING/ASSIGNED/LABELED/QC_PASSED/QC_FAILED/ARCHIVED',
13
    storage_path   VARCHAR(256) NOT NULL,
14
    file_size_gb   DECIMAL(10, 2) NOT NULL,
15
    created_at     DATETIME(3)  NOT NULL,
16
    PRIMARY KEY (clip_id),
17
    INDEX idx_status_priority (status, priority, capture_time),
18
    INDEX idx_vehicle (vehicle_id, capture_time)
19
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
20
  PARTITION BY RANGE (TO_DAYS(capture_time)) (
21
    PARTITION p202606 VALUES LESS THAN (TO_DAYS('2026-07-01')),
22
    -- 按月分区
23
  );
24

25
-- 2. 标注任务表
26
CREATE TABLE annotation_task (
27
    task_id        VARCHAR(64)  NOT NULL,
28
    clip_id        VARCHAR(64)  NOT NULL,
29
    task_type      VARCHAR(32)  NOT NULL COMMENT '2D_BOX/3D_BOX/SEGMENT/BEHAVIOR/LANE',
30
    difficulty     TINYINT      NOT NULL,
31
    estimated_min  INT          NOT NULL COMMENT '预计耗时（分钟）',
32
    required_level VARCHAR(16)  NOT NULL COMMENT 'L1/L2/L3 标注员等级',
33
    reward         DECIMAL(8, 2) NOT NULL COMMENT '单 clip 报酬',
34
    status         VARCHAR(16)  NOT NULL COMMENT 'PENDING/ASSIGNED/SUBMITTED/QC/DONE',
35
    assigned_to    BIGINT       NULL,
36
    assigned_at    DATETIME(3)  NULL,
37
    submitted_at   DATETIME(3)  NULL,
38
    PRIMARY KEY (task_id),
39
    INDEX idx_status_difficulty (status, difficulty),
40
    INDEX idx_assigned (assigned_to, status)
41
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
42

43
-- 3. 标注结果表
44
CREATE TABLE annotation_result (
45
    result_id      BIGINT       NOT NULL AUTO_INCREMENT,
46
    task_id        VARCHAR(64)  NOT NULL,
47
    clip_id        VARCHAR(64)  NOT NULL,
48
    user_id        BIGINT       NOT NULL,
49
    annotation     JSON         NOT NULL COMMENT '标注内容（框、分割等）',
50
    submitted_at   DATETIME(3)  NOT NULL,
51
    qc_status      VARCHAR(16)  NOT NULL DEFAULT 'PENDING' COMMENT 'PENDING/PASSED/REJECTED/REVIEW',
52
    qc_score       DECIMAL(5, 2) NULL,
53
    qc_user_id     BIGINT       NULL,
54
    qc_notes       TEXT         NULL,
55
    PRIMARY KEY (result_id, submitted_at),
56
    INDEX idx_task (task_id),
57
    INDEX idx_clip (clip_id),
58
    INDEX idx_user_qc (user_id, qc_status)
59
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
60

61
-- 4. 标注员表
62
CREATE TABLE annotator (
63
    user_id        BIGINT       NOT NULL,
64
    level          VARCHAR(16)  NOT NULL COMMENT 'L1/L2/L3',
65
    accuracy       DECIMAL(5, 4) NOT NULL DEFAULT 0 COMMENT '历史准确率',
66
    speed          DECIMAL(8, 2) NOT NULL DEFAULT 0 COMMENT '单 clip 平均耗时',
67
    daily_quota    INT          NOT NULL DEFAULT 100 COMMENT '每日配额',
68
    status         VARCHAR(16)  NOT NULL COMMENT 'ACTIVE/PAUSED/BANNED',
69
    PRIMARY KEY (user_id)
70
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

1
@Service
2
class TaskDispatchService(
3
    private val clipRepo: DataClipRepository,
4
    private val taskRepo: AnnotationTaskRepository,
5
    private val annotatorRepo: AnnotatorRepository,
6
    private val skillMatcher: SkillMatcher,
7
    private val loadBalancer: LoadBalancer
8
) {
9
    /**
10
     * 任务分配（核心调度逻辑）
11
     */
12
    fun dispatchTask(taskId: String): DispatchResult {
13
        val task = taskRepo.findById(taskId) ?: return DispatchResult.fail("TASK_NOT_FOUND")
14
        if (task.status != "PENDING") return DispatchResult.fail("TASK_NOT_PENDING")
15

16
        // 1. 找到合适的标注员
17
        val candidates = findEligibleAnnotators(task)
18
        if (candidates.isEmpty()) {
19
            return DispatchResult.fail("NO_ANNOTATOR_AVAILABLE")
20
        }
21

22
        // 2. 多目标评分（公平 + 紧急 + 准确率）
23
        val selected = candidates.minBy { annotator ->
24
            calculateDispatchScore(annotator, task)
25
        }
26

27
        // 3. 分配任务（加锁防并发）
28
        return assignTask(task, selected)
29
    }
30

31
    /**
32
     * 找到符合要求的标注员
33
     */
34
    private fun findEligibleAnnotators(task: AnnotationTask): List<Annotator> {
35
        return annotatorRepo.findBy {
36
            and(
37
                Annotator::status eq "ACTIVE",
38
                Annotator::level gte task.requiredLevel
39
            )
40
        }
41
        .filter { it.dailyCompleted < it.dailyQuota }  // 未超额
42
        .filter { it.accuracy >= 0.90 }  // 准确率门槛
43
        .filter { skillMatcher.hasSkill(it.userId, task.taskType) }  // 技能匹配
44
    }
45

46
    /**
47
     * 计算调度得分（越低越优先）
48
     */
49
    private fun calculateDispatchScore(annotator: Annotator, task: AnnotationTask): Double {
50
        // 1. 负载因子（完成的越少越优先）
51
        val loadFactor = annotator.dailyCompleted.toDouble() / annotator.dailyQuota
52

53
        // 2. 历史准确率（越高越优先）
54
        val accuracyFactor = 1.0 - annotator.accuracy
55

56
        // 3. 速度匹配（任务难度 vs 标注员速度）
57
        val speedFactor = (task.estimatedMin - annotator.avgSpeedMin) / task.estimatedMin.toDouble()
58

59
        // 4. 紧急程度
60
        val urgencyFactor = 1.0 / task.priority
61

62
        return loadFactor * 0.4 + accuracyFactor * 0.3 + speedFactor * 0.2 + urgencyFactor * 0.1
63
    }
64
}

1
/**
2
 * 协同标注：多人标同一 clip，互相校验
3
 */
4
@Service
5
class CollaborativeAnnotationService(
6
    private val redisTemplate: RedisTemplate,
7
    private val annotationRepo: AnnotationResultRepository
8
) {
9
    /**
10
     * 多人标同一 clip 模式：M-of-N 共识
11
     */
12
    fun submitAnnotation(taskId: String, userId: Long, annotation: Annotation): SubmitResult {
13
        // 1. 获取该 clip 的所有提交
14
        val existingResults = annotationRepo.findByTaskId(taskId)
15
        val requiredSubmissions = 3  // 3 人标
16

17
        // 2. 保存当前提交
18
        val saved = annotationRepo.save(AnnotationResult(
19
            taskId = taskId,
20
            userId = userId,
21
            annotation = annotation,
22
            submittedAt = Instant.now()
23
        ))
24

25
        // 3. 是否到达共识数量
26
        val totalSubmissions = existingResults.size + 1
27
        if (totalSubmissions < requiredSubmissions) {
28
            return SubmitResult.pending(totalSubmissions, requiredSubmissions)
29
        }
30

31
        // 4. 计算共识（IoU 加权投票）
32
        val consensus = calculateConsensus(existingResults + saved)
33
        return SubmitResult.consensus(consensus)
34
    }
35

36
    /**
37
     * 共识算法：IoU 加权投票
38
     */
39
    private fun calculateConsensus(results: List<AnnotationResult>): ConsensusAnnotation {
40
        val allBoxes = results.flatMap { it.annotation.boxes }
41

42
        // 按类别分组
43
        val byCategory = allBoxes.groupBy { it.category }
44

45
        return ConsensusAnnotation(
46
            boxes = byCategory.map { (category, boxes) ->
47
                val consensusBox = findConsensusBox(boxes)
48
                consensusBox
49
            }
50
        )
51
    }
52

53
    private fun findConsensusBox(boxes: List<BoundingBox>): BoundingBox {
54
        // 找出现次数 >= 2 的框（3 人中至少 2 人同意）
55
        val groups = boxes.groupBy {
56
            // 用坐标量化到网格
57
            (it.x / 10).toInt() * 10000 + (it.y / 10).toInt() * 100 + (it.width / 10).toInt() * 10 + (it.height / 10).toInt()
58
        }
59

60
        val consensus = groups.filter { it.value.size >= 2 }
61
            .map { (key, list) ->
62
                // 取平均值
63
                val avgBox = BoundingBox(
64
                    x = list.map { it.x }.average(),
65
                    y = list.map { it.y }.average(),
66
                    width = list.map { it.width }.average(),
67
                    height = list.map { it.height }.average(),
68
                    category = list.first().category
69
                )
70
                avgBox
71
            }
72

73
        return consensus.first()
74
    }
75
}

1
/**
2
 * 质检工作流：分层质检
3
 */
4
@Service
5
class QualityControlService(
6
    private val annotationRepo: AnnotationResultRepository,
7
    private val qcRuleEngine: QcRuleEngine,
8
    private val autoQaModel: AutoQaModel
9
) {
10
    companion object {
11
        // 抽样率
12
        private const val NORMAL_SAMPLE_RATE = 0.10  // 普通任务 10% 抽样
13
        private const val KEY_SAMPLE_RATE = 1.00      // 关键任务 100% 全检
14
    }
15

16
    /**
17
     * 提交后质检
18
     */
19
    fun qualityCheck(annotationResult: AnnotationResult): QcResult {
20
        val task = taskRepo.findById(annotationResult.taskId)!!
21

22
        // 1. 自动质检（AI 模型）
23
        val autoQa = autoQaModel.evaluate(annotationResult)
24

25
        // 2. 规则质检
26
        val ruleViolations = qcRuleEngine.check(annotationResult, task)
27

28
        // 3. 综合判断
29
        if (autoQa.confidence > 0.95 && ruleViolations.isEmpty()) {
30
            // AI 高置信度通过 + 无规则违反 → 直接通过
31
            return QcResult.passed(autoQa)
32
        }
33

34
        if (autoQa.confidence < 0.5 || ruleViolations.size > 3) {
35
            // AI 低置信度 或 多规则违反 → 直接拒绝
36
            return QcResult.rejected(autoQa, ruleViolations)
37
        }
38

39
        // 4. 抽样规则
40
        val sampleRate = if (task.priority >= 4) KEY_SAMPLE_RATE else NORMAL_SAMPLE_RATE
41
        if (Random.nextDouble() < sampleRate) {
42
            // 抽样中 → 进入人工审核队列
43
            return QcResult.queued(autoQa, ruleViolations)
44
        }
45

46
        // 未抽中 → 通过
47
        return QcResult.passed(autoQa)
48
    }
49
}

1
/**
2
 * 模型预标注：用最新模型先标一遍，人工修正
3
 */
4
@Service
5
class ModelPreAnnotationService(
6
    private val modelInference: ModelInferenceService
7
) {
8
    /**
9
     * 预标注（提升效率 3-5 倍）
10
     */
11
    fun preAnnotate(clip: DataClip): Annotation {
12
        // 1. 加载数据
13
        val images = loadImages(clip)
14
        val pointClouds = loadPointClouds(clip)
15

16
        // 2. 模型推理
17
        val predictions = modelInference.predict(images, pointClouds)
18

19
        // 3. 后处理（NMS、置信度过滤）
20
        val filtered = postProcess(predictions, confThreshold = 0.7)
21

22
        // 4. 转换为标注格式
23
        return Annotation(
24
            boxes = filtered.boundingBoxes,
25
            segmentations = filtered.masks,
26
            confidence = filtered.scores
27
        )
28
    }
29
}

1
// 标注员动态评级
2
fun updateLevel(userId: Long) {
3
    val recentAccuracy = calculateRecentAccuracy(userId, last30days = true)
4
    val newLevel = when {
5
        recentAccuracy >= 0.98 -> "L3"
6
        recentAccuracy >= 0.95 -> "L2"
7
        else -> "L1"
8
    }
9
    annotatorRepo.updateLevel(userId, newLevel)
10
}

1
// 争议处理：5 人标注，3 人同意即通过；分歧太大交专家
2
fun resolveDispute(taskId: String): DisputeResult {
3
    val results = annotationRepo.findByTaskId(taskId)
4

5
    // 计算 IoU 分布
6
    val iouDistribution = results.map { calculateIoU(it, consensus) }
7

8
    return when {
9
        iouDistribution.average() > 0.9 -> DisputeResult.consensus(consensus)
10
        iouDistribution.average() > 0.7 -> DisputeResult.partial(consensus)  // 部分通过
11
        else -> DisputeResult.expertReview(taskId)  // 交专家审核
12
    }
13
}

1
// 标注结果变更 → Kafka → 训练 pipeline
2
@KafkaListener(topics = ["annotation.verified"])
3
fun onAnnotationVerified(event: AnnotationVerifiedEvent) {
4
    // 1. 转换格式（标注格式 → TFRecord）
5
    val trainingData = convertToTrainingFormat(event.annotation, event.clip)
6

7
    // 2. 上传到训练存储
8
    trainingDataStore.put("training/${event.clipId}", trainingData)
9

10
    // 3. 通知训练调度
11
    trainingScheduler.notifyNewData(event.clipId, event.taskType)
12
}