Beyond the API Key: Architecting a Multi-Provider AI Gateway with RBAC, Cost Routing, and Fallback Chains — Lessons from 40k+ Star OSS Tools

┌─────────────────────────────────────────────────────────────┐
│  Layer 5: Application Layer                                 │
│  (Your service calls a single /v1/chat/completions endpoint)│
├─────────────────────────────────────────────────────────────┤
│  Layer 4: Fallback & Resilience Engine                      │
│  (circuit breakers, retry with backoff, provider cascading) │
├─────────────────────────────────────────────────────────────┤
│  Layer 3: Routing & Cost Optimization                        │
│  (cost-aware model selection, budget enforcement)           │
├─────────────────────────────────────────────────────────────┤
│  Layer 2: Authorization & RBAC                              │
│  (per-team, per-model, per-feature access control)         │
├─────────────────────────────────────────────────────────────┤
│  Layer 1: Provider Abstraction & Normalization              │
│  (unified schema, adapter pattern, streaming passthrough)   │
└─────────────────────────────────────────────────────────────┘

// The internal, provider-agnostic contract
interface ChatRequest {
  model: string;          // logical model alias, e.g. "fast-7b" or "reasoning-large"
  messages: Message[];
  temperature?: number;
  maxTokens?: number;
  responseFormat?: "text" | "json" | "structured";
  metadata?: Record<string, unknown>; // caller-supplied, opaque to the gateway
}

interface ChatResponse {
  id: string;             // provider-unique response ID
  model: string;          // the *actual* model that served the request
  content: string;
  usage: { promptTokens: number; completionTokens: number; totalTokens: number };
  cost: { usd: number; currency: string };
  provider: string;       // which provider actually served it
  latencyMs: number;
  fallbackDepth: number;  // 0 = primary, 1 = first fallback, etc.
}

// Provider adapter interface
interface ProviderAdapter {
  readonly name: string;
  chat(req: NormalizedRequest, ctx: RequestContext): AsyncIterable<StreamChunk> | Promise<NormalizedResponse>;
  embeddings(req: EmbeddingRequest): Promise<NormalizedEmbeddingResponse>;
  isHealthy(): Promise<boolean>;
}

class OpenAIAdapter implements ProviderAdapter {
  readonly name = "openai";

  async chat(req: NormalizedRequest, ctx: RequestContext): AsyncIterable<StreamChunk> {
    const resp = await this.client.chat.completions.create({
      model: req.physicalModel,  // resolved alias → physical model
      messages: req.messages,
      temperature: req.temperature,
      max_tokens: req.maxTokens,
      stream: true,
    });

    // Translate OpenAI SSE chunks into the normalized StreamChunk format
    for await (const chunk of resp) {
      yield {
        contentDelta: chunk.choices[0]?.delta?.content ?? "",
        finishReason: chunk.choices[0]?.finish_reason ?? null,
        usage: chunk.usage, // OpenAI now streams usage in the final chunk
      };
    }
  }
}

permission = (team, model_alias, action, budget_cap, rate_limit)

{
  "team": "customer-support",
  "grants": [
    {
      "model": "fast-7b",
      "actions": ["chat", "stream"],
      "budgetUSD": { "daily": 25, "monthly": 400 },
      "rateLimit": { "requestsPerMin": 30 }
    },
    {
      "model": "reasoning-large",
      "actions": ["chat"],
      "budgetUSD": { "daily": 10, "monthly": 150 },
      "rateLimit": { "requestsPerMin": 5 }
    }
  ]
}

route(model_alias, ctx) →
  candidates = providers supporting model_alias
  filtered = candidates.filter(c => c.isHealthy() && c.hasCapacity(ctx.rateLimit))
  scored = filtered.map(c =>
    w_cost * normalizedCost(c, model_alias) +
    w_quality * (1 - qualityScore(c, model_alias)) +
    w_latency * normalizedP99Latency(c)
  )
  return argmin(scored)

