API Gateway — infra.bea.ai

The API gateway is the front door to your inference service. But LLM serving has gateway requirements that differ fundamentally from traditional web APIs: requests vary in cost by 100×, responses stream for seconds, and prompts contain sensitive data that must be logged carefully.

1. Why LLM Gateways Are Different

Property	Traditional API	LLM API
Request cost	~uniform	100× variance (10 tokens vs 10,000)
Response time	50-200ms	500ms-60s
Response delivery	Single JSON	Token-by-token stream (SSE)
Rate limiting	Per-request	Per-token (cost varies dramatically)
Payload sensitivity	Moderate	High (prompts contain user data)
Backend selection	Round-robin	Model-specific, prefix-cache-aware

A gateway that treats LLM requests like REST requests will fail — you need token-aware rate limiting, streaming-compatible proxying, and model-specific routing.

2. Architecture

API Gateway Architecture

Client → Gateway → Router → Inference Pool → Streamed Response
           │                    │
           ├── Auth             ├── vLLM instance 1
           ├── Rate limit       ├── vLLM instance 2
           ├── Log              └── vLLM instance 3
           └── Transform

Request Flow

class LLMGateway:
    async def handle_request(self, request: Request) -> StreamingResponse:
        # 1. Authenticate
        tenant = await self.authenticate(request)
        
        # 2. Parse and validate
        body = await request.json()
        model = body.get("model", "default")
        max_tokens = body.get("max_tokens", 4096)
        
        # 3. Token-based rate limiting
        estimated_tokens = count_tokens(body["messages"]) + max_tokens
        if not await self.rate_limiter.allow(tenant.id, estimated_tokens):
            raise HTTPException(429, "Token rate limit exceeded")
        
        # 4. Route to appropriate backend
        backend = self.router.select(model, tenant.tier)
        
        # 5. Proxy request with streaming
        async def stream_response():
            actual_tokens = 0
            async for chunk in backend.stream(body):
                actual_tokens += chunk.token_count
                yield f"data: {json.dumps(chunk.dict())}\n\n"
            
            # 6. Log completion (after streaming finishes)
            await self.logger.log_completion(
                tenant=tenant,
                model=model,
                input_tokens=count_tokens(body["messages"]),
                output_tokens=actual_tokens,
                latency_ms=elapsed(),
            )
            
            # 7. Update billing meter
            await self.billing.record(tenant.id, actual_tokens)
        
        return StreamingResponse(stream_response(), media_type="text/event-stream")

3. Token-Based Rate Limiting

Per-request rate limiting is unfair — one 10-token request and one 10,000-token request cost vastly different amounts.

Token Bucket with Per-Minute Limits

class TokenRateLimiter:
    """Rate limit by token count, not request count."""
    
    def __init__(self, redis: Redis):
        self.redis = redis
    
    async def allow(self, tenant_id: str, estimated_tokens: int) -> bool:
        key = f"ratelimit:{tenant_id}:{current_minute()}"
        
        # Get current usage
        current = int(await self.redis.get(key) or 0)
        limit = await self.get_limit(tenant_id)
        
        if current + estimated_tokens > limit:
            return False
        
        # Increment usage
        pipe = self.redis.pipeline()
        pipe.incrby(key, estimated_tokens)
        pipe.expire(key, 120)  # TTL: 2 minutes
        await pipe.execute()
        return True
    
    async def get_limit(self, tenant_id: str) -> int:
        """Per-tenant token limits."""
        tier = await self.get_tier(tenant_id)
        return {
            "free":       10_000,     # 10K tokens/min
            "starter":   100_000,     # 100K tokens/min
            "pro":     1_000_000,     # 1M tokens/min
            "enterprise": 10_000_000, # 10M tokens/min
        }[tier]

Multi-Dimensional Limits

Production systems need limits on multiple axes:

Limit dimension	Typical values	Why
Tokens per minute	10K-10M	Cost control
Requests per minute	60-10,000	Prevent abuse patterns
Concurrent requests	10-1,000	Prevent resource monopolization
Max tokens per request	4K-128K	Prevent single huge requests
Monthly token budget	1M-1B	Billing cap

4. Request Routing

Model-Based Routing

