多平台 AI 回答采集任务如何做调度、归一化和结果入库？

一、问题的提出

当我们决定系统性地观察品牌在 AI 回答中的表现时，遇到的第一个工程问题不是“怎么分析”，而是“怎么把数据高质量地收回来”。

想象一个典型场景：我们需要对 5 个主流 AI 平台、面向 100 个品牌、每个品牌配置 20 个测试问题、每个问题重复 3 轮采样。这意味着一轮完整测评需要发起 30000 次独立的 AI 问答请求。这些请求需要有序执行、失败可重试、结果可追溯、回答可归一化，最终落库供分析层查询。

这不是一个脚本能解决的问题，而是一条需要认真设计的数据链路。

本文将从云上任务调度、多平台采集适配、失败重试机制、回答归一化处理、结构化入库和一致性结果查询六个层面，完整拆解这条链路的工程实践思路。

二、整体架构概览

整条链路的核心抽象只有四个字：采、存、算、查。

任务管理层 → 采集执行层 → 归一化处理层 → 结构化入库层 → 查询分析层

• 任务管理层：负责拆解测评任务、生成采集子任务、管理执行状态
• 采集执行层：对接不同 AI 平台的问答接口，完成实际采样
• 归一化处理层：清洗原始回答、识别品牌实体、标准化数据结构
• 结构化入库层：将归一化结果写入数据库，建立索引和关联关系
• 查询分析层：提供面向业务的一致性查询能力

下面逐层展开。

三、任务调度：从“跑脚本”到“任务编排”

3.1 为什么需要调度层

直接写一个循环遍历所有品牌、问题和平台的脚本，在数据量小的时候勉强能用。但当采样规模上升到数万次，以下问题会集中爆发：

• 执行时间过长：串行请求 30000 次，按每次 10 秒估算，需要 80 多个小时
• 失败不可控：某个平台的接口临时限流，整个脚本可能中断，已采集的数据状态不明确
• 不可观测：不知道当前进度、失败分布、异常原因
• 无法复用：每次新测评都要重新写脚本

调度层的核心价值，是把“一次大规模采集”拆解为“一批可管理、可观测、可重试的子任务”。

3.2 任务拆解策略

我们采用三级任务模型：

测评任务（Project）
  └── 采集批次（Batch）
        └── 采样单元（Sample Unit）

• 测评任务：一次完整的品牌测评，包含被测对象、问题集、平台列表、采样轮次等元信息
• 采集批次：按平台或品牌维度划分的执行分组，例如“豆包平台 × 品牌A”为一个批次
• 采样单元：最小执行粒度，即“某个平台 × 某个品牌 × 某个问题 × 某轮采样”

这种拆法有几个好处。批次之间可以并行，同一批次内的采样单元串行还是并行取决于平台限制。失败重试可以精确到采样单元级别，不影响已完成的数据。

3.3 调度实现选型

在云上实现任务调度，几种常见方案的对比如下：

实践中，我们倾向于组合使用：用 Serverless Workflow 管理整体流程编排，具体的采样请求通过消息队列异步投递到云函数消费者执行。这样既保证了流程的可视化和重试能力，又获得了云函数的弹性伸缩。

3.4 状态机设计

每个采样单元在生命周期中经历以下状态：

PENDING → QUEUED → RUNNING → SUCCESS
                           → FAILED → RETRYING → SUCCESS
                                           → PERMANENTLY_FAILED

关键设计点：

• 状态持久化：每次状态变更写入数据库，不依赖内存
• 超时检测：RUNNING 状态超过阈值自动标记为 TIMEOUT，进入重试队列
• 终态明确：SUCCESS 和 PERMANENTLY_FAILED 都是终态，不会被再次调度
• 幂等保证：同一个采样单元重复执行不会产生重复记录

四、多平台采集：统一抽象下的差异化适配

4.1 平台差异的现实

不同的 AI 平台，接口形式、认证方式、返回结构、联网能力各不相同。有的提供公开 API，有的只有 Web 端交互，有的支持联网搜索，有的完全依赖模型自身知识。

这意味着采集层需要处理三种典型的交互模式：

• API 调用模式：有标准接口，直接请求获取结构化返回
• Web 自动化模式：需要模拟浏览器交互，从页面提取回答
• 平台集成模式：通过云平台提供的 SDK 调用模型服务

4.2 适配器模式设计

我们为每个平台实现统一的 PlatformAdapter 接口：

class PlatformAdapter(ABC):
    @abstractmethod
    def ask(self, question: str, context: dict) -> SampleResult:
        """执行一次问答采样"""
        pass
    
    @abstractmethod
    def validate_response(self, raw: dict) -> bool:
        """校验返回结果是否有效"""
        pass
    
    @abstractmethod 
    def extract_answer(self, raw: dict) -> str:
        """从原始返回中提取纯文本回答"""
        pass