function estimateCost(provider: string, model: string, promptTokens: number, completionTokens: number): number {
  const pricing = PRICING_TABLE[provider][model]; // { inputPerM, outputPerM, cacheReadPerM }
  let inputCost = (promptTokens / 1_000_000) * pricing.inputPerM;
  // Anthropic-style context-length tiering
  if (provider === "anthropic" && promptTokens > 100_000) {
    inputCost = (promptTokens / 1_000_000) * pricing.inputPerM * 2.0; // 200k tier
  }
  // Prompt caching discount
  const cachedTokens = ctx.cachedPromptTokens ?? 0;
  inputCost -= (cachedTokens / 1_000_000) * pricing.cacheReadPerM;
  const outputCost = (completionTokens / 1_000_000) * pricing.outputPerM;
  return inputCost + outputCost;
}

{
  "alias": "fast-7b",
  "fallbackChain": [
    { "provider": "together", "model": "meta-llama/Llama-3-8B-Instruct", "weight": 0.7 },
    { "provider": "openai", "model": "gpt-4o-mini", "weight": 0.2 },
    { "provider": "ollama-local", "model": "llama3:8b", "weight": 0.1 }
  ],
  "circuitBreaker": {
    "failureThreshold": 5,
    "openDurationSec": 60,
    "halfOpenMaxRequests": 2
  }
}

{
  "requestId": "uuid",
  "teamId": "customer-support",
  "modelAlias": "fast-7b",
  "provider": "together",
  "actualModel": "meta-llama/Llama-3-8B-Instruct",
  "promptTokens": 1247,
  "completionTokens": 312,
  "totalTokens": 1559,
  "costUSD": 0.0034,
  "latencyMs": 2340,
  "fallbackDepth": 0,
  "status": "success",
  "timestamp": "2025-01-15T09:32:17Z",
  "routingStrategy": "balanced",
  "requestId": "uuid"
}

import { ProviderAdapter, ChatRequest, ChatResponse } from "./types";
import { OpenAIAdapter } from "./adapters/openai";
import { AnthropicAdapter } from "./adapters/anthropic";
import { TogetherAdapter } from "./adapters/together";
import { PolicyEngine } from "./rbac/policy-engine";
import { CostRouter } from "./routing/cost-router";
import { FallbackChain, CircuitBreaker } from "./resilience/fallback";

const adapters: Record<string, ProviderAdapter> = {
  openai: new OpenAIAdapter(),
  anthropic: new AnthropicAdapter(),
  together: new TogetherAdapter(),
};

const policyEngine = new PolicyEngine(); // loads RBAC policies from DB/config
const costRouter = new CostRouter(adapters, PRICING_TABLE);
const fallback = new FallbackChain(config.fallbackChains);

export async function handleChat(req: ChatRequest, auth: AuthContext): Promise<ChatResponse> {
  // LAYER 2: RBAC
  const permission = policyEngine.evaluate(auth.teamId, req.model, req.metadata?.action ?? "chat");
  if (!permission.allowed) {
    throw new GatewayError("403", `Team '${auth.teamId}' not authorized for model '${req.model}'`);
  }
  if (permission.budgetExceeded) {
    throw new GatewayError("429", "Team budget exceeded for this model", { retryAfter: permission.nextBudgetReset });
  }

  // LAYER 3: Cost-aware routing
  const route = costRouter.route(req.model, auth.teamId, {
    promptTokensEstimate: estimateTokens(req.messages),
    strategy: permission.routingStrategy ?? "balanced",
  });

  // LAYER 4: Fallback chain with circuit breakers
  let depth = 0;
  for (const target of route.chain) {
    const breaker = fallback.getBreaker(target.provider, target.model);
    if (breaker.state === "open") { depth++; continue; }

    try {
      const adapter = adapters[target.provider];
      const response = await adapter.chat(
        { ...req, physicalModel: target.model },
        { teamId: auth.teamId, circuitBreaker: breaker }
      );
      breaker.recordSuccess();
      return { ...response, provider: target.provider, fallbackDepth: depth, cost: { usd: response.usage.totalTokens * target.costPerToken, currency: "USD" } };
    } catch (err) {
      const failureType = classifyFailure(err); // "4xx" | "5xx" | "timeout" | "content"
      if (failureType === "4xx" && !(err instanceof BudgetExceededError)) {
        depth++; // skip to next provider, do NOT retry same
        continue;
      }
      breaker.recordFailure();
      if (failureType === "5xx" && depth === 0 && route.chain.length > 1) {
        // retry same provider once before cascading
        depth++;
        continue;
      }
      depth++;
    }
  }

  throw new GatewayError("502", "All providers in fallback chain failed");
}

