在AI回答监测系统中，问题库扮演着“生产计划表”这一核心角色。它并非简单的问题堆砌——从工程层面来看，它至少需要承担三项关键职责：

• 数据生产入口：每条问题都对应一组采集任务，问题库的结构直接决定了后续任务编排的复杂度与效率。
• 场景分类载体：每条问题都会携带场景标签，例如推荐决策、对比分析、风险判断等。该标签将贯穿采集、诊断及整个分析链路。
• 采样基线锚点：同一组问题在不同周期内反复使用，构成趋势对比的数据基础。问题一旦上线，不宜随意更改，否则前后数据将失去可比性。

因此，问题库的设计必须在可配置性与稳定性之间取得平衡——既要方便新增和维护，又要确保跨周期的数据一致性。这两个方面同等重要，缺一不可。

一、问题库表结构设计

首先来看核心表结构。关键在于将“问题”本身作为独立实体进行管理，同时完整记录其生命周期：

CREATE TABLE question_bank (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    question_id VARCHAR(32) NOT NULL UNIQUE COMMENT '问题唯一标识, 如 Q20260701_001',
    scene_type VARCHAR(30) NOT NULL COMMENT '场景分类',
    query_text TEXT NOT NULL COMMENT '提问原文',
    target_keywords VARCHAR(255) COMMENT '监测关键词, 逗号分隔',
    intent_description VARCHAR(255) COMMENT '意图说明, 供运营参考',
    is_active TINYINT(1) DEFAULT 1 COMMENT '是否启用',
    version INT DEFAULT 1 COMMENT '版本号, 修改后递增',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_scene_active (scene_type, is_active),
    INDEX idx_question_id (question_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

我们来逐一拆解几个关键字段。question_id 采用的是业务主键而非自增 ID，这能确保在开发、测试、生产等不同环境中保持一致。version 字段至关重要，它记录了问题的修改次数。每当文本内容发生变化，版本号就会递增。这样一来，旧版本的数据可以通过 question_id + version 关联到历史样本。is_active 用于控制问题的上下线状态，下线后调度器会自动跳过该问题，但历史样本数据会完整保留，不会被误删。

此外，还有一张辅助表：

CREATE TABLE question_usage_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    question_id VARCHAR(32) NOT NULL,
    batch_id VARCHAR(64) NOT NULL COMMENT '批次ID',
    platform VARCHAR(30) NOT NULL,
    round INT NOT NULL COMMENT '采样轮次',
    sub_task_id VARCHAR(64) NOT NULL COMMENT '对应的采集子任务ID',
    status VARCHAR(20) DEFAULT 'pending' COMMENT 'pending/success/failed',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_batch (batch_id),
    INDEX idx_question (question_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

question_usage_log 扮演着问题库与采集任务之间的桥梁角色。每次调度都会写入一条记录，用于追踪“某问题在某批次中是否被成功采集”。这为后续的失败追溯和趋势分析奠定了坚实基础。

二、批次生成：把问题矩阵拆分为可管理的任务组

一次完整的监测任务，通常涉及 N 个品牌 × M 个场景 × P 个平台 × R 轮采样。如果直接进行组合，任务量将非常庞大。因此，必须按“批次”进行拆分调度。

2.1 批次划分策略

那么具体该如何划分呢？当数据量上升时，如果全部任务同时涌入，系统很容易面临压力。通常有以下三种思路：

在实际项目中，更推荐按品牌进行分批。原因很简单：后续分析通常以品牌为单位。同一批次的品牌数据，其时间窗口非常接近，对比的公平性自然更有保障。

2.2 批次元数据表

CREATE TABLE task_batch (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    batch_id VARCHAR(64) NOT NULL UNIQUE COMMENT '批次ID, 如 BATCH_20260707_001',
    brand_name VARCHAR(100) NOT NULL,
    scene_types JSON COMMENT '包含的场景列表',
    platforms JSON COMMENT '包含的平台列表',
    total_questions INT DEFAULT 0 COMMENT '本批次问题总数',
    total_sub_tasks INT DEFAULT 0 COMMENT '子任务总数(含多轮)',
    completed_sub_tasks INT DEFAULT 0 COMMENT '已完成子任务数',
    failed_sub_tasks INT DEFAULT 0 COMMENT '失败子任务数',
    batch_status VARCHAR(20) DEFAULT 'pending' COMMENT 'pending/running/completed/partial',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    completed_at TIMESTAMP NULL,
    INDEX idx_brand (brand_name),
    INDEX idx_status (batch_status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

这张表的核心作用，是完整记录整个批次的生命周期。从 pending 到 running，再到 completed 或 partial，状态变化一目了然。

2.3 批次生成逻辑

批次生成的逻辑其实并不复杂，核心就是三层循环：

def create_batch(brand_name: str, scene_types: list, platforms: list, rounds: int = 3) -> str:
    batch_id = f"BATCH_{datetime.now().strftime('%Y%m%d%H%M%S')}_{brand_name[:4].upper()}"
    
    # 从问题库拉取该品牌、指定场景的启用问题
    questions = db.query("SELECT question_id, scene_type, query_text FROM question_bank "
                         "WHERE is_active = 1 AND scene_type IN %s", scene_types)
    
    total_sub_tasks = 0
    for q in questions:
        for platform in platforms:
            for r in range(1, rounds + 1):
                sub_task = {
                    "sub_task_id": f"{batch_id}_Q{q['question_id']}_{platform}_R{r}",
                    "batch_id": batch_id,
                    "question_id": q["question_id"],
                    "query_text": q["query_text"],
                    "scene_type": q["scene_type"],
                    "platform": platform,
                    "round": r,
                    "brand_name": brand_name,
                }
                send_to_queue(sub_task)
                total_sub_tasks += 1
                
    db.insert("task_batch", {
        "batch_id": batch_id,
        "brand_name": brand_name,
        "total_questions": len(questions),
        "total_sub_tasks": total_sub_tasks,
        "batch_status": "running",
    })
    
    return batch_id

这个逻辑清晰明了：先拉取问题，再按平台和轮次展开，每个子任务都拥有唯一 ID，随后被投递到队列中。批次生成后，状态立即更新为 running。

三、任务调度：批次内并发与跨批次隔离

3.1 批次级并发控制

同一批次内的子任务理论上可以全部并发执行，但现实情况是——AI 平台的 API 普遍存在限流机制。因此，调度策略需要分两级进行控制：

具体实现时，可以定义一个调度器：

class BatchScheduler:
    def __init__(self):
        self.platform_limiters = {
            "deepseek": TokenBucket(rate=5, capacity=10),
            "kimi": TokenBucket(rate=3, capacity=6),
            "tongyi": TokenBucket(rate=5, capacity=10),
        }
    
    def dispatch(self, sub_task: dict):
        limiter = self.platform_limiters.get(sub_task["platform"])
        if limiter:
            limiter.consume()  # 阻塞直到获取令牌
        execute_sub_task(sub_task)

3.2 批次状态追踪

每个子任务完成后，都需要及时更新批次计数器：

def on_sub_task_complete(sub_task: dict, success: bool):
    batch_id = sub_task["batch_id"]
    if success:
        db.execute("UPDATE task_batch SET completed_sub_tasks = completed_sub_tasks + 1 WHERE batch_id = %s", batch_id)
    else:
        db.execute("UPDATE task_batch SET failed_sub_tasks = failed_sub_tasks + 1 WHERE batch_id = %s", batch_id)
    
    # 检查批次是否全部完成
    batch = db.query("SELECT * FROM task_batch WHERE batch_id = %s", batch_id)
    if batch["completed_sub_tasks"] + batch["failed_sub_tasks"] == batch["total_sub_tasks"]:
        final_status = "completed" if batch["failed_sub_tasks"] == 0 else "partial"
        db.execute("UPDATE task_batch SET batch_status = %s, completed_at = NOW() WHERE batch_id = %s", final_status, batch_id)

这里的关键在于：只要存在任务失败，最终状态就会标记为 partial，而不是 completed。这有助于运营人员快速了解整体执行情况。

3.3 失败重试策略

失败的原因多种多样，不能采用一刀切的处理方式。实践中，我们将失败分为两大类：

RETRY_STRATEGY = {
    "temp_error": {"max_retries": 3, "backoff": "exponential"},  # 网络超时、限流
    "perm_error": {"max_retries": 0},  # API Key 失效、参数错误
}

def handle_failure(sub_task: dict, error_type: str):
    strategy = RETRY_STRATEGY.get(error_type, {"max_retries": 0})
    retry_count = sub_task.get("retry_count", 0)
    if retry_count < strategy["max_retries"]:
        sub_task["retry_count"] = retry_count + 1
        delay = 60 * (2 ** retry_count) if strategy["backoff"] == "exponential" else 30
        requeue_with_delay(sub_task, delay)
    else:
        # 写入失败记录，标记批次为 partial
        db.insert("question_usage_log", {
            "question_id": sub_task["question_id"],
            "batch_id": sub_task["batch_id"],
            "status": "failed",
            "error": error_type,
        })
        on_sub_task_complete(sub_task, success=False)

对于临时错误，例如网络超时或限流，采用指数退避策略，最多重试3次。对于永久错误，比如 API Key 失效，则直接放弃，并将失败记录写入 question_usage_log。

四、结果入库：从子任务到样本库

每个子任务完成后，数据需要分两部分入库：原始回答和提取的指标。

4.1 原始回答入库

CREATE TABLE answer_sample (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    sub_task_id VARCHAR(64) NOT NULL UNIQUE,
    batch_id VARCHAR(64) NOT NULL,
    question_id VARCHAR(32) NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    platform VARCHAR(30) NOT NULL,
    scene_type VARCHAR(30) NOT NULL,
    round INT NOT NULL,
    raw_response LONGTEXT COMMENT 'AI返回的原始回答',
    response_format VARCHAR(20) COMMENT 'markdown/json/text',
    response_length INT DEFAULT 0 COMMENT '回答字符数',
    response_time_ms INT COMMENT 'API响应耗时',
    cos_url VARCHAR(512) COMMENT 'COS存储链接',
    is_valid TINYINT(1) DEFAULT 1 COMMENT '是否有效样本',
    invalid_reason VARCHAR(50) COMMENT '无效原因: timeout/refuse/error',
    collected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_batch (batch_id),
    INDEX idx_brand_scene (brand_name, scene_type),
    INDEX idx_question_round (question_id, round)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

4.2 指标提取入库

CREATE TABLE answer_metric (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    sub_task_id VARCHAR(64) NOT NULL UNIQUE,
    batch_id VARCHAR(64) NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    platform VARCHAR(30) NOT NULL,
    scene_type VARCHAR(30) NOT NULL,
    question_id VARCHAR(32) NOT NULL,
    round INT NOT NULL,
    is_mentioned TINYINT(1) DEFAULT 0,
    is_recommended TINYINT(1) DEFAULT 0,
    recommendation_rank INT DEFAULT 0,
    has_citation TINYINT(1) DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_brand_scene (brand_name, scene_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

4.3 入库事务保证

一条子任务对应两条数据库写入——原始样本和指标。这中间必须保证原子性，否则数据会出现不一致：

def sa ve_sub_task_result(sub_task: dict, raw_response: str, metrics: dict):
    db.begin()
    try:
        db.insert("answer_sample", {...})
        db.insert("answer_metric", {...})
        db.insert("question_usage_log", {..., "status": "success"})
        db.commit()
    except Exception as e:
        db.rollback()
        raise

五、样本查询：按问题维度追溯

样本库建成之后，最常见的查询场景就是按问题维度追溯数据：

-- 某问题在多个平台的回答一致性
SELECT platform, is_mentioned, is_recommended, recommendation_rank
FROM answer_metric
WHERE question_id = 'Q20260701_001'
AND brand_name = '品牌A'
ORDER BY platform, round；

-- 某批次整体完成情况
SELECT 
    brand_name,
    COUNT(*) AS total_samples,
    SUM(is_valid) AS valid_samples,
    SUM(is_mentioned) AS mentioned,
    ROUND(SUM(is_mentioned)*100.0/SUM(is_valid), 1) AS mention_rate
FROM answer_metric
WHERE batch_id = 'BATCH_20260707_001_品牌A'
GROUP BY brand_name；

六、工程实践要点

1. 问题版本化是趋势分析的前提

问题文本一旦修改，新旧采集结果就不能直接进行对比。此时，version 字段和 question_usage_log 表的关联就显得格外重要。每次采集都绑定到具体的问题版本，进行趋势分析时按 question_id + version 分组，就能避免跨版本数据混杂在一起，从而得到干净的结果。

2. 批次超时要有兜底

在生产环境中，总有个别子任务长时间不返回——例如 AI 平台排队、网络挂起等情况。批次不能无限等待下去。建议设置批次超时时间，比如30分钟。超时后强制将状态标记为 partial，释放等待资源，失败的子任务记录在案，后续再单独补采。

3. 问题库需要定期巡检

问题库中的问题可能会随着业务变化而“过时”，比如提到的产品已经下架。建议每月巡检一次 question_usage_log，统计每条问题最近的成功采集时间。如果超过30天没有成功采集，就自动将其标记为 is_active=0，并通知运营人员确认。

七、结语

总而言之，问题库就是 AI 回答监测系统的“生产计划表”。批次生成将大规模任务拆解为可控的小单元，两级限流兼顾了执行效率与平台友好性，分表存储让原始样本和提取指标各得其所，问题版本化则确保了跨周期数据的可比性。这四层设计，让问题库从“静态配置”真正升级为“可调度、可追踪、可复盘”的任务编排引擎。

这套方案已在多个消费品牌和企业的多平台 AI 可见度监测中实际运行，支撑了每天数百条问题的自动化调度与样本归集。如果开发者有进一步需求，还可以在此基础上扩展问题库的自动生成能力（例如基于大模型按场景批量生成提问文本），进一步降低运营维护成本。