AI 回答监测指标计算任务如何做调度、聚合和结果快照？

原创

AI增长技术研究院

发布于 2026-07-09 11:45:48

1490

当结构化样本库中沉淀了数万条经过清洗的 AI 回答后，系统面临的最后一个工程问题是如何将这些零散的样本数据转化为品牌方真正关心的指标——提及率、推荐率、场景覆盖度，以及最重要的：这次和上次相比，是变好了还是变差了。本文将围绕指标计算任务的调度触发、批次聚合、结果快照和查询接口四个环节，拆解从结构化样本到品牌指标结果的完整数据链路。

一、指标计算在整体架构中的位置

在 AI 回答监测系统中，指标计算处于数据链路的末端，承上启下：

上游：结构化样本库，包含每条回答的 is_mentioned、is_recommended、recommendation_rank 等清洗后的字段。
下游：品牌看板、趋势图、竞品对比表、定期报告。

指标计算的核心任务是把“单条样本的二值判断”聚合为“品牌在特定维度下的统计指标”，并形成可跨周期对比的快照。

二、整体数据链路

链路分五个环节：计算触发、样本读取、分组聚合、快照生成、对比输出。

二、计算触发：事件驱动 + 定时兜底

指标计算有两种触发模式，各自适用于不同场景：

触发模式	适用场景	实现方式
事件驱动	批次采集完成，立即出结果	采集批次状态变为 completed 时发送触发消息
定时兜底	周期性报告、补算	定时触发器（如每日凌晨 2:00）扫描未计算批次

2.1 事件驱动触发

def on_batch_completed(batch_id: str):
    batch = db.query("SELECT * FROM task_batch WHERE batch_id = %s", batch_id)
    if batch["batch_status"] == "completed" and batch["failed_sub_tasks"] == 0:
        trigger_metric_calculation(batch_id)
    elif batch["batch_status"] == "partial":
        # 部分失败的批次也触发计算，但在快照中标记数据完整度
        trigger_metric_calculation(batch_id, is_partial=True)

2.2 定时兜底扫描

-- 定时任务：查找已完成但未生成快照的批次
SELECT batch_id 
FROM task_batch 
WHERE batch_status IN ('completed', 'partial')
  AND batch_id NOT IN (SELECT DISTINCT batch_id FROM metric_snapshot)
  AND completed_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
ORDER BY completed_at ASC；

2.3 计算任务消息体

{
  "calc_task_id": "calc-uuid-xxxx",
  "batch_id": "BATCH_20260709_001",
  "brand_name": "品牌A",
  "is_partial": false,
  "sample_count": 480,
  "trigger_type": "event",
  "created_at": "2026-07-09T10:30:00Z"
}

三、聚合计算：从样本到指标

3.1 聚合维度定义

一次完整的指标计算需要覆盖以下维度组合：

聚合维度	说明	输出指标示例
品牌 × 平台	单平台表现	品牌A在DeepSeek上的提及率
品牌 × 场景	单场景表现	品牌A在推荐场景下的推荐率
品牌 × 平台 × 场景	交叉分析	品牌A在DeepSeek推荐场景下的首位推荐率
品牌 × 全部平台	综合表现	品牌A的多平台综合提及率

3.2 核心指标计算 SQL

