Ep 84: AI Gateway Architecture: One API to Rule Them Al

You're using OpenAI for GPT-4. Claude for coding. Gemini for long context. A self-hosted Llama for cost-sensitive tasks.

Four different SDKs. Four different auth methods. Four different error formats. Four different rate limits to track.

And when OpenAI has an outage? Your app goes down.

An AI Gateway fixes this. It's a unified API layer that sits between your app and multiple AI providers. Route intelligently, fail gracefully, optimize costs automatically.

What Is an AI Gateway?

Think of it like a reverse proxy, but for AI APIs:

Your App → AI Gateway → OpenAI (primary)

example
code
↓
                    → Anthropic (fallback)
                    ↓
                    → Self-hosted (cost optimization)

Core Responsibilities

Function Description

Routing Send requests to the right model/provider

Fallbacks Retry with backup providers on failure

Load balancing Distribute across multiple keys/instances

Rate limiting Prevent quota exhaustion

Caching	Semantic + exact match caching
Cost tracking	Unified billing across providers
Observability	Single pane for all AI calls

⚡

One interface, any provider

snippet
code
response = litellm.completion(
model="gpt-4",  # or "claude-3-sonnet-20240229" or "gemini-pro"
    messages=[{"role": "user", "content": "Hello!"}]

)

Response format is identical regardless of provider

Key features:

100+ provider integrations

Unified response format

Automatic retries and fallbacks

Built-in cost tracking

OpenRouter (Managed)

Hosted gateway with a single API key for all providers.

python

import openai

Point OpenAI SDK at OpenRouter

snippet
code
openai.api_base = "https://openrouter.ai/api/v1"

openai.api_key = OPENROUTER_API_KEY

Access any model

snippet
code
response = openlitellm.chat.completions.create(
model="anthropic/claude-3.5-sonnet",  # Provider/model format
    messages=[{"role": "user", "content": "Hello!"}]

)

Pricing: Markup on provider costs, but no infrastructure to run.

Portkey (Managed, Enterprise)

Production-focused with advanced features.

python

from portkey import Portkey

snippet
code
portkey = Portkey(api_key=PORTKEY_API_KEY)
response = portkey.chat.completions.create(
model="gpt-4",
    messages=messages,
    config={
        "retry": {"attempts": 3},
        "fallback": [{"model": "claude-3-sonnet"}],
        "cache": {"enabled": True}
    }

)

🔧

Building a Simple Gateway

Here's a minimal FastAPI gateway:

python

from fastapi import FastAPI, HTTPException

import httpx

import random

snippet
code
app = FastAPI()

Provider configurations

PROVIDERS = {

example
code
"openai": {
        "base_url": "https://api.openai.com/v1",
        "api_key": OPENAI_KEY,
        "models": ["gpt-4", "gpt-3.5-turbo"]
    },
    "anthropic": {
        "base_url": "https://api.anthropic.com/v1",
        "api_key": ANTHROPIC_KEY,
        "models": ["claude-3-opus", "claude-3-sonnet"]
    }

}

@app.post("/v1/chat/completions")

async def chat_completions(request: dict):

example
code
model = request.get("model")

example
code
# Find provider for model
    provider = find_provider_for_model(model)
    if not provider:
        raise HTTPException(400, f"Model {model} not supported")

example
code
# Try primary provider
    try:
        return await call_provider(provider, request)
    except Exception as e:
        # Fallback logic
        fallback = get_fallback_provider(model)
        if fallback:
            return await call_provider(fallback, request)
        raise HTTPException(503, "All providers failed")

async def call_provider(provider: dict, request: dict):

example
code
async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{provider['base_url']}/chat/completions",
            headers={"Authorization": f"Bearer {provider['api_key']}"},
            json=request,
            timeout=60.0
        )
        return response.json()

📊

Intelligent Routing Strategies

Cost-Based Routing

python

snippet
code
def route_by_cost(prompt: str, complexity: str = "auto"):
if complexity == "low":
        return "gpt-3.5-turbo"  # $0.50/1M tokens
    elif complexity == "high":
        return "gpt-4"  # $30/1M tokens
# Auto-detect
    if is_simple_task(prompt):
        return "claude-3.5-haiku"  # Cheapest
    return "gpt-4-turbo"  # Default

Capability-Based Routing

python

CAPABILITIES = {

example
code
"code": ["claude-3.5-sonnet", "gpt-4"],
    "long_context": ["gemini-1.5-pro", "claude-3-opus"],
    "vision": ["gpt-4-vision", "claude-3-opus"],
    "cheap": ["claude-3.5-haiku", "gpt-3.5-turbo"]

}

def route_by_capability(task_type: str):

example
code
providers = CAPABILITIES.get(task_type, ["gpt-4"])
    return providers[0]  # Try best first

Latency-Based Routing

python

class LatencyTracker:

