在实际项目中,同一个大模型可能由多个平台提供。不同平台使用的 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}"
)
这段代码完成了三个动作:
- 根据
priority确定平台调用顺序; - 跳过健康检查判定为不可用的平台;
- 实际请求发生异常时,继续尝试下一个平台。
为了突出设计思路,示例直接捕获了 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 和配置,而不需要改动业务代码。