多平台AI回答采集系统：适配器模式 异常处理与数据入库实践

采集多个AI平台的模型回答时，每个平台的API认证、请求格式、响应结构和限流策略都不同。直接拼接调用会导致代码耦合、数据混乱和难以维护。本文从工程实践出发，介绍如何通过适配器模式设计一个多平台采集系统，包括统一数据模型、适配器接口、异常处理与重试策略，以及原始回答入库的实现。适合需要构建或优化AI回答采集系统的后端开发者、数据工程师和AI应用开发者。前提是熟悉HTTP请求、JSON处理和基本的错误处理模式。

业务背景与实际约束

为了持续观察不同AI平台对特定问题的回答情况，需要定期向多个平台发送预设问题并收集原始回答。实际约束包括：

• 认证方式不同：有的平台使用Bearer Token（如豆包），有的使用Access Token（如文心一言），还有的通过签名认证。
• 请求参数格式不一致：字段名、嵌套层级、是否支持流式等各有差异。
• 响应结构差异大：有的直接返回文本，有的包含多层嵌套（如choices[0].message.content）。
• 限流策略不同：有的按QPS限流，有的按每分钟Token数限流，返回的HTTP状态码和Retry-After头也不统一。
• 网络超时、服务不可用等异常需要统一处理。
• 原始回答必须完整保存，以便后续解析和复查。

问题现象与复现过程

初期采用硬编码方式，为每个平台写独立的采集函数。问题很快暴露：

• 新增平台时，需要复制大量重复代码。
• 某个平台的API更新后，需要修改多处调用点。
• 限流和重试逻辑散落在各个函数中，难以统一调整。
• 响应解析失败时，原始数据丢失，无法回溯。

例如，豆包和文心一言的响应结构不同：

// 豆包响应示例（假设）
{
  "id": "chatcmpl-xxx",
  "choices": [{"message": {"content": "回答内容"}}]
}

// 文心一言响应示例（假设）
{
  "result": "回答内容"
}

硬编码方式下，每个平台的解析逻辑都耦合在采集函数中，维护成本高。

原因分析

根本原因是平台差异没有被隔离，导致代码重复和耦合。需要引入设计模式将变化封装起来，对外暴露统一接口。适配器模式适合解决这类问题。

候选技术方案对比

方案	优点	缺点
硬编码	简单直接	重复代码多，维护困难
工厂模式	创建对象灵活	仍需要处理不同接口
适配器模式	隔离平台差异，扩展性好	需要定义统一接口
策略模式	运行时切换算法	不适合固定平台差异

选择适配器模式，因为每个平台的差异是固定的，适配器可以封装所有差异，新增平台只需添加新适配器。

核心实现过程

1. 统一数据模型

定义一个基础的回答模型，所有适配器返回此模型：

from pydantic import BaseModel
from datetime import datetime
from typing import Optional

class AIResponse(BaseModel):
    platform: str  # 平台标识，如"doubao"
    model: str     # 模型名称
    question: str  # 原始问题
    raw_text: str  # 模型返回的原始文本
    request_id: Optional[str]  # 平台请求ID，用于追踪
    timestamp: datetime  # 请求时间
    status: str  # "success" 或 "failed"
    error_message: Optional[str]  # 失败时的错误信息

2. 适配器接口

定义抽象基类：

from abc import ABC, abstractmethod

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

    @abstractmethod
    def parse_response(self, raw_response: dict) -> str:
        """从原始响应中提取回答文本"""
        pass

3. 具体适配器示例

以下为通用示例，具体参数请参考各平台官方文档。

import requests

class DoubaoAdapter(PlatformAdapter):
    def __init__(self, api_key: str, endpoint: str):
        self.api_key = api_key
        self.endpoint = endpoint

    def send_request(self, question: str) -> AIResponse:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "",  # 替换为实际模型名
            "messages": [{"role": "user", "content": question}]
        }
        try:
            resp = requests.post(self.endpoint, headers=headers, json=payload, timeout=30)
            resp.raise_for_status()
            data = resp.json()
            raw_text = self.parse_response(data)
            return AIResponse(
                platform="doubao",
                model="",
                question=question,
                raw_text=raw_text,
                request_id=data.get("id"),
                timestamp=datetime.utcnow(),
                status="success"
            )
        except requests.exceptions.RequestException as e:
            return AIResponse(
                platform="doubao",
                model="",
                question=question,
                raw_text="",
                timestamp=datetime.utcnow(),
                status="failed",
                error_message=str(e)
            )

    def parse_response(self, raw_response: dict) -> str:
        # 假设响应结构为 {"choices": [{"message": {"content": "..."}}]}
        return raw_response["choices"][0]["message"]["content"]

4. 调度器与统一调用

from typing import List
import time

class Collector:
    def __init__(self, adapters: List[PlatformAdapter]):
        self.adapters = adapters

    def collect(self, question: str) -> List[AIResponse]:
        results = []
        for adapter in self.adapters:
            result = adapter.send_request(question)
            results.append(result)
            # 简单限流：每个请求间隔1秒
            time.sleep(1)
        return results

5. 数据存储

将结果存入数据库，保留原始JSON：

def save_response(db_connection, response: AIResponse):
    query = """
        INSERT INTO raw_responses (platform, model, question, raw_text, request_id, timestamp, status, error_message)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
    """
    params = (
        response.platform,
        response.model,
        response.question,
        response.raw_text,
        response.request_id,
        response.timestamp,
        response.status,
        response.error_message
    )
    db_connection.execute(query, params)

关键代码或配置

• 适配器类：每个平台一个，实现PlatformAdapter接口。
• 限流配置：在Collector中通过time.sleep控制请求间隔，或使用更精细的令牌桶。
• 重试装饰器：使用tenacity库实现指数退避重试。

测试结果与性能数据

正常情况下，每个平台应返回成功状态，raw_text非空，request_id存在。限流时日志显示重试，最终成功或失败。网络异常时日志记录错误，状态为failed。数据完整性可通过数据库查询验证。

具体性能数据（如QPS、响应时间）取决于网络环境和平台限流策略，本文不提供虚构数据。

踩坑及风险边界

• API版本变化：平台可能更新接口，适配器需要及时更新。建议为每个适配器标注测试日期。
• 模型输出不稳定：同一问题可能得到不同回答，属于正常现象，采集时应保留所有版本。
• 成本控制：每次调用消耗Token，需监控每日调用量，设置预算告警。
• 权限安全：API Key应存储在环境变量或密钥管理服务中，不要硬编码。
• 数据合规：采集的内容可能涉及版权，仅用于内部监测，不对外发布原始回答。

可复用经验总结

适配器模式有效隔离了平台差异，新增平台只需添加一个适配器类，不影响现有逻辑。统一数据模型保证了下游解析和指标计算的稳定性。异常处理和重试机制提高了系统的鲁棒性。但需注意，平台接口频繁变更时，适配器的维护成本会上升，建议定期回归测试。另外，不同平台的响应时间差异可能影响调度策略，后续可考虑异步并发采集以提升效率。