场景化 AI 回答采集任务，如何做队列调度和结果入库？

{
  "task_id": "uuid-xxxx",
  "scene_type": "RECOMMEND",
  "platform": "kimi",
  "query_text": "有哪些适合敏感肌的防晒霜推荐？",
  "target_brand": "某品牌",
  "round": 3,
  "max_retries": 3,
  "created_at": "2026-06-29T09:00:00Z"
}

CREATE TABLE question_library (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    scene_type VARCHAR(30) NOT NULL COMMENT '场景: RECOMMEND/COMPARE/RISK等',
    query_text TEXT NOT NULL,
    target_brand VARCHAR(100),
    status TINYINT DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_scene (scene_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

def handle_message(message):
    task = parse_message(message)
    adapter = PlatformAdapterFactory.get(task.platform)
    
    try:
        raw_response = adapter.call_api(task.query_text, timeout=60)
        cos_key = save_raw_to_cos(task, raw_response)
        structured = extract_metrics(task, raw_response)
        save_to_db(task, cos_key, structured)
        return "SUCCESS"
    except TimeoutError:
        handle_retry(task)
    except Exception as e:
        send_to_dlq(task, str(e))

def handle_retry(task):
    task['retry_count'] = task.get('retry_count', 0) + 1
    if task['retry_count'] <= task['max_retries']:
        delay = 60 * (2 ** (task['retry_count'] - 1))  # 指数退避
        requeue_with_delay(task, delay)
    else:
        send_to_dlq(task, "MAX_RETRIES_EXCEEDED")

// COS 文件命名: {date}/{platform}/{scene_type}/{task_id}.json
{
  "task_id": "uuid-xxxx",
  "platform": "kimi",
  "scene_type": "RECOMMEND",
  "query": "有哪些适合敏感肌的防晒霜推荐？",
  "raw_response": "针对敏感肌，以下是几个值得考虑的品牌：...",
  "response_at": "2026-06-29T09:00:35Z",
  "duration_ms": 34800
}

CREATE TABLE monitoring_result (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    task_id VARCHAR(64) NOT NULL UNIQUE,
    platform VARCHAR(30) NOT NULL,
    scene_type VARCHAR(30) NOT NULL COMMENT '场景分类',
    target_brand VARCHAR(100) NOT NULL,
    
    is_mentioned TINYINT(1) DEFAULT 0 COMMENT '是否被提及',
    is_recommended TINYINT(1) DEFAULT 0 COMMENT '是否被推荐',
    recommendation_rank INT DEFAULT 0 COMMENT '推荐位次,0表示未推荐',
    has_citation TINYINT(1) DEFAULT 0 COMMENT '是否引用来源',
    
    is_valid TINYINT(1) DEFAULT 1 COMMENT '样本有效性',
    raw_data_url VARCHAR(512) COMMENT 'COS文件链接',
    sampled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_brand_scene (target_brand, scene_type),
    INDEX idx_platform_date (platform, sampled_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

SELECT 
    scene_type,
    COUNT(*) AS total,
    SUM(is_mentioned) AS mentioned,
    SUM(is_recommended) AS recommended,
    ROUND(SUM(is_mentioned)*100.0/COUNT(*), 1) AS mention_rate,
    ROUND(SUM(is_recommended)*100.0/COUNT(*), 1) AS recommend_rate
FROM monitoring_result
WHERE target_brand = '某品牌防晒霜'
  AND is_valid = 1
  AND sampled_at >= '2026-06-01'
GROUP BY scene_type
ORDER BY mention_rate DESC；