每个平台的具体适配器负责处理差异：

• 认证差异：API Key、Token、Cookie 等统一在适配器初始化时注入
• 请求格式差异：多轮对话、系统提示词、联网开关等参数在适配器内封装
• 返回格式差异：统一提取为纯文本回答 + 元信息（模型版本、思考过程、引用来源等）
• 限流策略差异：每个适配器维护独立的速率控制逻辑

4.3 采样结果的统一结构

无论来自哪个平台，一次采样结果最终都归一化为统一结构：

{
  "sample_id": "uuid",
  "platform": "doubao",
  "model_version": "xxx",
  "question": "有哪些适合企业使用的品牌监测工具？",
  "question_intent": "推荐决策",
  "brand_name": "某品牌",
  "raw_answer": "AI 的完整原始回答...",
  "clean_answer": "清洗后的回答文本",
  "think_content": "思考过程（如有）",
  "citations": ["引用来源1", "引用来源2"],
  "sample_time": "2026-07-06T10:00:00Z",
  "round": 2,
  "duration_ms": 3200,
  "status": "success"
}

这个统一结构是后续归一化处理的基础。

五、失败重试：不只是“再来一次”

5.1 失败分类

不是所有失败都应该重试。我们需要区分三类失败：

可重试失败：网络超时、服务暂时不可用、限流拒绝、连接重置等临时性问题。重试策略：指数退避，间隔 1s / 5s / 15s / 60s。

需换策失败：账号被限制、IP 被封、触发内容审核、返回明显异常（如空回答、乱码）。重试策略：切换账号或出口 IP，间隔较长。

不可重试失败：问题本身不合法、平台明确拒绝此类内容、模型版本不支持。直接标记为 PERMANENTLY_FAILED，记录原因。

5.2 退避与熔断

简单的固定间隔重试在海量请求下可能导致雪崩。设计要点：

• 指数退避加随机抖动：避免重试风暴同时到达
• 平台级熔断：某个平台连续失败超过阈值，暂停该平台所有采样一段时间
• 批次级隔离：一个批次的重试不影响其他批次的正常执行

5.3 重试的幂等性保证

采样单元使用确定性 ID 生成规则，例如：

sample_id = hash(platform + brand + question_id + round + date)

同一轮测评中，相同参数生成的 ID 相同。重试时使用 upsert 语义写入，保证重复执行不会产生重复记录，也方便查询“这个采样单元最终成功了没有”。

六、回答归一化：从原始文本到结构化指标

6.1 为什么需要归一化

不同平台返回的回答形态差异巨大。有的返回纯文本，有的包含 Markdown 格式，有的附带思考过程，有的引用来源混在正文中。归一化的目标是把这些异构数据，转化为可以统一计算指标的结构化记录。

6.2 归一化流程

归一化处理分为四个步骤：

第一步：文本清洗

• 去除 Markdown 标记、HTML 标签、多余空白
• 分离思考过程（如 DeepSeek 的 reasoning_content）和正式回答
• 提取引用链接和来源说明
• 保留回答的段落结构信息

第二步：品牌实体识别

• 基于预设的“品牌-别名”映射表，在回答中识别目标品牌及其变体
• 处理简称与全称（“绿雪智能科技” vs “绿雪”）、中英文名、产品名与公司名
• 处理同名歧义（通过上下文判断是否为测评对象）
• 标记“提及位置”（首段、中段、尾段、推荐列表中等）

第三步：语义角色标注

• 判断品牌在回答中扮演的角色：
  • 被提及：回答中出现了品牌名称
  • 被推荐：品牌出现在推荐列表、建议方案中
  • 被引用：回答引用了品牌官网或公开内容作为来源
  • 被比较：品牌与竞品一起出现，存在对比语义
  • 被描述：品牌被解释、评价，存在正/负/中性倾向

第四步：指标聚合

• 每个采样单元生成一条归一化记录
• 同一问题 × 同一品牌 × 同一平台的多轮结果，聚合为“该问题维度”的初步指标
• 跨问题、跨平台的聚合留到查询分析层处理

6.3 一个归一化记录的例子

{
  "sample_id": "xxx",
  "brand": "绿雪智能科技",
  "brand_alias_matched": "绿雪",
  "mentioned": true,
  "recommended": true,
  "recommend_position": "首位推荐",
  "cited": false,
  "sentiment": "positive",
  "sentiment_keywords": ["功能完整", "值得考虑", "适合企业"],
  "risk_flags": [],
  "anomaly_type": null,
  "human_review_required": false
}

这样一层处理之后，查询“某个品牌在所有平台上的推荐率”就不再需要扫描原始回答文本，直接做结构化聚合即可。

七、结构化入库：为查询而设计

7.1 数据模型设计

