AI 回答诊断任务如何做队列调度、异常标注和结果入库？

{
  "diagnosis_id": "diag-uuid-xxxx",
  "task_id": "collect-uuid-xxxx",
  "platform": "deepseek",
  "scene_type": "RECOMMEND",
  "target_brand": "品牌A",
  "raw_data_url": "cos://bucket/2026-07-01/deepseek/RECOMMEND/collect-uuid-xxxx.json",
  "created_at": "2026-07-01T10:00:00Z"
}

CREATE TABLE diagnosis_result (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    task_id VARCHAR(64) NOT NULL UNIQUE,
    -- 其余字段略
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

def diagnose(raw_response: dict) -> dict:
    text = raw_response["answer"]
    brand = raw_response["target_brand"]
    
    # Step 1: 有效性判断
    if is_timeout(text):
        return {"validity": "invalid_timeout"}
    if is_refuse(text):
        return {"validity": "invalid_refuse"}
    
    # Step 2: 提及与推荐提取
    mention_result = extract_mention(text, brand)
    recommend_result = extract_recommendation(text, brand)
    
    # Step 3: 解释段落抽取（如有提及）
    explanation = ""
    if mention_result["is_mentioned"]:
        explanation = extract_explanation(text, brand)
    
    # Step 4: 三级风险标签生成
    risk_tags = []
    
    if check_confusion(explanation, brand):
        risk_tags.append("confusion")
    
    if check_fact_error(explanation, brand):
        risk_tags.append("fact_error")
    
    if check_omission(explanation):
        risk_tags.append("omission")
    
    if not risk_tags:
        risk_tags.append("clean")
    
    return {
        "validity": "valid",
        "is_mentioned": mention_result["is_mentioned"],
        "is_recommended": recommend_result["is_recommended"],
        "recommendation_rank": recommend_result["rank"],
        "explanation_text": explanation,
        "risk_tags": risk_tags,
        "risk_level": "high" if ("confusion" in risk_tags or "fact_error" in risk_tags) else "low"
    }

-- 主诊断结果表
CREATE TABLE diagnosis_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,
    target_brand VARCHAR(100) NOT NULL,
    
    validity VARCHAR(20) NOT NULL COMMENT '有效/无效类型',
    is_mentioned TINYINT(1) DEFAULT 0,
    is_recommended TINYINT(1) DEFAULT 0,
    recommendation_rank INT DEFAULT 0,
    has_explanation TINYINT(1) DEFAULT 0 COMMENT '是否有解释段落',
    explanation_char_count INT DEFAULT 0,
    
    risk_level VARCHAR(10) DEFAULT NULL COMMENT 'high/low',
    raw_data_url VARCHAR(512),
    diagnosed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_brand_valid (target_brand, validity),
    INDEX idx_risk (risk_level)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

-- 异常标签明细表
CREATE TABLE diagnosis_risk_tag (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    task_id VARCHAR(64) NOT NULL,
    tag_type VARCHAR(30) NOT NULL COMMENT 'confusion/fact_error/omission/outdated/vague',
    tag_detail TEXT COMMENT '标签触发详情',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_task (task_id),
    FOREIGN KEY (task_id) REFERENCES diagnosis_result(task_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

SELECT 
    target_brand,
    COUNT(*) AS total_samples,
    SUM(CASE WHEN is_mentioned = 1 THEN 1 ELSE 0 END) AS mentioned_count,
    SUM(CASE WHEN is_recommended = 1 THEN 1 ELSE 0 END) AS recommended_count,
    SUM(CASE WHEN risk_level = 'high' THEN 1 ELSE 0 END) AS high_risk_count
FROM diagnosis_result
WHERE validity = 'valid'
  AND diagnosed_at >= '2026-07-01'
GROUP BY target_brand；

SELECT 
    d.target_brand,
    r.tag_type,
    COUNT(*) AS tag_count
FROM diagnosis_risk_tag r
JOIN diagnosis_result d ON r.task_id = d.task_id
WHERE d.diagnosed_at >= '2026-07-01'
GROUP BY d.target_brand, r.tag_type
ORDER BY d.target_brand, tag_count DESC；

SELECT 
    task_id,
    platform,
    scene_type,
    risk_level,
    raw_data_url
FROM diagnosis_result
WHERE target_brand = '品牌A'
  AND risk_level = 'high'
ORDER BY diagnosed_at DESC；