$ cat ~/articles/117 _

用适配器模式与工厂模式实现多平台 LLM 调用和故障切换

作者:jaifire 2026-07-18 22:42 0 阅读

在实际项目中,同一个大模型可能由多个平台提供。不同平台使用的 SDK、接口地址、认证方式、模型名称和异常类型往往并不一致。

例如,业务层只想调用逻辑模型 glm-5.2,但它背后可能有多个可选端点:

glm-5.2
├── 智谱平台:glm-5.2
├── 小米平台:glm-5.2
└── 其他平台:glm-5.2

如果业务代码直接依赖每个平台的 SDK,不仅会产生大量分支判断,切换平台时也需要修改业务代码。除此之外,当主平台出现额度不足、限流、超时或服务故障时,系统还需要自动尝试备用平台。

本文使用一个简化示例,介绍如何通过适配器模式、工厂模式和故障转移对象解决这个问题。

一、设计目标

我们希望业务代码始终保持一致:

llm = LLMFactory.create("glm-5.2")
result = llm.chat(messages)

业务层不需要关心:

  • 当前请求最终由哪个平台处理;
  • 平台使用什么 SDK 和认证方式;
  • 平台要求的真实模型名称;
  • 主平台失败后应该切换到哪个备用平台。

整体结构如下:

业务代码
   ↓
LLMFactory
   ↓
FailoverLLM
   ├── Provider A Adapter
   ├── Provider B Adapter
   └── Provider C Adapter

其中:

  • Adapter 负责屏蔽不同平台的接口差异;
  • Factory 负责根据配置创建统一的 LLM 对象;
  • FailoverLLM 负责按优先级调用平台,并在失败后尝试备用平台;
  • 配置层 负责描述平台信息以及逻辑模型和物理模型之间的映射。

二、拆分平台配置与模型端点

平台配置保存认证信息、接口地址和使用的 Adapter:

@dataclass(frozen=True)
class ProviderConfig:
    name: str
    adapter: str
    api_key: str
    base_url: str | None = None

模型端点表示一个逻辑模型在某个平台上的具体实现:

@dataclass(frozen=True)
class ModelEndpoint:
    provider: str
    remote_model: str
    priority: int = 100
    timeout: float = 60
    max_retries: int = 1
    extra: dict[str, Any] = field(default_factory=dict)

这样,平台信息和模型信息不会混在一起。同一个 Provider 可以提供多个模型,同一个逻辑模型也可以配置多个 Provider。

例如:

MODEL_CFG = {
    "glm-5.2": [
        ModelEndpoint(
            provider="zhipu",
            remote_model="glm-5.2",
            priority=10,
        ),
        ModelEndpoint(
            provider="xiaomi",
            remote_model="glm-5.2",
            priority=20,
        ),
        ModelEndpoint(
            provider="deepseek",
            remote_model="glm-5.2",
            priority=30,
        ),
    ]
}

业务代码只使用 glm-5.2 这个逻辑名称。每个平台真正需要的模型名称由 remote_model 决定。

三、使用 Adapter 屏蔽平台差异

所有平台 Adapter 都实现相同接口:

class BaseLLMAdapter(ABC):
    def __init__(self, provider, model_config):
        self.provider = provider
        self.model_config = model_config

    @abstractmethod
    def chat(self, messages: list[dict], **kwargs) -> str:
        pass

兼容 OpenAI 接口的平台可以共用一个 Adapter:

class OpenAICompatibleAdapter(BaseLLMAdapter):
    def chat(self, messages, **kwargs):
        # 使用 provider.base_url 和 provider.api_key 创建客户端
        # 使用 model_config.remote_model 调用真实模型
        return "openai compatible response"

SDK 完全不同的平台则使用独立 Adapter:

class CustomProviderAdapter(BaseLLMAdapter):
    def chat(self, messages, **kwargs):
        # 将统一消息转换成平台格式
        # 调用平台专用 SDK
        # 将响应转换成统一结果
        return "custom provider response"

