AI 回答样本清洗任务如何做队列调度、状态流转和结果入库？

原创

AI增长技术研究院

发布于 2026-07-08 11:16:12

1230

采集回来的 AI 回答是原始矿石，需要经过清洗才能变成可分析的结构化数据。但清洗链路本身也有工程复杂度：无效回答如何自动识别？品牌别名如何合并？清洗失败的任务如何重试？整条链路的处理状态如何追踪？本文围绕清洗队列调度、状态机流转、实体识别和结构化入库四个环节，拆解从原始回答到高质量样本的完整数据链路。

一、为什么需要独立的清洗链路

在 AI 回答监测系统中，采集和清洗如果耦合在同一个函数里，会出现三个问题：

职责不清：采集函数的超时时间是给 AI 平台留的（60-120 秒），清洗逻辑混在里面会挤占采集的容错空间。
无法独立重试：清洗失败（如实体识别出错）不应触发重新采集——那会浪费 API 配额并引入新的采样偏差。
状态不可追踪：混在一起时，只能知道“最终成功或失败”，无法区分是采集阶段挂了还是清洗阶段挂了。

因此，清洗链路应该作为独立环节，通过消息队列与采集链路解耦，拥有自己的状态机和重试策略。

二、整体数据链路

整个清洗链路包含五个处理节点：有效性判断 → 实体识别 → 指标提取 → 状态更新 → 结果入库。每个节点独立可重入，状态通过状态机统一管理。

三、清洗任务生成与队列调度

3.1 任务消息体设计

采集完成后，采集函数生成一条清洗任务消息：

{
  "clean_task_id": "clean-uuid-xxxx",
  "source_task_id": "collect-uuid-xxxx",
  "batch_id": "BATCH_20260708_001",
  "brand_name": "品牌A",
  "brand_aliases": ["品牌A", "BrandA", "A牌"],
  "platform": "deepseek",
  "scene_type": "RECOMMEND",
  "query_text": "有哪些适合企业的云数据库方案？",
  "round": 2,
  "raw_data_url": "cos://bucket/2026-07-08/deepseek/RECOMMEND/collect-uuid-xxxx.json",
  "max_clean_retries": 2,
  "created_at": "2026-07-08T10:00:00Z"
}

关键设计点：brand_aliases 由采集阶段的品牌配置带入清洗任务，保证别名列表的一致性和可追溯性。

3.2 队列配置

清洗任务属于计算密集型（正则匹配、文本解析），比采集任务（I/O 密集型）更适合高并发消费：

配置项	采集队列	清洗队列
消息超时	120s	30s
单实例并发	5-10	10-20
批量消费	可选	建议开启（每次拉 10 条）
重试策略	指数退避	立即重试 1 次后进 DLQ

批量消费可以减少网络往返开销。单次拉取 10 条消息，云函数内串行处理，单条清洗耗时约 200-500ms，整体在超时阈值内完成。

3.3 幂等性保障

清洗任务可能因消息队列的“至少一次投递”语义被重复消费。通过数据库唯一约束保障幂等：

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

清洗函数入口处检查 clean_task_id 是否已存在，存在则直接返回成功，跳过重复处理。

四、状态机设计：让每一步可追踪

清洗不是“一把过”的单一操作，而是多个步骤的顺序执行。引入有限状态机让每一步的状态可查询、可重入。

4.1 状态定义

状态	说明	可流转到
pending	消息已投递，等待消费	validating
validating	正在判断样本有效性	valid invalid failed
valid	样本有效，待实体识别	extracting
invalid	样本无效（超时/拒绝/不相关）	completed
extracting	正在进行实体识别和指标提取	extracted failed
extracted	指标提取完成，待入库	completed
completed	清洗完成	终态
failed	清洗失败（重试耗尽）	pending（手动重推）

4.2 状态流转表