# gateway/app.py
import time
import json
import hashlib
import httpx
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

class Provider(Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    MISTRAL = "mistral"
    LOCAL_LLAMA = "local_llama"

class Role(Enum):
    SUPERVISOR = "supervisor"
    DEVELOPER = "developer"
    GUEST = "guest"

# ─── RBAC ───────────────────────────────────────────────────────────
RBAC_MATRIX: dict[Role, list[str]] = {
    Role.SUPERVISOR: ["*"],
    Role.DEVELOPER:  [
        "providers:openai:invoke",
        "providers:anthropic:invoke",
        "providers:mistral:invoke",
        "providers:local_llama:invoke",
        "cost_budget:read",
        "audit_log:read",
    ],
    Role.GUEST: [
        "providers:openai:invoke",
        "providers:local_llama:invoke",
    ],
}

def permission_granted(role: Role, action: str) -> bool:
    perms = RBAC_MATRIX.get(role, [])
    return "*" in perms or action in perms

# ─── Cost routing ──────────────────────────────────────────────────
@dataclass
class CostProfile:
    input_per_1k: float
    output_per_1k: float
    rate_limit_rpm: int
    burst_window_s: int

COST_TABLE: dict[Provider, CostProfile] = {
    Provider.OPENAI:      CostProfile(0.005, 0.015, 500, 60),
    Provider.ANTHROPIC:  CostProfile(0.003, 0.015, 300, 60),
    Provider.MISTRAL:     CostProfile(0.002, 0.006, 400, 60),
    Provider.LOCAL_LLAMA: CostProfile(0.000, 0.000, 9999, 60),
}

def estimate_cost(provider: Provider, in_tokens: int, out_tokens: int) -> float:
    p = COST_TABLE[provider]
    return (in_tokens / 1000) * p.input_per_1k + (out_tokens / 1000) * p.output_per_1k

def cheapest_provider(budget_per_call: float, in_tokens: int, out_tokens: int) -> Provider:
    candidates = []
    for prov, profile in COST_TABLE.items():
        est = estimate_cost(prov, in_tokens, out_tokens)
        if est <= budget_per_call:
            candidates.append((est, prov))
    if not candidates:
        raise HTTPException(402, "All providers exceed per-call budget")
    candidates.sort(key=lambda x: x[0])
    return candidates[0][1]

# ─── Fallback chain ────────────────────────────────────────────────
@dataclass
class FallbackChain:
    ordered_providers: list[Provider]
    max_retries_per_node: int = 2
    circuit_breaker_threshold: int = 5
    circuit_breaker_cooldown_s: int = 30

DEFAULT_CHAIN = FallbackChain(
    ordered_providers=[Provider.OPENAI, Provider.ANTHROPIC, Provider.MISTRAL, Provider.LOCAL_LLAMA]
)

class CircuitBreaker:
    def __init__(self, threshold: int, cooldown: int):
        self.failures = 0
        self.threshold = threshold
        self.cooldown = cooldown
        self.opened_at: Optional[float] = None

    def record_success(self):
        self.failures = 0
        self.opened_at = None

    def record_failure(self):
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.time()

    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.time() - self.opened_at < self.cooldown:
            return True
        self.opened_at = None  # half-open → allow one probe
        self.failures = self.threshold - 1
        return False

# ─── API surface ───────────────────────────────────────────────────
app = FastAPI(title="Multi-Provider AI Gateway")

class InvokeRequest(BaseModel):
    model_hint: str = "chat"
    prompt: str
    est_in_tokens: int = 200
    est_out_tokens: int = 300
    budget_per_call: float = 0.05
    force_provider: Optional[Provider] = None

class InvokeResponse(BaseModel):
    provider: str
    cost_estimate: float
    response: str
    latency_ms: int
    fallback_triggered: bool = False

breakers: dict[Provider, CircuitBreaker] = {
    p: CircuitBreaker(DEFAULT_CHAIN.circuit_breaker_threshold,
                      DEFAULT_CHAIN.circuit_breaker_cooldown_s)
    for p in Provider
}

@app.post("/v1/invoke", response_model=InvokeResponse)
async def invoke(
    body: InvokeRequest,
    x_api_key: str = Header(...),
    x_role: str = Header(default=Role.DEVELOPER.value),
):
    role = Role(x_role)
    # 1. RBAC gate
    for p in DEFAULT_CHAIN.ordered_providers:
        action = f"providers:{p.value}:invoke"
        if not permission_granted(role, action):
            # filter chain to only allowed providers
            DEFAULT_CHAIN.ordered_providers = [
                pp for pp in DEFAULT_CHAIN.ordered_providers
                if permission_granted(role, f"providers:{pp.value}:invoke")
            ]
            break

    if not DEFAULT_CHAIN.ordered_providers:
        raise HTTPException(403, "No providers permitted for your role")

    # 2. Cost routing (unless a provider is forced)
    if body.force_provider:
        target = body.force_provider
    else:
        target = cheapest_provider(body.budget_per_call, body.est_in_tokens, body.est_out_tokens)

    # 3. Fallback chain with circuit breakers
    chain = [target] + [p for p in DEFAULT_CHAIN.ordered_providers if p != target]
    fallback_triggered = False

    for i, prov in enumerate(chain):
        breaker = breakers[prov]
        if breaker.is_open():
            continue
        start = time.perf_counter()
        try:
            # In production: call the actual provider SDK
            response_text = await _call_provider(prov, body)
            breaker.record_success()
            latency = int((time.perf_counter() - start) * 1000)
            return InvokeResponse(
                provider=prov.value,
                cost_estimate=round(estimate_cost(prov, body.est_in_tokens, body.est_out_tokens), 6),
                response=response_text,
                latency_ms=latency,
                fallback_triggered=fallback_triggered,
            )
        except Exception as exc:
            breaker.record_failure()
            if i < len(chain) - 1:
                fallback_triggered = True
                continue
            raise HTTPException(502, f"All providers failed. Last error: {exc}")

async def _call_provider(prov: Provider, body: InvokeRequest) -> str:
    """Placeholder – replace with real SDK calls (openai, anthropic, etc.)"""
    import random
    if random.random() < 0.1:
        raise ConnectionError(f"{prov.value} simulated timeout")
    return f"[{prov.value}] Echo: {body.prompt[:80]}"

@app.get("/v1/cost/estimate")
def cost_estimate(provider: Provider, in_tokens: int = 1000, out_tokens: int = 1000):
    return {"cost_usd": estimate_cost(provider, in_tokens, out_tokens)}

uvicorn gateway.app:app --port 8443 --workers 4
curl -X POST http://localhost:8443/v1/invoke \
  -H "Content-Type: application/json" \
  -H "x-api-key: sk-test-abc" \
  -H "x-role: developer" \
  -d '{"prompt": "Explain circuit breakers", "est_in_tokens": 50, "est_out_tokens": 200, "budget_per_call": 0.01}'

score(provider) = α · (1 / est_cost)            # cost efficiency
               + β · quality_tier(provider)     # 1.0–0.4 by model class
               + γ · (1 - circuit_open)         # health signal
               + δ · rate_headroom(provider)    # remaining RPM fraction

QUALITY_TIER: dict[Provider, float] = {
    Provider.OPENAI: 1.0,
    Provider.ANTHROPIC: 0.95,
    Provider.MISTRAL: 0.85,
    Provider.LOCAL_LLAMA: 0.70,
}

def weighted_router(
    in_tokens: int, out_tokens: int,
    budget: float,
    alphas: tuple[float, float, float, float] = (0.4, 0.3, 0.2, 0.1),
) -> Provider:
    α, β, γ, δ = alphas
    best, best_score = None, -1.0
    for prov in Provider:
        est = estimate_cost(prov, in_tokens, out_tokens)
        if est > budget:
            continue
        cost_score = α / max(est, 0.001)
        quality = β * QUALITY_TIER[prov]
        health  = γ * (0.0 if breakers[prov].is_open() else 1.0)
        # rate_headroom: approximate as 1 - (current_rpm / limit)
        headroom = δ * 0.8  # simplified; use Redis counters in prod
        total = cost_score + quality + health + headroom
        if total > best_score:
            best, best_score = prov, total
    if best is None:
        raise HTTPException(402, "No provider within budget")
    return best

{
  "sub": "user:8f3a",
  "role": "developer",
  "team": "ml-platform",
  "permissions": [
    "providers:openai:invoke",
    "providers:mistral:invoke",
    "cost_budget:read"
  ],
  "budget_monthly_usd": 50.00,
  "i_at": 1717000000,
  "exp": 1717003600
}

from jose import jwt, JWTError

def require_role(
    authorization: str = Header(...),
):
    token = authorization.replace("Bearer ", "")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        raise HTTPException(401, "Invalid token")
    return Role(payload["role"])

INSERT INTO gateway_audit (ts, api_key_hash, role, provider, in_tok, out_tok, est_cost, status)
VALUES (now(), sha256(:key), :role, :prov, :in, :out, :cost, 'ok');

async def execute_with_backoff(prov: Provider, body: InvokeRequest, max_retries: int = 2):
    for attempt in range(max_retries):
        try:
            return await _call_provider(prov, body)
        except (httpx.ReadError, httpx.ConnectTimeout):
            if attempt == max_retries - 1:
                raise
            wait = min(2 ** attempt + random.random(), 30)
            await asyncio.sleep(wait)
    raise

class UnifiedStream:
    def __aiter__(self) -> AsyncIterator[str]:
        ...

def to_unified(provider: Provider, raw_response) -> UnifiedStream:
    match provider:
        case Provider.OPENAI:
            return OpenAIStream(raw_response)      # iterates choices[0].delta.content
        case Provider.ANTHROPIC:
            return AnthropicStream(raw_response)   # iterates message deltas
        case Provider.MISTRAL:
            return MistralStream(raw_response)
        case Provider.LOCAL_LLAMA:
            return LocalStream(raw_response)       # Ollama / vLLM NDJSON

# metrics.yaml (Grafana data source)
- gateway_request_total{provider, model, role, status}
- gateway_latency_seconds{provider, status}
- gateway_cost_usd_total{provider, team}
- gateway_fallback_triggered_total{from_provider, to_provider}
- gateway_circuit_breaker_state{provider}      # 0=closed, 1=open, 2=half-open
- gateway_budget_exhausted_total{team}
- gateway_rate_limit_429_total{provider}
- gateway_token_in_total{provider}
- gateway_token_out_total{provider}
- gateway_stream_chunks_total{provider}
- gateway_stream_first_token_ms{provider}
- gateway_rbac_denied_total{role, provider}
- gateway_cache_hit_ratio{provider}             # semantic cache
- gateway_tls_handshake_failures{provider}

- alert: TeamBudgetRun
  expr: sum by (team) (increase(gateway_cost_usd_total[24h])) > 0.8 * team_budget
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "{{ $labels.team }} at 80% of 24h budget"

import numpy as np
from sentence_transformers import SentenceTransformer

_model = SentenceTransformer("all-MiniLM-L6-v2")

def semantic_cache_lookup(prompt: str, threshold: float = 0.92) -> Optional[str]:
    embedding = _model.encode(prompt, normalize_embeddings=True)
    best_score, best_val = 0.0, None
    for cached_emb, cached_val in _redis_zset("prompt_cache"):
        score = float(np.dot(embedding, cached_emb))
        if score > best_score:
            best_score, best_val = score, cached_val
    return best_val if best_score >= threshold else None

def semantic_cache_store(prompt: str, response: str):
    embedding = _model.encode(prompt, normalize_embeddings=True)
    _redis_zadd("prompt_cache", embedding.tolist(), response, ttl=3600)