品牌团队在监测AI平台（如豆包、文心一言、通义千问等）对品牌相关问题的回答时，面临着多平台API协议、认证方式、限流策略和响应格式各异的挑战。直接拼接调用会导致代码耦合高、维护成本大、异常难以追踪。本文将深入介绍一种基于适配器模式的多平台采集架构，旨在统一调用接口、数据格式和错误处理，并提供关键实现片段及异常处理策略，帮助您高效、稳定地构建多平台AI回答采集系统。

业务背景与实际约束

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

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

实际约束包括：

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

小提示：实际部署时，建议为每个平台设置独立的袋里或网络策略，以应对可能的网络波动。

问题现象与复现过程

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

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

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

原因分析

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

候选技术方案对比

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

核心实现过程

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

小提示：AIResponse 数据结构统一了成功和失败状态，便于后续处理与存储。

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

小提示：重试策略中采用了递增的等待时间（retry_delay * (attempt + 1)），以避免对平台造成持续压力。

4. 数据格式统一与入库

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

import sqlite3

def sa ve_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次调用，结果如下：

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

踩坑及风险边界

以下是实现过程中需要特别注意的风险点和应对策略：

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

常见问题

Q1：重试时如何避免重复计费？

通过为每个请求生成唯一请求ID（例如UUID），并将其附加到请求中。平台方根据此ID实现幂等性，确保同一ID的请求只被处理一次。

Q2：当平台限流时，应如何处理？

在捕获到HTTP 429状态码后，首先读取响应头中的Retry-After字段，并使用该值作为等待时间。如果该字段不存在，可采用指数退避策略进行重试。同时，记录限流事件，以便后续优化调度策略。

Q3：如何安全地管理多个平台的API Key？

建议将所有API Key存储在环境变量或专业的密钥管理服务（如HashiCorp Vault、AWS Secrets Manager）中。在代码中通过环境变量注入，避免硬编码。

可复用经验总结

以下是从本方案中提炼出的核心经验，可为您在实际项目中提供参考：

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

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