Back to Details

multi-model-routing

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Multi-Model Routing

多模型路由

Part of Agent Skills™ by googleadsagent.ai™
由 googleadsagent.ai™ 提供的 Agent Skills™ 的一部分

Description

描述

Multi-Model Routing is the intelligent dispatch of agent tasks to the optimal model provider based on task characteristics, cost constraints, latency requirements, and availability. Production AI systems that rely on a single model provider are fragile and expensive. Multi-Model Routing creates a resilient, cost-efficient agent architecture that leverages the strengths of Claude, GPT, Gemini, and open-source models, automatically selecting the best model for each task and failing over gracefully when a provider is unavailable.
This skill documents the multi-model routing architecture powering the Buddy™ agent at googleadsagent.ai™, which routes between Claude (primary — strongest reasoning), GPT-4o (secondary — strong function calling), and Gemini (tertiary — large context, low cost) based on task classification. The routing layer reduced costs by 45% compared to using Claude for all tasks while maintaining equivalent quality scores, because many subtasks (formatting, summarization, data extraction) perform identically on cheaper models.
The routing decision incorporates four factors: model strengths (code reasoning, long context, structured output, creative writing), cost per token (varies 100x between model tiers), latency targets (real-time vs. batch), and availability (rate limits, outages, degraded performance). A circuit breaker pattern ensures that temporary provider issues don't cascade into user-facing failures.
多模型路由是根据任务特征、成本约束、延迟要求和可用性，将 Agent 任务智能分派给最优模型提供商的机制。依赖单一模型提供商的生产级 AI 系统既脆弱又昂贵。多模型路由可构建具备韧性的成本高效 Agent 架构，利用 Claude、GPT、Gemini 和开源模型的优势，自动为每项任务选择最佳模型，并在提供商不可用时平滑完成故障转移。
本 skill 记录了为 googleadsagent.ai™ 上的 Buddy™ Agent 提供支持的多模型路由架构。该架构根据任务分类，在 Claude（首选——推理能力最强）、GPT-4o（次选——函数调用能力强）与 Gemini（第三选择——上下文窗口大、成本低）之间进行路由。相较于所有任务都使用 Claude，路由层在保持同等质量评分的同时将成本降低了 45%，因为许多子任务（格式化、摘要、数据提取）使用成本更低的模型时也能取得相同效果。
路由决策会综合四个因素：模型优势（代码推理、长上下文、结构化输出、创意写作）、每 token 成本（不同模型层级之间可相差 100 倍）、延迟目标（实时与批处理），以及可用性（速率限制、服务中断、性能下降）。熔断器模式可确保提供商的临时问题不会级联为用户可见的故障。

Use When

适用场景

  • Monthly AI costs need reduction without sacrificing quality
  • You need resilience against single-provider outages or rate limits
  • Different subtasks have fundamentally different model requirements
  • Latency-sensitive and latency-tolerant tasks coexist in the same system
  • You want to evaluate new models without fully committing to them
  • Compliance requires not being locked into a single AI vendor
  • 需要在不牺牲质量的前提下降低每月 AI 成本
  • 需要防范单一提供商宕机或速率限制，并增强系统韧性
  • 不同子任务对模型有本质不同的要求
  • 同一系统中同时存在延迟敏感型和可容忍延迟的任务
  • 希望在不全面绑定新模型的情况下对其进行评估
  • 合规要求需要避免被单一 AI 厂商锁定

How It Works

工作原理

mermaid
graph TD
    A[Incoming Task] --> B[Task Classifier]
    B --> C{Task Type}
    C -->|Code Reasoning| D[Claude Sonnet/Opus]
    C -->|Structured Extraction| E[GPT-4o / Claude Haiku]
    C -->|Long Context| F[Gemini Pro]
    C -->|Simple Format| G[Claude Haiku / GPT-4o-mini]
    
    D --> H{Provider Available?}
    E --> H
    F --> H
    G --> H
    
    H -->|Yes| I[Execute]
    H -->|No| J[Fallback Chain]
    J --> K[Next Provider]
    K --> H
    
    I --> L[Result + Metrics]
    L --> M[Router Learning]
    M --> B
The routing pipeline classifies each incoming task by type (code reasoning, structured extraction, long context processing, simple formatting), then maps to the optimal model provider. Before dispatch, a circuit breaker checks provider availability — if a provider has failed recently, the task is immediately routed to the fallback chain. After execution, the result quality and performance metrics feed back into the router, allowing it to refine its model-task mapping over time.
mermaid
graph TD
    A[Incoming Task] --> B[Task Classifier]
    B --> C{Task Type}
    C -->|Code Reasoning| D[Claude Sonnet/Opus]
    C -->|Structured Extraction| E[GPT-4o / Claude Haiku]
    C -->|Long Context| F[Gemini Pro]
    C -->|Simple Format| G[Claude Haiku / GPT-4o-mini]
    
    D --> H{Provider Available?}
    E --> H
    F --> H
    G --> H
    
    H -->|Yes| I[Execute]
    H -->|No| J[Fallback Chain]
    J --> K[Next Provider]
    K --> H
    
    I --> L[Result + Metrics]
    L --> M[Router Learning]
    M --> B
