TokenMix Team · 2026-03-21

用一个 API 构建多模型 AI 应用

用一个 API 构建多模型 AI 应用

大多数生产环境的 AI 系统只用一个模型。这在大部分时间没问题，直到出问题的那天：模型宕机、新版本在你的场景上效果变差、或者你发现另一个模型处理某些查询更好。多模型架构解决这些问题，还能解锁单模型做不到的优化策略。

这篇文章介绍四种生产级的多模型架构模式，全部通过 TokenMix API 作为统一接入层。

模式一：智能回退链

最基本的多模型模式，也是每个生产系统都应该有的：主模型挂了，自动切到备选模型，而不是返回错误。

import openai

client = openai.OpenAI(
    base_url="https://api.tokenmix.ai/v1",
    api_key="your-tokenmix-api-key",
    timeout=30.0
)

class FallbackChain:
    def __init__(self, models: list[str]):
        self.models = models

    def complete(self, messages: list, **kwargs) -> tuple[str, str]:
        """按顺序尝试每个模型，返回 (响应, 使用的模型)"""
        last_error = None
        for model in self.models:
            try:
                response = client.chat.completions.create(
                    model=model, messages=messages, **kwargs
                )
                return response.choices[0].message.content, model
            except (openai.RateLimitError, openai.APITimeoutError, openai.APIError) as e:
                last_error = f"{model}: {e}"
                continue
        raise RuntimeError(f"所有模型都失败了。最后的错误: {last_error}")

# 使用
chain = FallbackChain([
    "claude-sonnet-4",     # 首选：质量最好
    "gpt-4o",              # 备选1：不同厂商
    "gemini-2.0-flash"     # 备选2：快且稳定
])
result, model_used = chain.complete(
    messages=[{"role": "user", "content": "分析这段错误日志..."}]
)

几个关键设计点：

模式二：A/B 测试

在切换模型之前，先用真实流量对比效果。把一部分流量导到候选模型，收集质量数据。

import random
import time
from dataclasses import dataclass, field

@dataclass
class ABTest:
    control_model: str
    challenger_model: str
    challenger_pct: float = 0.1  # 10% 流量给候选模型
    results: dict = field(default_factory=lambda: {"control": [], "challenger": []})

    def route(self) -> str:
        return self.challenger_model if random.random() < self.challenger_pct else self.control_model

    def complete_and_record(self, messages: list, **kwargs):
        model = self.route()
        start = time.time()
        response = client.chat.completions.create(model=model, messages=messages, **kwargs)
        latency = (time.time() - start) * 1000

        variant = "challenger" if model == self.challenger_model else "control"
        self.results[variant].append({
            "model": model,
            "latency_ms": latency,
            "tokens": response.usage.completion_tokens
        })
        return response.choices[0].message.content

test = ABTest(control_model="gpt-4o", challenger_model="claude-sonnet-4", challenger_pct=0.2)

光看延迟不够。建议把响应日志下来，定期用 judge 模型或人工评审来评估质量差异。

模式三：质量评分与自动选择

对于输出质量差异大的任务，同时调用多个模型，用一个轻量 judge 打分，选最好的。

from concurrent.futures import ThreadPoolExecutor

def generate_candidates(messages, models, **kwargs):
    """并行调用多个模型"""
    def call(model):
        start = time.time()
        try:
            resp = client.chat.completions.create(model=model, messages=messages, **kwargs)
            return {
                "model": model,
                "response": resp.choices[0].message.content,
                "latency_ms": (time.time() - start) * 1000
            }
        except Exception as e:
            return {"model": model, "error": str(e)}

    with ThreadPoolExecutor(max_workers=len(models)) as pool:
        return [r for r in pool.map(call, models) if "error" not in r]

def score_response(prompt, response):
    """用便宜的模型打分"""
    judge = client.chat.completions.create(
        model="gemini-2.0-flash",
        messages=[
            {"role": "system", "content": "给这个回复打 1-10 分。考虑准确性、完整性和简洁性。只回复数字。"},
            {"role": "user", "content": f"问题: {prompt}\n\n回复: {response}"}
        ],
        max_tokens=5, temperature=0
    )
    try:
        return float(judge.choices[0].message.content.strip())
    except ValueError:
        return 5.0

def best_of_n(messages, models, **kwargs):
    candidates = generate_candidates(messages, models, **kwargs)
    if not candidates:
        raise RuntimeError("所有模型都失败了")
    prompt = messages[-1]["content"]
    for c in candidates:
        c["score"] = score_response(prompt, c["response"])
    return max(candidates, key=lambda c: c["score"])

这个模式每次请求的成本是 3 倍以上（多个模型 + judge），所以只在高价值输出时使用。

模式四：熔断器降级

生产环境需要比简单回退更强的机制。熔断器模式会检测模型是否持续失败，临时将它从轮换中移除，防止级联故障。

from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"       # 正常
    OPEN = "open"           # 已熔断
    HALF_OPEN = "half_open" # 恢复测试中

class CircuitBreaker:
    def __init__(self, model, failure_threshold=5, recovery_timeout=60):
        self.model = model
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0

    def can_execute(self):
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        return True

    def record_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

把熔断器和回退链结合起来，就是一个真正的生产级多模型系统。

模式选择指南

模式 适用场景 成本影响 复杂度
回退链 所有生产系统 无（只为成功的调用付费） 低
A/B 测试 模型迁移前 低 中
质量评分 高价值输出 高（3倍以上） 中
熔断器 高可用系统 无 中

建议从回退链开始，这是投入产出比最高的模式。然后加熔断器保证可用性。模型切换前用 A/B 测试。质量评分留给最重要的输出。

为什么统一 API 是关键

上面四种模式都依赖一个前提：用同样的代码调用不同的模型。没有统一 API，每种模式都意味着管理多个 SDK、多个 Key、多套鉴权和不同的响应格式。TokenMix 把这些复杂性全部消除了，所有模型通过一个 OpenAI 兼容的端点访问。

这不只是方便 -- 这是架构层面的使能器。当切换模型只需要改一行代码而不是一周的集成工作时，多模型架构才真正可行。

上线前检查清单