multi-model-routing

多模型路由

Agent Skills™ 的一部分，由 googleadsagent.ai™ 提供

描述

多模型路由是指根据任务特征、成本限制、延迟要求和可用性，将 Agent 任务智能分派给最合适的模型提供商。依赖单一模型提供商的生产级 AI 系统既脆弱又昂贵。多模型路由打造了一种具有韧性且成本高效的 Agent 架构，可发挥 Claude、GPT、Gemini 和开源模型各自的优势，自动为每项任务选择最佳模型，并在提供商不可用时平稳故障转移。

本文档介绍了 Buddy™ Agent 所采用的多模型路由架构，该架构由 googleadsagent.ai™ 提供支持。它会根据任务分类，在 Claude（主模型——推理能力最强）、GPT-4o（次级模型——函数调用能力强）和 Gemini（第三级模型——上下文容量大、成本低）之间进行路由。与所有任务都使用 Claude 相比，该路由层将成本降低了 45%，同时保持了相当的质量评分，因为许多子任务（格式整理、摘要、数据提取）使用更便宜的模型也能取得相同效果。

路由决策会考虑四个因素：模型优势（代码推理、长上下文、结构化输出、创意写作）、每个 Token 的成本（不同模型档位间相差可达 100 倍）、延迟目标（实时或批处理）以及可用性（速率限制、服务中断、性能下降）。断路器模式可确保提供商的临时问题不会演变为面向用户的故障。

适用场景

• 需要在不牺牲质量的情况下降低每月 AI 成本
• 需要应对单一提供商服务中断或速率限制的韧性
• 不同子任务对模型有本质不同的要求
• 同一系统中同时存在对延迟敏感和对延迟不敏感的任务
• 想要评估新模型，但不想完全依赖它们
• 合规要求不能被单一 AI 供应商锁定

工作原理

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

路由流水线会按类型对每项传入任务进行分类（代码推理、结构化提取、长上下文处理、简单格式整理），然后将其映射到最合适的模型提供商。分派前，断路器会检查提供商是否可用；如果某个提供商近期发生过故障，任务会立即被路由到备用链。执行完成后，结果质量和性能指标会反馈给路由器，使其能够随着时间推移不断优化模型与任务的映射关系。

实现

模型提供商注册表：

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" },
  },
];

智能路由器：

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"])

断路器：

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

统一执行接口：

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}

最佳实践

1. 通过实测了解模型优势——针对你的具体任务类型对每个模型进行基准测试；已发布的基准测试很少能反映特定领域的表现。
2. 为每个提供商实现断路器——不要让一个提供商的服务中断引发重试风暴；快速失败并切换到其他选项。
3. 跟踪每项成功任务的成本——原始 Token 成本具有误导性；衡量总成本时应包含重试、故障转移和失败尝试。
4. 使用满足质量阈值的最低价模型——对于许多任务，Haiku 级别的模型能够产出与 Opus 级别模型相同的结果，成本却只有其 1/60。
5. 统一各提供商的消息格式——在内部维护统一的消息格式，并在适配器层转换为各提供商专用的格式。
6. 记录路由决策以供分析——每项路由决策都应记录任务类型、所选模型、约束和结果，以便持续优化路由器。
7. 每季度检查备用链——模型能力、定价和可用性经常变化；应据此更新提供商注册表和路由映射。

平台兼容性

功能	Claude Code	Cursor	Codex	Gemini CLI
多模型选择	✅ --model 标志	✅ 模型选择器	⚠️ 仅支持 OpenAI	⚠️ 仅支持 Google
程序化路由	✅ 通过 MCP/API	✅ 通过扩展	✅ 通过 API	✅ 通过 API
熔断器	✅ 自定义	✅ 自定义	✅ 自定义	✅ 自定义
成本跟踪	✅ 自定义	✅ 自定义	✅ 自定义	✅ 自定义
回退链	✅ 自定义	✅ 自定义	✅ 自定义	✅ 自定义

相关技能

• 令牌优化 - 模型路由是优化令牌成本最具杠杆效应的技术
• 并行 Agent 编排 - 并行子 Agent 使用统一的模型选择，有助于保持质量一致
• MCP 服务创建 - 模型路由逻辑集成到 MCP 服务工具执行流水线中
• 预算优化 - 投资组合优化原则与模型成本和质量之间的权衡相似

关键词

多模型路由、模型选择、成本优化、熔断器、回退链、提供商弹性、任务分类、延迟路由、负载均衡、Agent 技能

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License