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

在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密集型。两种不同类型的任务，队列配置自然也要差异化对待：

批量消费可以减少网络往返开销。单次拉取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 状态定义

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")
        sa ve_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: 入库
    sa ve_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可见度监测中实际运行，日均处理清洗任务千条以上。开发者可以在此基础上，扩展实体识别的能力，比如产品名识别、关系抽取，或者引入轻量级文本分类模型，进一步提升无效样本的自动识别准确率。