example
code
def __init__(self):
        self.latencies = defaultdict(list)

example
code
def get_fastest_provider(self, model_tier: str):
        # Return provider with lowest p99 latency
        candidates = MODEL_TIERS[model_tier]
        return min(candidates, key=lambda p: self.get_p99(p))

Fallback Strategies

Simple Retry

python

async def call_with_fallback(request):

example
code
providers = ["openai", "anthropic", "local"]

example
code
for provider in providers:
        try:
            return await call_provider(provider, request)
        except Exception as e:
            logger.warning(f"{provider} failed: {e}")
            continue

example
code
raise Exception("All providers exhausted")

Model Degradation

python

FALLBACK_CHAIN = {

example
code
"gpt-4": ["claude-3-opus", "gpt-4-turbo", "claude-3-sonnet"],
    "claude-3-opus": ["gpt-4", "claude-3-sonnet", "gpt-3.5-turbo"]

}

async def call_with_degradation(primary_model: str, request: dict):

example
code
chain = [primary_model] + FALLBACK_CHAIN.get(primary_model, [])

example
code
for model in chain:
        try:
            request["model"] = model
            return await call_provider(model, request)
        except Exception:
            continue

example
code
raise Exception("All models in chain failed")

💡

Rate Limiting and Quota Management

Token Bucket Algorithm

python

import time

from dataclasses import dataclass

@dataclass

class RateLimiter:

example
code
requests_per_minute: int
    tokens_per_minute: int

example
code
def __post_init__(self):
        self.request_bucket = self.requests_per_minute
        self.token_bucket = self.tokens_per_minute
        self.last_update = time.time()

example
code
def consume(self, tokens: int) -> bool:
        now = time.time()
        elapsed = now - self.last_update

example
code
# Replenish buckets
        self.request_bucket = min(
            self.requests_per_minute,
            self.request_bucket + elapsed * (self.requests_per_minute / 60)
        )
        self.token_bucket = min(
            self.tokens_per_minute,
            self.token_bucket + elapsed * (self.tokens_per_minute / 60)
        )

example
code
# Try to consume
        if self.request_bucket >= 1 and self.token_bucket >= tokens:
            self.request_bucket -= 1
            self.token_bucket -= tokens
            self.last_update = now
            return True
        return False

🔬

Unified Observability

The gateway becomes your single source of truth:

python

@app.post("/v1/chat/completions")

async def chat_completions(request: dict):

example
code
start = time.time()
    provider = None
    error = None

example
code
try:
        provider = select_provider(request)
        response = await call_provider(provider, request)

example
code
# Log to your metrics system
        metrics.record(
            provider=provider,
            model=request["model"],
            latency=time.time() - start,
            input_tokens=response["usage"]["prompt_tokens"],
            output_tokens=response["usage"]["completion_tokens"],
            cost=calculate_cost(response),
            cached=False
        )

example
code
return response

example
code
except Exception as e:
        metrics.record_error(provider, request["model"], str(e))
        raise

🛡️

Practical Takeaways

Start with LiteLLM — Free, open source, 100+ providers

Implement fallback chains — GPT-4 → Claude → cheaper models

Route by task type — Coding to Claude, speed to GPT-3.5, cost to Haiku

Centralize observability — One dashboard for all AI spend

Rate limit per user — Prevent one user from exhausting quotas

Cache at the gateway — Semantic + exact match in one place

Abstract provider details — Your app shouldn't know which API it's calling

📦

What's Next?

Episode 95: How Midjourney Works — Diffusion models, upscaling pipelines, text-to-image, and aesthetic training. The visual side of generative AI.

← Previous

Ep 83: Semantic Caching

Next: Episode 85 — Model Registries

You've trained a model. It works. You push it to production. Then someone asks: which version is running? Model registries solve this.

This is part of a 98-episode series covering AI engineering from tokens to production deployment.

AI Gateway Architecture: One API to Rule Them All

Core Responsibilities

Function Description

Routing Send requests to the right model/provider

Fallbacks Retry with backup providers on failure

Load balancing Distribute across multiple keys/instances

Rate limiting Prevent quota exhaustion

Popular Gateway Solutions

One interface, any provider

Response format is identical regardless of provider

Key features:

Unified response format

Automatic retries and fallbacks

Built-in cost tracking

Point OpenAI SDK at OpenRouter

Access any model

Building a Simple Gateway

Provider configurations

PROVIDERS = {

Intelligent Routing Strategies

CAPABILITIES = {

Fallback Strategies

Simple Retry

Model Degradation

FALLBACK_CHAIN = {

Rate Limiting and Quota Management

Token Bucket Algorithm

Unified Observability

The gateway becomes your single source of truth:

Practical Takeaways

Ep 83: Semantic Caching

Ep 85: Model Registries

Next: Episode 85 — Model Registries