-- 品牌 × 平台 × 场景 维度的指标聚合
INSERT INTO metric_snapshot_detail (
    snapshot_id, brand_name, platform, scene_type,
    total_samples, valid_samples, invalid_samples,
    mentioned_count, recommended_count, cited_count,
    mention_rate, recommend_rate, citation_rate,
    avg_recommend_rank, top1_recommend_count, top1_recommend_rate,
    stability_score
)
SELECT 
    %s AS snapshot_id,
    brand_name,
    platform,
    scene_type,
    COUNT(*) AS total_samples,
    SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END) AS valid_samples,
    SUM(CASE WHEN is_valid = 0 THEN 1 ELSE 0 END) AS invalid_samples,
    SUM(CASE WHEN is_valid = 1 AND is_mentioned = 1 THEN 1 ELSE 0 END) AS mentioned_count,
    SUM(CASE WHEN is_valid = 1 AND is_recommended = 1 THEN 1 ELSE 0 END) AS recommended_count,
    SUM(CASE WHEN is_valid = 1 AND has_citation = 1 THEN 1 ELSE 0 END) AS cited_count,
    ROUND(SUM(CASE WHEN is_valid = 1 AND is_mentioned = 1 THEN 1 ELSE 0 END) * 100.0 / 
          NULLIF(SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END), 0), 1) AS mention_rate,
    ROUND(SUM(CASE WHEN is_valid = 1 AND is_recommended = 1 THEN 1 ELSE 0 END) * 100.0 / 
          NULLIF(SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END), 0), 1) AS recommend_rate,
    ROUND(SUM(CASE WHEN is_valid = 1 AND has_citation = 1 THEN 1 ELSE 0 END) * 100.0 / 
          NULLIF(SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END), 0), 1) AS citation_rate,
    ROUND(AVG(CASE WHEN is_valid = 1 AND is_recommended = 1 THEN recommendation_rank END), 1) AS avg_recommend_rank,
    SUM(CASE WHEN is_valid = 1 AND is_recommended = 1 AND recommendation_rank = 1 THEN 1 ELSE 0 END) AS top1_recommend_count,
    ROUND(SUM(CASE WHEN is_valid = 1 AND is_recommended = 1 AND recommendation_rank = 1 THEN 1 ELSE 0 END) * 100.0 /
          NULLIF(SUM(CASE WHEN is_valid = 1 AND is_recommended = 1 THEN 1 ELSE 0 END), 0), 1) AS top1_recommend_rate,
    -- 稳定性：多轮采样中提及结果的标准差（越小越稳定）
    ROUND(STDDEV(CASE WHEN is_valid = 1 THEN is_mentioned ELSE NULL END), 3) AS stability_score
FROM clean_result
WHERE batch_id = %s
  AND clean_state = 'completed'
GROUP BY brand_name, platform, scene_type；

3.3 多平台综合指标

-- 品牌 × 场景（跨平台综合）
SELECT 
    %s AS snapshot_id,
    brand_name,
    'ALL' AS platform,
    scene_type,
    COUNT(*) AS total_samples,
    SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END) AS valid_samples,
    SUM(CASE WHEN is_valid = 1 AND is_mentioned = 1 THEN 1 ELSE 0 END) AS mentioned_count,
    ROUND(SUM(CASE WHEN is_valid = 1 AND is_mentioned = 1 THEN 1 ELSE 0 END) * 100.0 / 
          NULLIF(SUM(CASE WHEN is_valid = 1 THEN 1 ELSE 0 END), 0), 1) AS mention_rate,
    -- 平台覆盖度：品牌在多少个平台上被提及
    COUNT(DISTINCT CASE WHEN is_valid = 1 AND is_mentioned = 1 THEN platform END) AS platform_coverage
FROM clean_result
WHERE batch_id = %s
  AND clean_state = 'completed'
GROUP BY brand_name, scene_type；

四、结果快照：让每次计算结果可追溯

4.1 为什么需要快照

如果指标计算是“实时查询样本库并即时聚合”，会存在三个问题：

样本库可能被后续采集追加，同一批次在不同时间点查询结果不同。
趋势图需要对比“上周的提及率”和“本周的提及率”，实时计算无法稳定复现历史值。
排障时需要回溯“这次计算时用了哪些样本”。

因此，每次计算的结果需要固化为快照，独立存储。

4.2 快照表设计

