多平台 AI 答案采集的架构选型:同步、异步与断点续跑

多平台 AI 答案采集的难点，从来不只是“调用 API”，而是同时对接多个协议完全不同的接口，并确保其中一个平台失败时，不会拖垮整批任务。

如果只采集单个平台，并且只运行一次，那么同步循环配合 try-except 通常已经够用。但当任务扩大到几十道问题、多个平台，并且需要定期复跑和对比时，同步阻塞式脚本很快会遇到两个问题：

1. 某个平台限流或超时，会拖慢整批任务。
2. 中途失败后，无法准确区分“尚未执行”“执行失败”和“执行成功”。

这三种状态一旦混在一起，断点续跑和重试逻辑就很容易出错。

先说结论：多平台采集系统的核心不是选择同步还是异步框架，而是如何定义一条任务已经“完成”。只要完成状态设计清楚，并发模型反而是次要问题。

一、协议差异很难通过统一接口完全屏蔽

不同对话式 AI 平台的联网检索和引用接口，在协议设计上可能存在明显差异。

例如：

• 有的平台在普通 /chat/completions 端点中不支持联网检索参数，必须切换到专门的响应式端点，才能获得检索结果。
• 有的平台同时提供官方原生协议和 OpenAI 兼容协议。兼容协议接入更方便，但可能无法返回完整的检索过程和引用字段。
• 有的平台会直接下线旧模型别名。如果代码中写死了模型名称，接口可能在没有代码变更的情况下突然失效。
• 部分接口使用普通 JSON 响应，另一些接口则依赖 SSE 流式事件，解析和结束判断方式并不相同。

因此，“统一采集层”在实践中往往只能统一输出，无法真正统一请求。

更现实的架构是：为每个平台分别编写独立适配器，对外输出相同的数据结构，内部则自行处理协议差异。

统一记录可以包含以下字段：

{
  "question_id": "q_001",
  "question": "问题内容",
  "platform": "platform_a",
  "raw_response": {},
  "citations": [],
  "status": "success",
  "status_code": 200,
  "error_message": null
}

这种设计遵循一个简单原则：

适配器之间不共享请求逻辑，只共享落盘格式。

当某个平台修改鉴权方式、模型名称或返回结构时，只需要调整对应适配器，不会影响其他平台。

SSE 流式响应必须等待明确的完成事件

使用 SSE 接收流式响应时，不能简单地把“连接断开”或“请求超时”当作回答结束。

更稳妥的方式是显式等待平台定义的完成事件，例如：

response.completed
message.completed
[DONE]

如果在没有收到完成事件时就保存结果，可能会把尚未生成完毕的半截答案当作最终数据，后续又很难通过普通状态码发现问题。

因此，流式接口的有效性判断至少应同时检查：

• 是否收到明确的完成事件；
• 是否获得完整正文；
• 引用字段是否完成解析；
• 响应是否包含平台返回的错误事件。

二、“完成”的定义决定断点续跑是否可靠

断点续跑最常见的问题，并不是代码不会读取历史文件，而是“完成”这个状态定义得过于粗糙。

如果系统仅使用“这道题是否运行过”作为完成标记，就会遗漏一种重要情况：

请求已经执行，但采集失败，或者返回内容未通过有效性校验。

这种记录不应该被视为已完成。否则后续复跑时，系统会永久跳过这条失败数据。

使用“题目＋平台”作为任务唯一键

多平台采集任务的最小执行单位不是一道题，而是一个“题目×平台”组合。

可以使用类似下面的唯一键：

question_id + platform

例如：

q_001:platform_a
q_001:platform_b
q_001:platform_c

只有通过有效性校验的记录，才加入“有效完成集合”。

失败记录仍然需要落盘，但状态应明确标记为失败：

{
  "question_id": "q_001",
  "platform": "platform_b",
  "raw_response": null,
  "citations": [],
  "status": "failed",
  "status_code": 429,
  "error_message": "rate limit exceeded"
}

失败记录不能从统计分母中消失

假设一次任务需要采集 53 道问题，每道问题运行三个平台，那么理论任务总数为：

53 × 3 = 159

如果其中一条采集失败，最终仍然应该产生 159 条记录，其中：

• 158 条成功；
• 1 条失败占位记录。

统计分母仍然是 159，而不是因为失败自动缩小为 158。

失败占位记录有两个作用：

1. 保留完整的执行和审计痕迹；
2. 避免统计指标因为失败数据被删除而产生偏差。

断点续跑只需要一条规则

复跑时遍历全部“题目×平台”组合：

for question in questions:
    for platform in platforms:
        task_key = f"{question.id}:{platform}"

        if task_key not in valid_completed_keys:
            run_collection(question, platform)

凡是不在有效完成集合中的任务，无论是从未执行，还是执行过但失败，都统一重新请求。

这样不需要额外维护容易失真的“重试队列”，整个采集过程也更接近幂等执行：

• 成功任务不会重复采集；
• 失败任务可以自动补跑；
• 中途退出后可以继续；
• 多次复跑不会改变已经确认有效的结果。

