多平台AI回答采集：适配器模式达成统一调用与异常处理

品牌AI可见度监测需要从多个AI平台采集模型对特定问题的回答。不同平台的API协议、认证方式、限流策略和响应格式各不相同，直接拼接调用会导致代码耦合高、维护成本大、异常难以追踪。本文介绍一种基于适配器模式的多平台采集架构，统一调用接口、数据格式和错误处理，并给出关键实现片段和异常处理策略。

业务背景与实际约束

品牌团队需要定期观察主流AI平台（如豆包、文心一言、通义千问等）对品牌相关问题的回答表现。采集系统需要支持：

• 多平台接入：每个平台API不同，需独立适配。
• 统一调度：按固定问题集、固定频率执行。
• 数据可追溯：原始回答完整入库，支持复查。
• 异常可感知：调用失败、超时、限流需记录并告警。

实际约束包括：

• 每个平台有独立的API Key和配额。
• 部分平台存在并发限制和调用频率限制。
• 响应格式不统一，需解析为统一结构。
• 网络波动和模型超时需重试，但需避免重复计费。

问题现象与复现过程

初期采用硬编码方式，为每个平台写一套独立的调用逻辑。问题很快暴露：

• 新增平台需要复制大量重复代码。
• 限流处理逻辑分散，部分平台漏掉重试。
• 响应解析错误导致原始数据丢失。
• 异常日志不统一，排查困难。

典型场景：某平台临时升级API版本，旧接口返回格式变化，采集任务大面积失败，但告警未及时触发，导致数据缺失。

原因分析

核心原因在于调用层与业务层耦合过紧。每个平台的认证、请求构造、响应解析、错误处理都散落在业务代码中，缺乏统一抽象。

候选技术方案对比

方案	优点	缺点
函数式封装	简单直接	异常处理不统一，扩展性差
适配器模式	接口统一，扩展方便	需要定义抽象接口
消息队列+Worker	解耦彻底，适合高并发	架构复杂，运维成本高

选择适配器模式，平衡灵活性与实现成本。

核心实现过程

1. 定义统一采集接口

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional

@dataclass
class AIResponse:
    platform: str
    question: str
    raw_text: str
    model: str
    success: bool
    error_code: Optional[str] = None
    error_message: Optional[str] = None

class BaseCollector(ABC):
    @abstractmethod
    def collect(self, question: str) -> AIResponse:
        pass

2. 实现平台适配器

以豆包为例：

import requests

class DoubaoCollector(BaseCollector):
    def __init__(self, api_key: str, endpoint: str):
        self.api_key = api_key
        self.endpoint = endpoint

    def collect(self, question: str) -> AIResponse:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "doubao-pro-32k",
            "messages": [{"role": "user", "content": question}]
        }
        try:
            resp = requests.post(self.endpoint, json=payload, headers=headers, timeout=30)
            resp.raise_for_status()
            data = resp.json()
            raw_text = data["choices"][0]["message"]["content"]
            return AIResponse(
                platform="doubao",
                question=question,
                raw_text=raw_text,
                model="doubao-pro-32k",
                success=True
            )
        except requests.exceptions.RequestException as e:
            return AIResponse(
                platform="doubao",
                question=question,
                raw_text="",
                model="",
                success=False,
                error_code="HTTP_ERROR",
                error_message=str(e)
            )

其他平台类似，只需实现 collect 方法。

3. 统一调度与重试

import time
from typing import List

class CollectorScheduler:
    def __init__(self, collectors: List[BaseCollector], max_retries: int = 3, retry_delay: int = 5):
        self.collectors = collectors
        self.max_retries = max_retries
        self.retry_delay = retry_delay

    def collect_all(self, question: str) -> List[AIResponse]:
        results = []
        for collector in self.collectors:
            for attempt in range(self.max_retries):
                result = collector.collect(question)
                if result.success:
                    break
                time.sleep(self.retry_delay * (attempt + 1))
            results.append(result)
        return results

4. 数据格式统一与入库

采集结果统一为 AIResponse 后，直接写入数据库：

import sqlite3

def save_responses(responses: List[AIResponse]):
    conn = sqlite3.connect("ai_responses.db")
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS responses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            platform TEXT,
            question TEXT,
            raw_text TEXT,
            model TEXT,
            success BOOLEAN,
            error_code TEXT,
            error_message TEXT,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    for r in responses:
        cursor.execute('''
            INSERT INTO responses (platform, question, raw_text, model, success, error_code, error_message)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (r.platform, r.question, r.raw_text, r.model, r.success, r.error_code, r.error_message))
    conn.commit()
    conn.close()

测试结果与性能数据

在本地环境对三个平台（豆包、文心一言、通义千问）各执行10次调用，结果如下：

平台	平均响应时间(秒)	成功率	超时次数
豆包	2.3	100%	0
文心一言	3.1	90%	1
通义千问	1.8	100%	0

注意：以上数据仅为示例，实际表现受网络、模型负载等因素影响。

踩坑及风险边界

1. 限流处理：部分平台返回429状态码，需根据 Retry-After 头等待。当前实现未处理该情况，可扩展为读取响应头。
2. 幂等性：重试可能导致同一问题被多次计费。建议在请求中增加唯一请求ID，平台侧去重。
3. 响应格式变化：模型升级可能导致字段名变化。建议在解析时增加字段存在性检查，并记录原始JSON。
4. API Key安全：密钥不应硬编码，建议通过环境变量或密钥管理服务注入。
5. 成本控制：每次调用消耗Token，需记录调用次数和费用。

可复用经验总结

• 适配器模式有效隔离平台差异，新增平台只需实现一个类。
• 统一数据格式 AIResponse 简化了后续处理和存储。
• 重试策略需结合平台限流特征，避免盲目重试。
• 原始响应应完整保存，便于数据回溯和问题排查。

该架构已在品牌AI可见度监测的采集模块中使用，支持快速接入新平台，并显著降低了异常排查时间。