AI Gateway

What Is an AI Gateway?

If your organization has more than one team calling LLM APIs, you need a gateway. An AI gateway sits between your application and LLM providers, centralizing cross-cutting concerns like authentication, rate limiting, routing, guardrails, and observability.

┌──────────┐     ┌──────────────────────────────────┐     ┌──────────────┐
│           │     │          AI Gateway               │     │  OpenAI      │
│  App /    │     │                                    │     ├──────────────┤
│  Agent    │────>│  Auth -> Rate Limit -> Guardrails  │────>│  Anthropic   │
│  Service  │<────│  -> Router -> Log -> Transform     │<────│  Google      │
│           │     │                                    │     ├──────────────┤
└──────────┘     └──────────────────────────────────┘     │  Local/OSS   │
                                                           └──────────────┘

Core Components

Every AI gateway is built from the same building blocks. Understanding each component helps you decide whether to buy a managed gateway or build a custom one -- and what to prioritize first.

Gateway Products and Frameworks

The gateway market ranges from open-source proxies you run yourself to fully managed commercial platforms. Your choice depends on whether you need simple routing or enterprise features like audit logging and cost attribution.

Authentication and Authorization

The gateway is the single point where you map internal team keys to provider credentials, enforce permissions, and set spending limits. Getting this right prevents unauthorized model access and surprise bills.

API Key Management

# Gateway authenticates app, not end-user
# Map internal keys to provider keys

KEY_MAP = {
    "app-key-team-alpha": {
        "team": "alpha",
        "budget_monthly_usd": 500,
        "allowed_models": ["gpt-4o", "claude-sonnet-4"],
        "rate_limit_rpm": 60
    }
}

def authenticate(request):
    api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
    config = KEY_MAP.get(api_key)
    if not config:
        raise HTTPException(401, "Invalid API key")
    return config

Per-Team Permissions

Rate Limiting

Without rate limiting, a single runaway script can burn through your entire monthly budget in hours. Rate limits protect against both abuse and accidental cost explosions.

Token Bucket Algorithm

import time

class TokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.tokens = capacity
        self.refill_rate = refill_rate  # tokens per second
        self.last_refill = time.time()

    def consume(self, tokens: int = 1) -> bool:
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

Rate Limit Tiers

Model Routing

Intelligent routing is where the gateway pays for itself -- sending simple queries to cheap models and complex ones to capable models can cut costs by 50-80% without sacrificing quality.

Routing Strategies

Router Implementation

ROUTING_TABLE = {
    "simple_qa": {
        "primary": "gpt-4o-mini",
        "fallback": ["claude-haiku-3.5", "gemini-flash"]
    },
    "complex_reasoning": {
        "primary": "claude-sonnet-4",
        "fallback": ["gpt-4o", "gemini-pro"]
    },
    "code_generation": {
        "primary": "claude-sonnet-4",
        "fallback": ["gpt-4o"]
    },
    "vision": {
        "primary": "gpt-4o",
        "fallback": ["gemini-pro", "claude-sonnet-4"]
    }
}

def classify_and_route(request):
    task_type = classify_task(request.messages)  # lightweight classifier
    config = ROUTING_TABLE[task_type]
    return config["primary"]

Cascading Pattern

async def cascade_call(request, models=["gpt-4o-mini", "gpt-4o", "claude-sonnet-4"]):
    """Try cheaper model first; escalate if response quality is low."""
    for model in models:
        response = await call_model(model, request)

        # Quality gate: check if response is good enough
        quality = assess_quality(request, response)
        if quality.score > 0.8:
            return response

        # If last model, return whatever we have
        if model == models[-1]:
            return response

    return response

Failover Patterns

LLM providers have outages -- and when your entire product depends on a single API, that outage becomes your outage. Multi-provider failover with circuit breakers keeps your application running through provider disruptions.

Circuit Breaker

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure = None
        self.state = "closed"  # closed=normal, open=failing, half_open=testing

    def call(self, fn, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half_open"
            else:
                raise CircuitOpenError()

        try:
            result = fn(*args, **kwargs)
            if self.state == "half_open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise

Caching

Caching is the easiest way to reduce LLM costs and latency simultaneously. For applications with repetitive queries, a semantic cache can eliminate 15-40% of LLM calls entirely.

Semantic Cache

def get_or_generate(query: str, threshold: float = 0.95):
    query_embedding = embed(query)

    # Search cache
    cached = cache_db.search(query_embedding, top_k=1)
    if cached and cached[0].score > threshold:
        return cached[0].response  # Cache hit

    # Cache miss: call LLM
    response = llm.generate(query)
    cache_db.insert(query_embedding, response, ttl=3600)
    return response

Cost Attribution

When multiple teams share LLM infrastructure, you need per-team cost tracking for budgeting, chargeback, and identifying optimization opportunities. Without attribution, nobody owns the bill.

Token-Based Cost Tracking

PRICING = {
    "gpt-4o":           {"input": 2.50, "output": 10.00},   # per 1M tokens
    "gpt-4o-mini":      {"input": 0.15, "output": 0.60},
    "claude-sonnet-4":  {"input": 3.00, "output": 15.00},
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00},
}

def calculate_cost(model, input_tokens, output_tokens):
    prices = PRICING[model]
    cost = (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
    return round(cost, 6)

Cost Dashboard Metrics

Logging and Observability

You cannot debug, optimize, or audit what you do not log. Structured logging at the gateway layer gives you a single pane of glass across all LLM providers and teams.

What to Log

Do NOT log: Full prompt/response text in production (PII risk). Use a separate, access-controlled audit store if needed.

Common Pitfalls

Gateway mistakes are amplified across every request in your organization. A missing timeout or absent rate limit affects every team and every application simultaneously.