再通过注册表建立 Adapter 类型和实现类之间的映射:

ADAPTER_REGISTRY = {
    AdapterType.OPENAI: OpenAICompatibleAdapter,
    AdapterType.CUSTOM: CustomProviderAdapter,
}

增加新平台时,只需要实现或复用一个 Adapter,并增加相应配置,业务代码不需要修改。

四、使用 FailoverLLM 完成自动切换

仅仅在创建对象时选择一个当前可用的平台还不够。即使平台在创建时是可用的,真正发起请求时仍然可能出现限流、超时或服务故障。

因此,Factory 不直接返回某个平台的 Adapter,而是返回一个持有全部候选端点的 FailoverLLM

class FailoverLLM:
    """按优先级调用提供商,失败时自动尝试下一个。"""

    def __init__(self, model_name, endpoints):
        self.model_name = model_name
        self.endpoints = sorted(
            endpoints,
            key=lambda item: item.priority,
        )

    def chat(self, messages, **kwargs):
        errors = []

        for endpoint in self.endpoints:
            provider = PROVIDER_CFG[endpoint.provider]

            if not LLMFactory.provider_available(provider):
                continue

            adapter_class = ADAPTER_REGISTRY[provider.adapter]
            adapter = adapter_class(provider, endpoint)

            try:
                return adapter.chat(messages, **kwargs)
            except Exception as exc:
                errors.append(f"{provider.name}: {exc}")

        detail = "; ".join(errors) or "没有可用的提供商"
        raise RuntimeError(
            f"模型 {self.model_name} 的所有提供商均调用失败:{detail}"
        )

这段代码完成了三个动作:

  1. 根据 priority 确定平台调用顺序;
  2. 跳过健康检查判定为不可用的平台;
  3. 实际请求发生异常时,继续尝试下一个平台。

为了突出设计思路,示例直接捕获了 Exception。实际项目中,可以由各平台 Adapter 将 SDK 异常转换为统一异常,并且只对额度不足、限流、超时和服务不可用等异常执行切换;参数错误通常应直接返回,不必尝试所有平台。

五、Factory 只负责组装对象

Factory 的代码因此变得非常简单:

class LLMFactory:
    @staticmethod
    def provider_available(provider_config):
        """实际项目中可以从 Redis 读取平台健康状态。"""
        return True

    @staticmethod
    def create(model_name: str) -> FailoverLLM:
        endpoints = MODEL_CFG.get(model_name)

        if not endpoints:
            raise ValueError(f"Model {model_name} not found")

        return FailoverLLM(model_name, endpoints)

provider_available() 可以读取异步监控进程写入 Redis 的健康状态。这样,系统既可以提前跳过已经确认故障的平台,也可以在真实请求失败时立即切换备用平台。

两种机制分别解决不同问题:

异步健康检查 + Redis
    → 主动避开已知不可用的平台

FailoverLLM 捕获请求异常
    → 处理调用期间发生的瞬时故障

六、最终调用方式

完成这些封装后,业务代码不再包含任何平台判断:

from llm_factory import LLMFactory

llm = LLMFactory.create("glm-5.2")

result = llm.chat([
    {"role": "user", "content": "请解释一下适配器模式"}
])

print(result)

无论最终使用智谱、小米还是其他平台,业务层的调用方式都保持不变。

总结

这个示例组合了三种常见的设计思想:

  • 适配器模式:统一不同平台的 SDK、认证方式、请求格式和响应结构;
  • 工厂模式:根据逻辑模型配置创建统一调用对象;
  • 责任链与故障转移:主平台调用失败后,按照优先级继续尝试备用平台。

它的核心价值不是减少几行调用代码,而是把“平台差异”“对象创建”和“故障切换”从业务逻辑中分离出来。以后新增模型、替换供应商或者调整平台优先级时,通常只需要修改 Adapter 和配置,而不需要改动业务代码。