class ModelRouter:
    """Route requests to the correct model backend."""
    
    def __init__(self, backends: dict[str, list[Backend]]):
        self.backends = backends  # model_name → [backend instances]
    
    def select(self, model: str, tenant_tier: str) -> Backend:
        candidates = self.backends.get(model, [])
        if not candidates:
            raise HTTPException(404, f"Model {model} not available")
        
        # Filter by tenant tier (enterprise gets dedicated instances)
        if tenant_tier == "enterprise":
            candidates = [b for b in candidates if b.pool == "dedicated"] or candidates
        
        # Select backend with shortest queue
        return min(candidates, key=lambda b: b.queue_depth)

A/B Traffic Splitting

Test new model versions on a subset of traffic:

class ABRouter:
    def __init__(self, control_backend, treatment_backend, treatment_pct=0.05):
        self.control = control_backend
        self.treatment = treatment_backend
        self.treatment_pct = treatment_pct
    
    def select(self, request_id: str) -> Backend:
        # Deterministic split based on request ID (consistent per user)
        if hash(request_id) % 100 < self.treatment_pct * 100:
            return self.treatment
        return self.control

5. Authentication and Tenant Management

class AuthMiddleware:
    async def authenticate(self, request: Request) -> Tenant:
        # API key authentication
        api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not api_key:
            raise HTTPException(401, "Missing API key")
        
        # Look up tenant (cached in Redis)
        tenant = await self.cache.get(f"apikey:{api_key}")
        if not tenant:
            tenant = await self.db.get_tenant_by_key(api_key)
            if not tenant:
                raise HTTPException(401, "Invalid API key")
            await self.cache.setex(f"apikey:{api_key}", 300, tenant.json())
        
        # Check if key is rotated/expired
        if tenant.key_expires_at and tenant.key_expires_at < datetime.utcnow():
            raise HTTPException(401, "API key expired")
        
        return tenant

Key Management

Feature	Implementation
Key rotation	Issue new key, grace period for old key (72h), then revoke
Key scoping	Per-key model access list, max tokens, IP allowlist
Usage attribution	Every request tagged with tenant + key ID for billing
Audit trail	Log key creation, rotation, revocation events

6. Request/Response Logging

LLM logs are sensitive — prompts contain user data, completions may contain PII.

class RequestLogger:
    async def log_completion(self, tenant, model, input_tokens, output_tokens, 
                              latency_ms, prompt=None, completion=None):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "tenant_id": tenant.id,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "latency_ms": latency_ms,
            "status": "success",
        }
        
        # PII redaction for prompt/completion logging
        if tenant.log_prompts:
            log_entry["prompt_hash"] = hashlib.sha256(prompt.encode()).hexdigest()
            if tenant.log_level == "full":
                log_entry["prompt"] = self.redact_pii(prompt)
                log_entry["completion"] = self.redact_pii(completion)
        
        await self.log_store.write(log_entry)
    
    def redact_pii(self, text: str) -> str:
        """Remove emails, phone numbers, SSNs, credit cards."""
        import re
        text = re.sub(r'\b[\w.+-]+@[\w-]+\.[\w.]+\b', '[EMAIL]', text)
        text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]', text)
        text = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]', text)
        text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', text)
        return text

Log Retention Policies

Data type	Retention	Storage
Request metadata (tokens, latency)	90 days	Time-series DB (ClickHouse)
Prompt/completion (redacted)	30 days	Encrypted object storage
Billing records	7 years	Immutable audit log
Error traces	14 days	Log aggregation (ELK)

7. Gateway Implementation Options

Option	Pros	Cons	Best for
Kong + plugins	Mature, plugin ecosystem	LLM streaming needs custom work	Enterprise with existing Kong
Envoy + Lua/WASM	High performance, streaming native	Complex configuration	High-throughput platforms
Custom (FastAPI/Go)	Full control, LLM-optimized	Build everything yourself	Startups, specific requirements
LiteLLM Proxy	LLM-native, multi-provider	Less battle-tested at scale	Multi-provider routing
Cloudflare AI Gateway	Managed, caching, analytics	Vendor lock-in	Quick start

Self-Check

Explain why per-request rate limiting is insufficient for LLM APIs
Implement token-based rate limiting with per-minute buckets
Design model-based routing with A/B traffic splitting
Build a request logger with PII redaction for prompt/completion data
Compare gateway implementation options for LLM serving

References

LiteLLM Proxy Documentation
Kong Gateway
Envoy Proxy
For LLM-specific load balancing, see Load Balancing for LLMs