Multi-provider routing with LiteLLM

Open in Colab Run this notebook in Colab

Multi-provider routing with LiteLLM#

Track 08 - Production · Notebook 02 · Runtime: ~30s LIVE, <1s replay

Same call site, three providers (Sonnet 4.6, Haiku 4.5, GPT-4o-mini). Cost via litellm.completion_cost, fallback via litellm.Router. We classify 20 support tickets and watch the Router recover when the primary returns 529.

from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root

import json
import os
import time
from pathlib import Path

s = bootstrap("08_production_02_litellm_router_fallbacks")

LIVE = bool(os.environ.get("ANTHROPIC_API_KEY") and os.environ.get("OPENAI_API_KEY"))
FIXTURE = json.loads((Path(repo_root()) / "notebooks/08_production/_fixtures/02_litellm.json").read_text())
TICKETS, TRUTH, LABELS = FIXTURE["tickets"], FIXTURE["ground_truth"], FIXTURE["labels"]
print(f"mode={'LIVE' if LIVE else 'REPLAY'}  tickets={len(TICKETS)}  labels={LABELS}")

Workload — 20 support tickets#

Each ticket gets one of three labels. The prompt is short on purpose: output cost is dominated by the 3-token label, so we are really comparing per-call input pricing across providers.

SYSTEM = (
    "Classify the support ticket as exactly one of: urgent, normal, spam. "
    "Reply with the single word and nothing else."
)
for t, y in zip(TICKETS[:3], TRUTH[:3]):
    print(f"  [{y:>6}] {t[:70]}{'...' if len(t) > 70 else ''}")

One call site, three providers#

litellm.completion accepts the OpenAI message shape and dispatches by the model string prefix (anthropic/..., openai/..., ollama/...). completion_cost(response) returns USD using the provider’s published rates.

def call(strategy_name: str, ticket_idx: int) -> dict:
    """LIVE: real litellm.completion. REPLAY: recorded label/cost/latency."""
    if not LIVE:
        return FIXTURE[strategy_name]["results"][ticket_idx]

    import litellm  # noqa: PLC0415
    model = FIXTURE[strategy_name]["model"]
    provider = "anthropic" if model.startswith("claude") else "openai"
    full_model = f"{provider}/{model}"

    t0 = time.perf_counter()
    resp = litellm.completion(
        model=full_model,
        messages=[{"role": "system", "content": SYSTEM},
                  {"role": "user",   "content": TICKETS[ticket_idx]}],
        max_tokens=8,
        temperature=0,
    )
    latency = time.perf_counter() - t0
    label = resp.choices[0].message.content.strip().lower()
    return {
        "label": label if label in LABELS else "normal",
        "input_tokens": resp.usage.prompt_tokens,
        "output_tokens": resp.usage.completion_tokens,
        "cost_usd": litellm.completion_cost(completion_response=resp),
        "latency_s": latency,
    }

Run all three strategies#

STRATEGIES = ["sonnet_only", "haiku_only", "gpt4o_mini"]
runs = {name: [call(name, i) for i in range(len(TICKETS))] for name in STRATEGIES}

def acc(rows): return sum(r["label"] == y for r, y in zip(rows, TRUTH)) / len(TRUTH)
def cost(rows): return sum(r["cost_usd"] for r in rows)
def lat(rows): return sum(r["latency_s"] for r in rows)

print(f"{'strategy':<14} {'acc':>6} {'$ total':>10} {'lat (s)':>9}")
for name in STRATEGIES:
    print(f"{name:<14} {acc(runs[name]):>6.0%} ${cost(runs[name]):>8.5f}  {lat(runs[name]):>8.2f}")

Fallback when the primary is overloaded#

litellm.Router accepts a model_list plus a fallbacks map. When the primary returns a retryable error (429/529/timeout), the router retries once on the same model, then falls back. We replay a recorded run where calls 4, 11, 17 hit a primary 529 and recovered via OpenAI.

def call_router(ticket_idx: int) -> dict:
    if not LIVE:
        return FIXTURE["fallback_run"]["results"][ticket_idx]

    from litellm import Router  # noqa: PLC0415
    router = Router(
        model_list=[
            {"model_name": "primary",  "litellm_params": {"model": "anthropic/claude-sonnet-4-6"}},
            {"model_name": "fallback", "litellm_params": {"model": "openai/gpt-4o-mini"}},
        ],
        fallbacks=[{"primary": ["fallback"]}],
        num_retries=1,
    )
    t0 = time.perf_counter()
    resp = router.completion(
        model="primary",
        messages=[{"role": "system", "content": SYSTEM},
                  {"role": "user",   "content": TICKETS[ticket_idx]}],
        max_tokens=8, temperature=0,
    )
    latency = time.perf_counter() - t0
    used = "fallback" if "gpt" in resp.model else "primary"
    return {
        "label": resp.choices[0].message.content.strip().lower(),
        "provider": used,
        "cost_usd": __import__("litellm").completion_cost(completion_response=resp),
        "latency_s": latency,
    }


