Project: a model gateway for a support team
Lesson 0 promised a gateway with cheap and careful models, fallbacks, caching and cost tracking. Here it serves a support desk app whose cheap provider is rate limited.
The gateway
model_list:
- model_name: quick
litellm_params:
model: flaky/small
- model_name: careful
litellm_params:
model: shop/large
litellm_settings:
custom_provider_map:
- {"provider": "shop", "custom_handler": flaky_llm.shop}
- {"provider": "flaky", "custom_handler": flaky_llm.flaky}
cache: true
cache_params:
type: local
router_settings:
num_retries: 3
fallbacks: [{"quick": ["careful"]}]
general_settings:
master_key: sk-gateway-1234quick is the cheap model, and it is the flaky provider from lesson 7, refusing its first two calls. num_retries: 3 rides out the rate limit (lesson 9); if quick still fails, the fallback sends the request to careful (lesson 11). Identical requests come from the cache (lesson 13), and only the master key gets in (lesson 17).
The desk
import json
from openai import OpenAI
client = OpenAI(base_url="http://127.0.0.1:4000", api_key="sk-gateway-1234")
TICKETS = [
"I was charged twice for one order",
"My parcel has not arrived",
"How do I change my password?",
"I was charged twice for one order",
]Four tickets, the last a repeat of the first, as happens all day on a real desk.
def ask(group, text):
raw = client.chat.completions.with_raw_response.create(
model=group, messages=[{"role": "user", "content": text}]
)
if "x-litellm-cache-key" in raw.headers:
how = "cache"
else:
how = f"{raw.headers['x-litellm-model-group']}, retries {raw.headers['x-litellm-attempted-retries']}"
return raw.parse().choices[0].message.content, how
def sort_ticket(text):
answer, how = ask("quick", text)
if not answer.startswith("{"):
answer, how = ask("careful", text)
how = "escalated to " + how
return answer, howThe escalation from lesson 14, now against gateway groups instead of an in-process Router. ask reads the headers from lesson 18 to report how each answer was served.
for text in TICKETS:
answer, how = sort_ticket(text)
print(f"{text} -> {answer} ({how})")Run it
litellm --config config.yaml --port 4000 > gateway.log 2>&1 &
until curl -s localhost:4000/health/liveliness > /dev/null; do sleep 1; done
python desk.py
kill %1Read it line by line. The first ticket was answered by quick after the gateway retried twice through the rate limit; the desk never saw a 429. The second went straight to quick. The password ticket got no usable answer from quick and escalated to careful. The repeated ticket came from the cache, without a model call.
Testing the routing rules
The gateway's rules can be tested without starting it, with the same Router in Python:
import litellm
from litellm import Router
import flaky_llm
litellm.suppress_debug_info = True
def make_router():
return Router(
model_list=[
{"model_name": "quick", "litellm_params": {"model": "down/large"}},
{"model_name": "careful", "litellm_params": {"model": "shop/small"}},
],
fallbacks=[{"quick": ["careful"]}],
num_retries=0,
)
def test_outage_falls_back_to_careful():
response = make_router().completion(model="quick", messages=[{"role": "user", "content": "My parcel has not arrived"}])
assert response.choices[0].message.content == '{"category": "shipping", "priority": 3}'
def test_no_fallback_raises():
router = Router(model_list=[{"model_name": "quick", "litellm_params": {"model": "down/large"}}], num_retries=0)
try:
router.completion(model="quick", messages=[{"role": "user", "content": "hi"}])
except litellm.ServiceUnavailableError:
return
raise AssertionError("expected ServiceUnavailableError")Building a Router makes two of LiteLLM's dependencies, Pydantic and backoff, print warnings about their own code under pytest. They say nothing about your tests, so a pytest.ini leaves those two out:
[pytest]
filterwarnings =
ignore::UserWarning:pydantic._internal._generate_schema
ignore::DeprecationWarning:backoffpytest -qWhere each piece came from
Things to add
- Replace
shop/largewith a real model, such asanthropic/claude-sonnet-4-5withapi_key: os.environ/ANTHROPIC_API_KEY, and keep the stand-in asquick. - Add a Router test for cooldowns, from lesson 12.
- Point an agent from the MCP or OpenAI Agents SDK course at the gateway through its OpenAI-compatible model setting.
What this course left out
| Topic | What it is for |
|---|---|
| Virtual keys and budgets | Per-team keys with model limits, spend limits and resets; needs PostgreSQL. |
| Admin UI | A web interface for keys, models, spend and logs. |
| Redis | Shared cache, cooldowns and rate limits across several gateway workers. |
| Timeouts | request_timeout and per-deployment timeouts, so a hung provider fails fast. |
| Tool calling and JSON mode | One format for function calling and structured output across providers. |
| Routing strategies | Latency-based, usage-based and cost-based routing instead of simple-shuffle. |
| Observability integrations | Sending every call to Langfuse, Datadog, OpenTelemetry and others. |
| Guardrails on the gateway | Checks on requests and responses applied centrally. |
| Deployment | Docker images, Kubernetes and health checks for a production gateway. |
This is what real progress feels like.