三、重试策略还要考虑网络路径切换

长批次采集中的失败，不一定都来自平台限流，也可能来自本地网络环境。

例如：

• 直连网络临时不可用；
• 代理节点发生切换；
• DNS 或 TLS 握手短暂失败；
• 某个平台只能通过特定网络出口访问；
• 代理状态变化导致已有连接中断。

如果重试逻辑只是“失败后立即使用原路径再次请求”，遇到网络出口抖动时，很可能连续失败，直到耗尽重试次数。

更稳妥的方案是让重试策略感知网络路径，在允许的网络环境中交替尝试不同出口：

第一次：直连
第二次：代理
第三次：直连
第四次：代理

同时还应加入以下控制：

• 指数退避或分级等待；
• 单次请求超时；
• 平台级并发上限；
• 全局请求间隔；
• 429、5xx 和网络异常的分类处理；
• 不可重试错误的提前终止。

例如，鉴权失败和参数错误通常不应持续重试，而限流、服务端错误和临时网络中断则可以进入退避重试。

RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}

if response.status_code in RETRYABLE_STATUS_CODES:
    schedule_retry()
else:
    mark_as_non_retryable_failure()

重试的目标不是无条件提高次数，而是根据失败类型选择下一步动作。

四、不要把单个平台故障扩散到整批任务

多平台任务需要做到故障隔离。

平台 A 超时，不应该阻止平台 B 和平台 C 继续执行；某一道问题失败，也不应该中断剩余几十道问题。

因此，每个“题目×平台”任务都应该拥有独立的：

• 超时控制；
• 重试次数；
• 状态记录；
• 异常捕获；
• 落盘结果。

异步并发可以提升吞吐量，但它并不是故障隔离的前提。即使使用同步实现，也应该保证单个任务异常不会跳出整个批次。

for task in tasks:
    try:
        result = collect(task)
        save_result(result)
    except Exception as exc:
        save_failure_placeholder(task, exc)

如果采用异步方式，则还需要避免一个协程抛出异常后取消整组任务。核心原则仍然不变：任务可以失败，但批次必须继续。

五、聚合统计最危险的是“静默出错”

采集阶段的接口错误通常比较明显：请求会超时、返回错误码，或者直接抛出异常。

聚合统计阶段的错误反而更隐蔽，因为它们可能不会产生任何报错。

一个典型案例是变量名冲突：

c = Counter(records)

for item in citations:
    for c in item:
        ...

前半部分使用变量 c 保存计数器，后面的循环又复用了同名变量。新的赋值覆盖了原来的计数对象。

如果后续代码使用字典的 .get() 方法取值：

value = c.get(target, 0)

即使目标键不存在，也不会抛出异常，只会返回默认值 0。

最终可能出现这样的结果：

正确命中数：7
报告输出值：0

整个过程没有异常，输出数字看起来也完全合理，但统计结果已经错误。

不要只依赖“编码时更仔细”

这类问题更适合通过结构性约束解决：

• 避免在同一作用域复用单字母变量名；
• 将采集、解析和统计拆分为不同函数；
• 对关键指标设置合理区间断言；
• 在生成报告前进行二次计算；
• 使用不同算法交叉核对关键数字；
• 检查成功数、失败数和总任务数是否守恒。

例如：

assert success_count + failed_count == total_expected

对于关键命中数，可以分别通过“记录级扫描”和“聚合结果求和”计算一次：

count_by_scan = sum(
    1 for record in records
    if target_keyword in record["answer"]
)

count_by_group = grouped_stats.get(target_keyword, 0)

assert count_by_scan == count_by_group

这种校验可以在最终报告生成前发现静默统计错误。

六、建议的数据处理流程

一个可长期复跑的多平台采集流程，可以拆分为以下阶段：

生成题目×平台任务
        ↓
读取历史有效记录
        ↓
过滤已经成功的任务
        ↓
按平台适配器执行请求
        ↓
等待完整响应并进行有效性校验
        ↓
成功记录或失败占位记录落盘
        ↓
重新计算有效完成集合
        ↓
聚合统计
        ↓
独立交叉校验
        ↓
生成最终报告

其中最关键的不是并发数量，而是以下三个边界是否清晰：

1. 什么是一个独立任务；
2. 什么样的响应才算有效完成；
3. 失败记录如何保留并参与统计。

总结

多平台 AI 答案采集系统的复杂度，主要集中在三个方面：

第一，平台协议差异应该通过独立适配器隔离。可以统一输出格式，但不要强行统一不同平台的请求逻辑。

第二，完成状态应该按照“题目＋平台＋有效性”定义。只有有效记录才算完成，失败记录需要保留为占位数据，并继续纳入总分母。

第三，聚合统计必须防范静默错误。关键指标不能只相信第一次计算结果，而应通过守恒校验和独立算法交叉核对。

这些基础设计稳定之后，无论系统最终采用同步、线程池还是异步协程，都可以在平台限流、网络抖动、中途退出和重复执行的情况下保持可恢复、可审计和可维护。

本文写作过程使用了 AI 辅助工具，内容已经人工核校。