fallback_runs = [call_router(i) for i in range(len(TICKETS))]

n_fallback = sum(1 for r in fallback_runs if r["provider"] == "fallback")
fallback_acc = sum(r["label"] == y for r, y in zip(fallback_runs, TRUTH)) / len(TRUTH)
print(f"fallback events: {n_fallback}/{len(TICKETS)}    accuracy: {fallback_acc:.0%}")
for i, r in enumerate(fallback_runs):
    if r["provider"] == "fallback":
        print(f"  call {i:>2}  fallback  {r['latency_s']:.2f}s  ${r['cost_usd']:.6f}")

Cost vs accuracy across the four strategies#

import matplotlib.pyplot as plt

names = STRATEGIES + ["router+fb"]
costs = [cost(runs[s]) for s in STRATEGIES] + [sum(r["cost_usd"] for r in fallback_runs)]
accs  = [acc(runs[s])  for s in STRATEGIES] + [fallback_acc]
lats  = [lat(runs[s])  for s in STRATEGIES] + [sum(r["latency_s"] for r in fallback_runs)]

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9.5, 3.4))
colors = ["tab:blue", "tab:green", "tab:orange", "tab:purple"]

ax1.bar(names, [c * 100 for c in costs], color=colors)
ax1.set_ylabel("total cost (cents)")
ax1.set_title(f"{len(TICKETS)} tickets · 4 strategies")
for i, v in enumerate(costs):
    ax1.text(i, v * 100, f"{v*100:.2f}c", ha="center", va="bottom", fontsize=9)

ax2.scatter([c * 100 for c in costs], [a * 100 for a in accs], s=80, c=colors)
for i, n in enumerate(names):
    ax2.annotate(n, (costs[i] * 100, accs[i] * 100), fontsize=8,
                 xytext=(4, -2), textcoords="offset points")
ax2.set_xlabel("cost (cents)"); ax2.set_ylabel("accuracy (%)")
ax2.set_title("cost vs accuracy"); ax2.grid(alpha=0.3); ax2.set_xscale("log")

fig.tight_layout(); plt.show()

Checks#

s.check(
    "all_strategies_above_85pct_accuracy",
    lambda: all(acc(runs[name]) >= 0.85 for name in STRATEGIES),
    msg=f"acc = {{n: round(acc(runs[n]), 3) for n in STRATEGIES}}",
)
s.check(
    "haiku_at_least_3x_cheaper_than_sonnet",
    lambda: cost(runs["haiku_only"]) <= cost(runs["sonnet_only"]) / 3,
    msg=f"haiku=${cost(runs['haiku_only']):.5f} sonnet=${cost(runs['sonnet_only']):.5f}",
)
s.check(
    "gpt4o_mini_cheapest",
    lambda: cost(runs["gpt4o_mini"]) < min(cost(runs["sonnet_only"]), cost(runs["haiku_only"])),
    msg=f"costs = {[round(cost(runs[n]), 6) for n in STRATEGIES]}",
)
s.check(
    "router_recovered_all_failures",
    lambda: len(fallback_runs) == len(TICKETS) and fallback_acc >= 0.85,
    msg=f"fallback events = {n_fallback}, acc = {fallback_acc:.2%}",
)
s.check(
    "fallback_calls_have_higher_latency_than_primary_median",
    lambda: max(r["latency_s"] for r in fallback_runs if r["provider"] == "fallback")
            > sorted(r["latency_s"] for r in fallback_runs if r["provider"] == "primary")[len([r for r in fallback_runs if r["provider"] == "primary"]) // 2],
    msg="fallback should pay retry-then-failover latency",
)

Notes for production#

  • Retry budget belongs in the Router, not your call site. num_retries, timeout, cooldown_time, and the fallbacks map cover ~all the failure modes you care about (429, 529, 5xx, network). Calling sites should be oblivious.

  • Pin model strings to provider prefix. anthropic/claude-sonnet-4-6 is unambiguous; bare claude-sonnet-4-6 works but breaks when LiteLLM updates its default mapping.

  • completion_cost reads provider pricing snapshots. When a provider changes prices, update litellm or pass custom_cost_per_token so your budget alerts stay correct.

  • Routing strategy (Router(routing_strategy=...)): simple-shuffle for load-balancing identical models; least-busy if you have rate-limit headroom info; latency-based to pick the historically-fastest replica.

  • Observability: litellm.callbacks = ["langfuse"] (or Helicone, Lunary, OpenTelemetry) traces every call to your dashboard with cost and latency tagged.

s.summary()
s.save()