CREATE TABLE clean_state (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    clean_task_id VARCHAR(64) NOT NULL UNIQUE,
    current_state VARCHAR(20) NOT NULL DEFAULT 'pending',
    state_history JSON COMMENT '状态流转日志',
    retry_count INT DEFAULT 0,
    last_error TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_state (current_state)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

每次状态变更时追加 state_history：

[
  {"from": "pending", "to": "validating", "at": "2026-07-08T10:00:01Z"},
  {"from": "validating", "to": "valid", "at": "2026-07-08T10:00:02Z"},
  {"from": "valid", "to": "extracting", "at": "2026-07-08T10:00:02Z"},
  {"from": "extracting", "to": "extracted", "at": "2026-07-08T10:00:03Z"},
  {"from": "extracted", "to": "completed", "at": "2026-07-08T10:00:04Z"}
]

state_history 是排查问题的核心数据源——当某条任务卡在 extracting 状态超过 5 分钟时，运维可以通过状态历史快速定位是哪个环节出了问题。

4.3 状态流转伪代码

def clean_sample(task: dict):
    state = get_or_create_state(task["clean_task_id"])
    
    # Step 1: 有效性判断
    transition(state, "validating")
    raw_text = fetch_raw_from_cos(task["raw_data_url"])
    validity = check_validity(raw_text)
    
    if validity != "valid":
        transition(state, "invalid")
        save_invalid_result(task, validity)
        transition(state, "completed")
        return
    
    transition(state, "valid")
    
    # Step 2: 实体识别
    transition(state, "extracting")
    try:
        entities = extract_entities(raw_text, task["brand_aliases"])
        metrics = extract_metrics(raw_text, entities)
        transition(state, "extracted")
    except Exception as e:
        transition(state, "failed", error=str(e))
        raise
    
    # Step 3: 入库
    save_clean_result(task, entities, metrics)
    transition(state, "completed")

五、实体识别：品牌别名合并与歧义消解

实体识别是清洗链路中最核心的 NLP 环节。主要解决三个问题：别名合并、歧义消解、上下文关联。

5.1 别名合并

品牌在不同 AI 回答中可能以不同形式出现——全称、简称、英文名、产品名。实体识别阶段需要将这些指代词统一映射到标准品牌名：

BRAND_ENTITY_MAP = {
    "品牌A": {"aliases": ["品牌A", "BrandA", "A牌", "A品牌"], "standard": "品牌A"},
    "品牌B": {"aliases": ["品牌B", "BrandB", "B牌"], "standard": "品牌B"},
}

def merge_aliases(text: str, brand_aliases: list, standard_name: str) -> dict:
    """识别文本中出现的所有品牌指代词，合并为标准名"""
    found_aliases = []
    for alias in brand_aliases:
        positions = find_all_positions(text, alias)
        for pos in positions:
            found_aliases.append({"alias": alias, "position": pos})
    
    found_aliases.sort(key=lambda x: x["position"])
    return {
        "standard_name": standard_name,
        "mentioned": len(found_aliases) > 0,
        "matched_aliases": found_aliases,
        "match_count": len(found_aliases),
    }

5.2 歧义消解

品牌名可能与通用词冲突。例如“苹果”可能是品牌也可能是水果。歧义消解通过上下文窗口判断：

def disambiguate(text: str, alias: str, position: int) -> bool:
    """判断某次匹配是否为有效的品牌提及"""
    window_start = max(0, position - 30)
    window_end = min(len(text), position + len(alias) + 30)
    context = text[window_start:window_end]
    
    # 排除规则
    exclude_patterns = {
        "苹果": ["吃苹果", "苹果味", "苹果汁", "苹果手机壳"],  # 非品牌语境
    }
    
    patterns = exclude_patterns.get(alias, [])
    for pattern in patterns:
        if pattern in context:
            return False
    return True

5.3 上下文关联

有些品牌提及以代词形式出现（“它”“该品牌”），需要通过上下文关联确定指代对象。简单策略是向前搜索最近的品牌名：

def resolve_pronoun_reference(text: str, pronoun_pos: int) -> str | None:
    """将代词关联到最近出现的品牌名"""
    preceding_text = text[:pronoun_pos]
    for brand_standard, config in BRAND_ENTITY_MAP.items():
        last_pos = -1
        for alias in config["aliases"]:
            pos = preceding_text.rfind(alias)
            if pos > last_pos:
                last_pos = pos
        if last_pos > -1:
            return brand_standard
    return None

六、结构化结果入库

清洗完成后，写入两张表：主结果表存指标，实体明细表存匹配细节。

-- 清洗结果主表
CREATE TABLE clean_result (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    clean_task_id VARCHAR(64) NOT NULL UNIQUE,
    source_task_id VARCHAR(64) NOT NULL,
    batch_id VARCHAR(64) NOT NULL,
    brand_name VARCHAR(100) NOT NULL,
    platform VARCHAR(30) NOT NULL,
    scene_type VARCHAR(30) NOT NULL,
    
    is_valid TINYINT(1) DEFAULT 1,
    invalid_reason VARCHAR(50),
    
    is_mentioned TINYINT(1) DEFAULT 0,
    mention_count INT DEFAULT 0,
    is_recommended TINYINT(1) DEFAULT 0,
    recommendation_rank INT DEFAULT 0,
    has_citation TINYINT(1) DEFAULT 0,
    
    clean_state VARCHAR(20) DEFAULT 'completed',
    raw_data_url VARCHAR(512),
    cleaned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_batch (batch_id),
    INDEX idx_brand_scene (brand_name, scene_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

-- 实体匹配明细表
CREATE TABLE entity_match_detail (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    clean_task_id VARCHAR(64) NOT NULL,
    standard_name VARCHAR(100) NOT NULL,
    matched_alias VARCHAR(100) NOT NULL,
    match_position INT COMMENT '在原文中的位置',
    is_valid_match TINYINT(1) DEFAULT 1 COMMENT '是否通过歧义消解',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_clean_task (clean_task_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

分表设计的理由：clean_result 是业务查询的主表，保持轻量；entity_match_detail 记录每次别名匹配的细节，当品牌方质疑“为什么某次没识别到我们”时，可以从明细表中回溯具体是别名没覆盖还是歧义消解误判。

七、无效样本分类处理

有效性判断是清洗链路的第一个关卡。不同类型的无效样本处理方式不同：

def check_validity(raw_text: str, query_text: str) -> tuple[str, str | None]:
    """返回 (有效性状态, 无效原因)"""
    if not raw_text or len(raw_text) < 20:
        return "invalid", "empty"
    
    refuse_patterns = ["我还不会", "暂时无法回答", "我还没有学会", "无法提供"]
    for pattern in refuse_patterns:
        if pattern in raw_text[:100]:
            return "invalid", f"refuse:{pattern}"
    
    # 可选：语义相关性判断
    similarity = compute_similarity(raw_text[:200], query_text)
    if similarity < 0.3:
        return "invalid", "irrelevant"
    
    return "valid", None

无效样本统一写入 clean_result 表（is_valid=0），保留在样本库中用于后续的“数据质量分析”——如果某个平台的拒绝率突然飙升，可能是接口变动或内容策略调整的信号。

八、失败重试与死信处理

清洗失败有两种情况，策略不同：

CLEAN_RETRY_CONFIG = {
    "transient": {"max_retries": 2, "delay_seconds": 30},   # 临时故障
    "permanent": {"max_retries": 0},                         # 永久失败
}

def handle_clean_failure(task: dict, error: Exception):
    error_type = classify_error(error)  # transient / permanent
    config = CLEAN_RETRY_CONFIG.get(error_type, {"max_retries": 0})
    
    retry_count = task.get("clean_retry_count", 0)
    if retry_count < config["max_retries"]:
        task["clean_retry_count"] = retry_count + 1
        requeue_with_delay(task, config["delay_seconds"])
    else:
        update_state(task["clean_task_id"], "failed", error=str(error))
        send_to_dlq(task, str(error))

永久失败（如原始数据格式损坏）直接进入死信队列，通知运维人工介入。临时失败（如数据库连接超时）延迟重试。

九、清洗质量监控查询

清洗链路稳定运行后，需要监控两类指标：

-- 清洗成功率（按小时）
SELECT 
    DATE_FORMAT(cleaned_at, '%Y-%m-%d %H:00') AS hour,
    COUNT(*) AS total,
    SUM(CASE WHEN clean_state = 'completed' THEN 1 ELSE 0 END) AS completed,
    SUM(CASE WHEN is_valid = 0 THEN 1 ELSE 0 END) AS invalid_count
FROM clean_result
WHERE cleaned_at >= DATE_SUB(NOW(), INTERVAL 24 HOUR)
GROUP BY hour
ORDER BY hour DESC；

-- 各品牌别名匹配覆盖率
SELECT 
    brand_name,
    COUNT(DISTINCT clean_task_id) AS total_tasks,
    SUM(is_mentioned) AS mentioned_tasks,
    ROUND(SUM(is_mentioned)*100.0/COUNT(*), 1) AS mention_rate
FROM clean_result
WHERE is_valid = 1
  AND cleaned_at >= '2026-07-01'
GROUP BY brand_name；

十、工程实践要点

1. 清洗状态历史是排障利器

生产中最难排查的问题不是“失败了”，而是“卡住了”。state_history 字段记录每次状态变更的时间戳，配合监控告警（如某状态停留超过 5 分钟），可以在问题扩大前定位到具体卡点。

2. 别名表要做成可配置的

品牌别名不是一成不变的——品牌会换名、新产品线会新增。将别名映射做成数据库配置表而非代码常量，运营人员可随时更新，清洗函数启动时从配置表加载到内存缓存。

3. 无效样本不要直接丢弃

标记为 invalid 的样本同样入库。一方面用于数据质量分析（无效率趋势），另一方面在诊断“为什么某品牌提及率低”时，需要区分“AI 没提到”和“样本本身无效”——两者业务含义完全不同。

十一、结语

样本清洗链路是 AI 回答监测系统从“能用”到“可靠”的分水岭。独立的清洗队列让采集和清洗各自聚焦，状态机让每一步处理可追踪可重入，实体识别和别名合并解决了多品牌指代词的核心痛点，分表存储兼顾了查询效率和细节回溯。

这套方案已在消费品牌、企业服务等多个行业的多平台 AI 可见度监测中实际运行，日均处理清洗任务千条以上。开发者可在此基础上，扩展实体识别的能力（如产品名识别、关系抽取），或引入轻量级文本分类模型提升无效样本的自动识别准确率。

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

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

人工智能