multi-llm-routing

multi-llm-routing

Covers intelligent routing across multiple LLM providers: cost tiers, latency SLAs, provider health, fallback chains, and a unified client that hides provider-specific quirks.

1) Provider registry

from dataclasses import dataclass, field
from enum import Enum

class Provider(str, Enum):
    ANTHROPIC = "anthropic"
    OPENAI    = "openai"
    MISTRAL   = "mistral"
    GROQ      = "groq"

@dataclass
class ModelSpec:
    provider:      Provider
    model_id:      str
    context_window: int
    cost_in:       float    # USD per 1M input tokens
    cost_out:      float    # USD per 1M output tokens
    avg_latency_ms: float   # p50 from benchmarks
    tier:          int      # 1=heavy, 2=standard, 3=fast/cheap

MODEL_REGISTRY: dict[str, ModelSpec] = {
    "opus":    ModelSpec(Provider.ANTHROPIC, "claude-opus-4-6",       200_000, 15.0, 75.0,  4000, 1),
    "sonnet":  ModelSpec(Provider.ANTHROPIC, "claude-sonnet-4-6",     200_000,  3.0, 15.0,  1500, 2),
    "haiku":   ModelSpec(Provider.ANTHROPIC, "claude-haiku-4-5-20251001", 200_000, 0.8,  4.0,   500, 3),
    "gpt4o":   ModelSpec(Provider.OPENAI,    "gpt-4o",                128_000,  5.0, 15.0,  2000, 2),
    "gpt4o-m": ModelSpec(Provider.OPENAI,    "gpt-4o-mini",           128_000,  0.15, 0.6,   700, 3),
    "mistral-l": ModelSpec(Provider.MISTRAL, "mistral-large-latest",  128_000,  2.0,  6.0,  1800, 2),
    "llama3":  ModelSpec(Provider.GROQ,      "llama3-70b-8192",         8_192,  0.59, 0.79,  300, 3),
}

2) Routing logic

import asyncio, time
from typing import Literal

TaskType = Literal["reasoning", "extraction", "classification", "generation", "summarisation", "coding"]

TASK_TIER: dict[TaskType, int] = {
    "reasoning":      1,   # always use heavy model
    "coding":         2,   # standard
    "generation":     2,
    "summarisation":  3,   # cheap is fine
    "extraction":     3,
    "classification": 3,
}

# Per-provider health state (updated by health-check loop)
_provider_health: dict[Provider, bool] = {p: True for p in Provider}

def select_model(
    task: TaskType,
    input_tokens: int,
    budget_usd: float | None = None,
    max_latency_ms: float | None = None,
    prefer_provider: Provider | None = None,
) -> ModelSpec:
    tier = TASK_TIER[task]
    candidates = [
        spec for spec in MODEL_REGISTRY.values()
        if spec.tier >= tier
        and _provider_health[spec.provider]
        and spec.context_window >= input_tokens + 2048  # leave room for output
    ]
    if prefer_provider:
        preferred = [c for c in candidates if c.provider == prefer_provider]
        if preferred:
            candidates = preferred

    if max_latency_ms:
        candidates = [c for c in candidates if c.avg_latency_ms <= max_latency_ms] or candidates

    if budget_usd:
        candidates = [
            c for c in candidates
            if (input_tokens / 1_000_000 * c.cost_in) < budget_usd
        ] or candidates

    # Pick lowest cost from acceptable candidates at target tier
    target = [c for c in candidates if c.tier == tier] or candidates
    return min(target, key=lambda c: c.cost_in + c.cost_out)