便宜稳定的国产模型 API 中转站怎么选：价格透明、0.2x 目录倍率与生产容灾回退实战

import os
import time
import random
from typing import List, Dict, Any, Generator
from openai import OpenAI, APIError, RateLimitError, APITimeoutError

class RobustModelGateway:
    def __init__(self, api_key: str, base_url: str = "https://api.gpt345.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url, timeout=25.0)
        # 主模型与备用降级模型定义
        self.primary_model = "deepseek-v4"
        self.fallback_model = "glm-5.3"

    def chat_completion(self, messages: List[Dict[str, str]], max_retries: int = 3, **kwargs) -> str:
        """带双模型主备降级与指数退避的聊天补全"""
        current_model = self.primary_model
        model_switched = False

        for attempt in range(1, max_retries + 1):
            try:
                # 发起模型请求
                response = self.client.chat.completions.create(
                    model=current_model,
                    messages=messages,
                    **kwargs
                )
                if model_switched:
                    print(f"[WARN] 请求通过备用模型 {current_model} 成功响应")
                return response.choices[0].message.content

            except (RateLimitError, APITimeoutError, APIError) as e:
                status_code = getattr(e, "status_code", None)
                print(f"[!] 尝试 {attempt}/{max_retries} 调用 {current_model} 异常 (HTTP {status_code}): {e}")

                # 若为主模型连续故障或出现严重错误，切换至备用模型
                if not model_switched and attempt >= 2:
                    print(f"[CIRCUIT BREAKER] 触发熔断降级：从 {self.primary_model} 切换至 {self.fallback_model}")
                    current_model = self.fallback_model
                    model_switched = True

                if attempt == max_retries:
                    raise RuntimeError(f"所有重试均告失败，最终错误: {e}")

                # 计算指数退避时间 + 随机扰动 (Jitter)
                backoff_time = (2 ** attempt) + random.uniform(0.1, 0.8)
                print(f"[*] 等待 {backoff_time:.2f} 秒后重试...")
                time.sleep(backoff_time)

    def stream_completion(self, messages: List[Dict[str, str]], **kwargs) -> Generator[str, None, None]:
        """流式输出，主模型故障自动切换备用模型流"""
        try:
            stream = self.client.chat.completions.create(
                model=self.primary_model,
                messages=messages,
                stream=True,
                **kwargs
            )
            for chunk in stream:
                content = chunk.choices[0].delta.content
                if content:
                    yield content
        except Exception as e:
            print(f"[!] 主模型流式中断: {e}，立即切换备用模型流式接力...")
            stream_fallback = self.client.chat.completions.create(
                model=self.fallback_model,
                messages=messages,
                stream=True,
                **kwargs
            )
            for chunk in stream_fallback:
                content = chunk.choices[0].delta.content
                if content:
                    yield content

if __name__ == "__main__":
    gateway = RobustModelGateway(api_key=os.getenv("GPT345_API_KEY", "sk-your-key"))
    
    test_msgs = [
        {"role": "system", "content": "你是一位资深架构师，请给出精炼回答。"},
        {"role": "user", "content": "请对比 DeepSeek V4 与 GLM-5.3 在高并发生产环境下的选型策略。"}
    ]
    
    print("[*] 正在通过高可用网关请求...")
    try:
        reply = gateway.chat_completion(messages=test_msgs, temperature=0.3)
        print("\n--- 模型回答 ---")
        print(reply)
    except Exception as err:
        print(f"[ERROR] 最终业务失败: {err}")