路由 pipeline 会按类型（代码推理、结构化提取、长上下文处理、简单格式化）对每个传入任务进行分类，然后映射到最优模型提供商。调度前，熔断器会检查提供商可用性；如果某个提供商近期发生过故障，任务会立即转入故障转移链。执行后，结果质量和性能指标会反馈给路由器，使其能够持续优化模型与任务之间的映射关系。

Implementation

实现

Model Provider Registry:
typescript
interface ModelProvider {
  id: string;
  name: string;
  models: ModelConfig[];
  circuitBreaker: CircuitBreakerState;
}

interface ModelConfig {
  id: string;
  strengths: string[];
  costPer1kInput: number;
  costPer1kOutput: number;
  maxContextTokens: number;
  avgLatencyMs: number;
}

const PROVIDERS: ModelProvider[] = [
  {
    id: "anthropic",
    name: "Anthropic",
    models: [
      { id: "claude-opus", strengths: ["reasoning", "code", "analysis"], costPer1kInput: 0.015, costPer1kOutput: 0.075, maxContextTokens: 200000, avgLatencyMs: 3000 },
      { id: "claude-sonnet", strengths: ["reasoning", "code", "balanced"], costPer1kInput: 0.003, costPer1kOutput: 0.015, maxContextTokens: 200000, avgLatencyMs: 1500 },
      { id: "claude-haiku", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00025, costPer1kOutput: 0.00125, maxContextTokens: 200000, avgLatencyMs: 500 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "openai",
    name: "OpenAI",
    models: [
      { id: "gpt-4o", strengths: ["function-calling", "structured-output", "general"], costPer1kInput: 0.005, costPer1kOutput: 0.015, maxContextTokens: 128000, avgLatencyMs: 1200 },
      { id: "gpt-4o-mini", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00015, costPer1kOutput: 0.0006, maxContextTokens: 128000, avgLatencyMs: 400 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "google",
    name: "Google",
    models: [
      { id: "gemini-pro", strengths: ["long-context", "multimodal", "general"], costPer1kInput: 0.00125, costPer1kOutput: 0.005, maxContextTokens: 1000000, avgLatencyMs: 2000 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
];
Intelligent Router:
python
class MultiModelRouter:
    TASK_MODEL_MAP = {
        "code_reasoning": ["claude-sonnet", "claude-opus", "gpt-4o"],
        "structured_extraction": ["claude-haiku", "gpt-4o-mini", "gpt-4o"],
        "long_context": ["gemini-pro", "claude-sonnet", "gpt-4o"],
        "simple_formatting": ["claude-haiku", "gpt-4o-mini", "gemini-pro"],
        "creative_writing": ["claude-opus", "claude-sonnet", "gpt-4o"],
        "analysis": ["claude-sonnet", "gpt-4o", "gemini-pro"],
    }

    def __init__(self, providers: dict, circuit_breakers: dict):
        self.providers = providers
        self.breakers = circuit_breakers

    def route(self, task_type: str, constraints: dict = None) -> str:
        candidates = self.TASK_MODEL_MAP.get(task_type, ["claude-sonnet"])
        constraints = constraints or {}

        for model_id in candidates:
            provider = self.get_provider(model_id)
            if self.breakers[provider].is_open():
                continue
            if constraints.get("max_cost") and self.estimate_cost(model_id, constraints) > constraints["max_cost"]:
                continue
            if constraints.get("max_latency_ms") and self.avg_latency(model_id) > constraints["max_latency_ms"]:
                continue
            return model_id

        return candidates[0]  # Last resort: try primary anyway

    def estimate_cost(self, model_id: str, constraints: dict) -> float:
        config = self.get_model_config(model_id)
        est_input = constraints.get("estimated_input_tokens", 1000)
        est_output = constraints.get("estimated_output_tokens", 500)
        return (est_input / 1000 * config["costPer1kInput"] +
                est_output / 1000 * config["costPer1kOutput"])
Circuit Breaker:
python
class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def is_open(self) -> bool:
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
                return False
            return True
        return False
Unified Execution Interface:
python
class UnifiedModelClient:
    def __init__(self, router: MultiModelRouter, clients: dict):
        self.router = router
        self.clients = clients

    async def generate(self, task_type: str, messages: list, **kwargs) -> dict:
        model_id = self.router.route(task_type, kwargs.get("constraints"))
        client = self.clients[self.router.get_provider(model_id)]

        try:
            result = await client.generate(model=model_id, messages=messages, **kwargs)
            self.router.breakers[self.router.get_provider(model_id)].record_success()
            return {"result": result, "model": model_id, "provider": self.router.get_provider(model_id)}
        except Exception as e:
            self.router.breakers[self.router.get_provider(model_id)].record_failure()
            fallback_model = self.router.route(task_type, {**kwargs.get("constraints", {}), "exclude": [model_id]})
            fallback_client = self.clients[self.router.get_provider(fallback_model)]
            result = await fallback_client.generate(model=fallback_model, messages=messages, **kwargs)
            return {"result": result, "model": fallback_model, "fallback": True}
模型提供商注册表：
typescript
interface ModelProvider {
  id: string;
  name: string;
  models: ModelConfig[];
  circuitBreaker: CircuitBreakerState;
}

interface ModelConfig {
  id: string;
  strengths: string[];
  costPer1kInput: number;
  costPer1kOutput: number;
  maxContextTokens: number;
  avgLatencyMs: number;
}

const PROVIDERS: ModelProvider[] = [
  {
    id: "anthropic",
    name: "Anthropic",
    models: [
      { id: "claude-opus", strengths: ["reasoning", "code", "analysis"], costPer1kInput: 0.015, costPer1kOutput: 0.075, maxContextTokens: 200000, avgLatencyMs: 3000 },
      { id: "claude-sonnet", strengths: ["reasoning", "code", "balanced"], costPer1kInput: 0.003, costPer1kOutput: 0.015, maxContextTokens: 200000, avgLatencyMs: 1500 },
      { id: "claude-haiku", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00025, costPer1kOutput: 0.00125, maxContextTokens: 200000, avgLatencyMs: 500 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "openai",
    name: "OpenAI",
    models: [
      { id: "gpt-4o", strengths: ["function-calling", "structured-output", "general"], costPer1kInput: 0.005, costPer1kOutput: 0.015, maxContextTokens: 128000, avgLatencyMs: 1200 },
      { id: "gpt-4o-mini", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00015, costPer1kOutput: 0.0006, maxContextTokens: 128000, avgLatencyMs: 400 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "google",
    name: "Google",
    models: [
      { id: "gemini-pro", strengths: ["long-context", "multimodal", "general"], costPer1kInput: 0.00125, costPer1kOutput: 0.005, maxContextTokens: 1000000, avgLatencyMs: 2000 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
];
智能路由器：
python
class MultiModelRouter:
    TASK_MODEL_MAP = {
        "code_reasoning": ["claude-sonnet", "claude-opus", "gpt-4o"],
        "structured_extraction": ["claude-haiku", "gpt-4o-mini", "gpt-4o"],
        "long_context": ["gemini-pro", "claude-sonnet", "gpt-4o"],
        "simple_formatting": ["claude-haiku", "gpt-4o-mini", "gemini-pro"],
        "creative_writing": ["claude-opus", "claude-sonnet", "gpt-4o"],
        "analysis": ["claude-sonnet", "gpt-4o", "gemini-pro"],
    }

    def __init__(self, providers: dict, circuit_breakers: dict):
        self.providers = providers
        self.breakers = circuit_breakers

    def route(self, task_type: str, constraints: dict = None) -> str:
        candidates = self.TASK_MODEL_MAP.get(task_type, ["claude-sonnet"])
        constraints = constraints or {}

        for model_id in candidates:
            provider = self.get_provider(model_id)
            if self.breakers[provider].is_open():
                continue
            if constraints.get("max_cost") and self.estimate_cost(model_id, constraints) > constraints["max_cost"]:
                continue
            if constraints.get("max_latency_ms") and self.avg_latency(model_id) > constraints["max_latency_ms"]:
                continue
            return model_id

        return candidates[0]  # Last resort: try primary anyway

    def estimate_cost(self, model_id: str, constraints: dict) -> float:
        config = self.get_model_config(model_id)
        est_input = constraints.get("estimated_input_tokens", 1000)
        est_output = constraints.get("estimated_output_tokens", 500)
        return (est_input / 1000 * config["costPer1kInput"] +
                est_output / 1000 * config["costPer1kOutput"])
熔断器：
python
class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def is_open(self) -> bool:
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
                return False
            return True
        return False
统一执行接口：
python
class UnifiedModelClient:
    def __init__(self, router: MultiModelRouter, clients: dict):
        self.router = router
        self.clients = clients

    async def generate(self, task_type: str, messages: list, **kwargs) -> dict:
        model_id = self.router.route(task_type, kwargs.get("constraints"))
        client = self.clients[self.router.get_provider(model_id)]

        try:
            result = await client.generate(model=model_id, messages=messages, **kwargs)
            self.router.breakers[self.router.get_provider(model_id)].record_success()
            return {"result": result, "model": model_id, "provider": self.router.get_provider(model_id)}
        except Exception as e:
            self.router.breakers[self.router.get_provider(model_id)].record_failure()
            fallback_model = self.router.route(task_type, {**kwargs.get("constraints", {}), "exclude": [model_id]})
            fallback_client = self.clients[self.router.get_provider(fallback_model)]
            result = await fallback_client.generate(model=fallback_model, messages=messages, **kwargs)
            return {"result": result, "model": fallback_model, "fallback": True}

Best Practices

最佳实践

  1. Map model strengths empirically — benchmark each model on your specific task types; published benchmarks rarely reflect domain-specific performance.
  2. Implement circuit breakers per provider — don't let one provider's outage cascade into retry storms; fail fast and fall over to alternatives.
  3. Track cost per successful task — raw token costs are misleading; measure total cost including retries, fallbacks, and failed attempts.
  4. Use the cheapest model that meets quality thresholds — for many tasks, a Haiku-class model produces identical results to an Opus-class model at 1/60th the cost.
  5. Normalize message formats across providers — maintain a unified message format internally and translate to provider-specific formats at the adapter layer.
  6. Log routing decisions for analysis — every routing decision should be logged with the task type, selected model, constraints, and outcome for continuous router optimization.
  7. Review fallback chains quarterly — model capabilities, pricing, and availability change frequently; update your provider registry and routing maps accordingly.
  1. 基于实测映射模型优势——针对你的具体任务类型对每个模型进行基准测试；公开基准测试很少能反映特定领域的表现。
  2. 为每个提供商实现熔断器——不要让某个提供商的故障引发连锁重试风暴；应快速失败并切换到替代方案。
  3. 跟踪每个成功任务的总成本——原始 token 成本具有误导性；应衡量包括重试、故障转移和失败尝试在内的总成本。
  4. 使用达到质量门槛的最便宜模型——对许多任务而言，Haiku 级模型仅需 Opus 级模型 1/60 的成本即可产生相同结果。
  5. 统一各提供商的消息格式——内部维持统一的消息格式，并在适配器层转换为提供商特定格式。
  6. 记录路由决策以供分析——每项路由决策都应记录任务类型、所选模型、约束条件和执行结果，以持续优化路由器。
  7. 每季度审查故障转移链——模型能力、定价和可用性变化频繁；应相应更新提供商注册表和路由映射。

Platform Compatibility

平台兼容性

FeatureClaude CodeCursorCodexGemini CLI
Multi-model selection✅ --model flag✅ Model picker⚠️ OpenAI only⚠️ Google only
Programmatic routing✅ Via MCP/API✅ Via extensions✅ Via API✅ Via API
Circuit breaker✅ Custom✅ Custom✅ Custom✅ Custom
Cost tracking✅ Custom✅ Custom✅ Custom✅ Custom
Fallback chains✅ Custom✅ Custom✅ Custom✅ Custom
功能Claude CodeCursorCodexGemini CLI
多模型选择✅ --model flag✅ 模型选择器⚠️ 仅限 OpenAI⚠️ 仅限 Google
程序化路由✅ 通过 MCP/API✅ 通过扩展✅ 通过 API✅ 通过 API
熔断器✅ 自定义✅ 自定义✅ 自定义✅ 自定义
成本追踪✅ 自定义✅ 自定义✅ 自定义✅ 自定义
故障转移链✅ 自定义✅ 自定义✅ 自定义✅ 自定义

Related Skills

相关 Skill

  • Token Optimization - Model routing is the highest-leverage token cost optimization technique
  • Parallel Agent Orchestration - Parallel subagents benefit from uniform model selection for consistent quality
  • MCP Server Creation - Model routing logic integrates into MCP server tool execution pipelines
  • Budget Optimization - Portfolio optimization principles parallel the model cost-quality tradeoff
  • Token Optimization——模型路由是优化 token 成本最有效的方法
  • Parallel Agent Orchestration——并行子 Agent 采用统一的模型选择，有助于保持一致的质量
  • MCP Server Creation——模型路由逻辑可集成到 MCP Server 的工具执行 pipeline 中
  • Budget Optimization——投资组合优化原则与模型的成本—质量权衡逻辑相似

Keywords

关键词

multi-model-routing, model-selection, cost-optimization, circuit-breaker, fallback-chains, provider-resilience, task-classification, latency-routing, load-balancing, agent-skills
© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License
multi-model-routing, model-selection, cost-optimization, circuit-breaker, fallback-chains, provider-resilience, task-classification, latency-routing, load-balancing, agent-skills
© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License