数据库设计需要同时满足写入效率和分析查询需求。我们区分三类表：

任务元数据表

project | batch | sample_unit | platform_config

记录任务状态、批次进度、平台配置等。

采样原始数据表

sample_result (sample_id, project_id, platform, question, raw_answer, ...)

完整保存每次采样的原始回答，用于追溯和复核。按时间分区，便于归档历史数据。

归一化结果表

normalized_result (sample_id, brand, mentioned, recommended, cited, sentiment, ...)

与采样记录一对一关联，存储归一化后的结构化指标。这是分析查询的主表。

7.2 索引策略

分析场景的查询模式相对固定，主要围绕：

• 按品牌查询所有采样结果
• 按平台查询某个品牌的指标
• 按问题意图分类汇总
• 按时间范围对比趋势变化

因此核心索引设计为：

-- 品牌维度查询
CREATE INDEX idx_brand_project ON normalized_result(brand, project_id);

-- 平台 × 品牌 × 意图 组合查询
CREATE INDEX idx_platform_intent ON normalized_result(platform, brand, intent);

-- 时间范围扫描
CREATE INDEX idx_sample_time ON sample_result(sample_time);

7.3 写入优化

大规模采样会产生较高的写入并发。几个优化策略：

• 批量写入：归一化结果攒到一定数量后批量 insert，减少事务开销
• 异步写入：采样完成和写入解耦，通过消息队列缓冲
• 写入确认机制：确保原始数据落库成功后才标记采样单元为 SUCCESS，保证不丢数据

八、一致性查询：让指标可解释

8.1 一致性问题的来源

AI 回答的动态性决定了，同一个问题今天问和明天问，结果可能不同。我们做测评，不是消灭这个差异，而是让这个差异可见、可量化、可追溯。

因此查询层需要回答两类问题：

• 快照查询：在某个时间点，品牌的各项指标是多少
• 一致性问题：同一个问题在不同轮次、不同时间的结果是否稳定

8.2 查询接口设计

面向业务层的查询接口，不是直接暴露数据表，而是提供语义清晰的聚合视图：

提及率查询

GET /api/metrics/mention-rate?brand=X&project_id=Y&platform=Z

返回：有效回答总数、提及次数、提及率、各平台分拆

推荐一致性查询

GET /api/metrics/recommend-stability?brand=X&question_id=Y

返回：该问题多轮采样中，品牌被推荐的次数占比，以及推荐位置的波动情况

异常采样下钻

GET /api/samples/anomalies?brand=X&anomaly_type=empty_response

返回：所有被标记为异常的采样记录，可直接查看原始回答

8.3 查询结果的边界说明

查询接口返回数据时，附带必要的上下文信息，避免数据被断章取义：

• 采样时间范围
• 涉及平台列表
• 有效样本总量
• 异常样本数量和比例
• 指标计算方式说明

这样做不是增加复杂度，而是让每一个指标都可解释、可复核。看到一个“提及率 65%”，用户应该能立刻知道这个数字是在什么条件下得出的。

九、工程实践中的几个关键取舍

9.1 实时性 vs 规模

单次采样的实时性和整体测评的完成时间是一对矛盾。我们的选择是：单个采样追求稳定而非速度，整体流程通过并行压缩时间。每个采样请求设置合理的超时（30-60 秒），不因为个别慢响应拖垮整体。大规模测评的整体耗时通过增加消费者实例数量来控制。

9.2 全量存储 vs 成本控制

AI 回答文本通常较长，全量存储所有轮次的原始回答会带来存储成本。但做归一化和复核又需要原始数据。折中方案是：原始回答全量保留，按时间分区，冷数据自动归档到对象存储。归一化结果永久保留在关系型数据库中供快速查询。

9.3 自动化 vs 人工复核

品牌识别和语义标注的自动化准确率可以做到较高，但边界样本始终存在。设计上预留人工复核接口，自动识别置信度较低的样本（品牌名歧义、语义模糊、回答异常），标记为 human_review_required，通过独立流程处理。自动化和人工不是替代关系，而是分层协作。

十、从工程链路到业务价值

回到文章开头的问题：多平台 AI 回答采集怎么做调度、归一化和结果入库。

本质上，我们是在把“生成式 AI 回答”这种原本分散、动态、非结构化的信息，转化为可以系统观察、持续监测、交叉复核的结构化数据资产。

这套工程链路本身不是目的。它真正的价值在于，当企业或研究团队想要了解“我的品牌在 AI 回答中到底表现如何”时，不再只能靠偶尔手动测试几个问题、凭感觉下结论，而是可以拿出一套有采样规模、有时间跨度、有平台对比、可追溯原始回答的量化观察结果。

对于关心品牌 AI 可见度的团队来说，这是从“感觉阶段”走向“数据阶段”的基础设施。