-- 指标快照主表
CREATE TABLE metric_snapshot (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    snapshot_id VARCHAR(64) NOT NULL UNIQUE COMMENT '快照ID, 如 SNAP_20260709_001',
    batch_id VARCHAR(64) NOT NULL COMMENT '关联的采集批次',
    brand_name VARCHAR(100) NOT NULL,
    calc_trigger VARCHAR(20) COMMENT '触发方式: event/schedule/manual',
    is_partial TINYINT(1) DEFAULT 0 COMMENT '是否为部分数据',
    total_samples INT DEFAULT 0,
    valid_samples INT DEFAULT 0,
    sample_completeness DECIMAL(4,1) COMMENT '样本完整度(%)',
    calculated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_brand_snapshot (brand_name, calculated_at),
    INDEX idx_batch (batch_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

-- 快照明细表：存储各维度的聚合结果
CREATE TABLE metric_snapshot_detail (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    snapshot_id VARCHAR(64) NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    platform VARCHAR(30) NOT NULL,
    scene_type VARCHAR(30) NOT NULL,
    
    total_samples INT DEFAULT 0,
    valid_samples INT DEFAULT 0,
    invalid_samples INT DEFAULT 0,
    mentioned_count INT DEFAULT 0,
    recommended_count INT DEFAULT 0,
    cited_count INT DEFAULT 0,
    
    mention_rate DECIMAL(5,1),
    recommend_rate DECIMAL(5,1),
    citation_rate DECIMAL(5,1),
    avg_recommend_rank DECIMAL(3,1),
    top1_recommend_rate DECIMAL(5,1),
    stability_score DECIMAL(5,3),
    
    INDEX idx_snapshot (snapshot_id),
    INDEX idx_brand_scene (brand_name, scene_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

主表记录每次计算的元信息（批次、完整度、时间），明细表存储各维度的聚合结果。查询时通过 snapshot_id 关联。

4.3 快照生成事务

def create_snapshot(batch_id: str, trigger_type: str):
    snapshot_id = f"SNAP_{datetime.now().strftime('%Y%m%d%H%M%S')}_{batch_id[:4]}"
    
    db.begin()
    try:
        # 写入主表
        db.insert("metric_snapshot", {
            "snapshot_id": snapshot_id,
            "batch_id": batch_id,
            "brand_name": get_brand_from_batch(batch_id),
            "calc_trigger": trigger_type,
            "is_partial": is_batch_partial(batch_id),
        })
        
        # 写入维度明细
        db.execute(AGGREGATE_BY_PLATFORM_SCENE_SQL, (snapshot_id, batch_id))
        db.execute(AGGREGATE_ALL_PLATFORM_SQL, (snapshot_id, batch_id))
        
        # 更新主表的汇总计数
        db.execute("""
            UPDATE metric_snapshot 
            SET total_samples = (SELECT SUM(total_samples) FROM metric_snapshot_detail WHERE snapshot_id = %s),
                valid_samples = (SELECT SUM(valid_samples) FROM metric_snapshot_detail WHERE snapshot_id = %s)
            WHERE snapshot_id = %s
        """, (snapshot_id, snapshot_id, snapshot_id))
        
        db.commit()
    except Exception:
        db.rollback()
        raise
    
    return snapshot_id

五、快照对比：让趋势可见

快照的价值不仅在于“记录当前值”，更在于“对比变化”。每次新快照生成后，自动与同品牌上一个完整快照做对比：

def compare_with_previous(snapshot_id: str):
    current = db.query(
        "SELECT * FROM metric_snapshot WHERE snapshot_id = %s", snapshot_id
    )
    previous = db.query(
        "SELECT * FROM metric_snapshot WHERE brand_name = %s AND is_partial = 0 AND snapshot_id < %s ORDER BY calculated_at DESC LIMIT 1",
        current["brand_name"], snapshot_id
    )
    
    if not previous:
        return None
    
    diffs = []
    current_details = db.query(
        "SELECT scene_type, platform, mention_rate, recommend_rate FROM metric_snapshot_detail WHERE snapshot_id = %s",
        snapshot_id
    )
    previous_details = db.query(
        "SELECT scene_type, platform, mention_rate, recommend_rate FROM metric_snapshot_detail WHERE snapshot_id = %s",
        previous["snapshot_id"]
    )
    
    # 按场景和平台匹配对比
    current_map = {(d["scene_type"], d["platform"]): d for d in current_details}
    previous_map = {(d["scene_type"], d["platform"]): d for d in previous_details}
    
    for key, curr in current_map.items():
        prev = previous_map.get(key)
        if prev:
            mention_change = curr["mention_rate"] - prev["mention_rate"]
            recommend_change = curr["recommend_rate"] - prev["recommend_rate"]
            if abs(mention_change) >= 5 or abs(recommend_change) >= 5:
                diffs.append({
                    "scene_type": key[0],
                    "platform": key[1],
                    "mention_change": round(mention_change, 1),
                    "recommend_change": round(recommend_change, 1),
                })
    
    return {
        "current_snapshot": snapshot_id,
        "previous_snapshot": previous["snapshot_id"],
        "significant_changes": diffs,
    }

对比结果写入变更通知表，超出阈值的变化触发告警。

六、查询接口设计

6.1 品牌最新指标概览

SELECT 
    m.snapshot_id,
    m.brand_name,
    m.calculated_at,
    m.valid_samples,
    d.platform,
    d.scene_type,
    d.mention_rate,
    d.recommend_rate,
    d.top1_recommend_rate
FROM metric_snapshot m
JOIN metric_snapshot_detail d ON m.snapshot_id = d.snapshot_id
WHERE m.brand_name = '品牌A'
  AND m.snapshot_id = (
      SELECT snapshot_id FROM metric_snapshot 
      WHERE brand_name = '品牌A' 
      ORDER BY calculated_at DESC LIMIT 1
  )
ORDER BY d.recommend_rate DESC；

6.2 品牌趋势数据

SELECT 
    m.snapshot_id,
    m.calculated_at,
    d.scene_type,
    d.mention_rate,
    d.recommend_rate
FROM metric_snapshot m
JOIN metric_snapshot_detail d ON m.snapshot_id = d.snapshot_id
WHERE m.brand_name = '品牌A'
  AND d.platform = 'ALL'
  AND d.scene_type = 'RECOMMEND'
  AND m.calculated_at >= '2026-06-01'
ORDER BY m.calculated_at ASC；

七、工程实践要点

1. 快照是“事实锚点”，不是“实时视图”

快照一旦生成就不应修改。如果发现数据问题（如某条样本被错误标记），应在样本库中修正后触发重新计算，生成新快照，而非修改旧快照。这保证了任何历史报告都有对应的数据锚点可回溯。

2. 部分批次也要出快照，但要标注

生产环境中 100% 采集成功是理想状态。is_partial 和 sample_completeness 字段让部分成功的批次也能产出可用数据，但调用方可以看到“本次数据完整度 92%，缺失部分来自 Kimi 平台第 3 轮采样”，而非等到全部完成才出结果。

3. 聚合计算适合用批量写入而非逐行插入

一次快照可能包含几十条维度明细记录。使用 INSERT INTO ... SELECT ... 批量写入比逐行 INSERT 快一个数量级，且数据库内聚合比应用层循环聚合效率更高。

八、结语

指标计算是 AI 回答监测系统的最后一公里。调度触发保证“样本到位即计算”，分组聚合将零散的二值判断转化为可对比的统计指标，快照固化为每次计算留下数据锚点，自动对比让趋势变化从“感觉”变为“数字”。

这套方案已在多个消费品牌和企业服务的多平台 AI 可见度监测中实际运行，支持每日快照生成和跨周期趋势对比。开发者可在此基础上扩展指标维度（如语义倾向评分、解释准确度量化），或引入环比/同比计算逻辑。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

人工智能