Upload production/production_server.py

production/production_server.py

@@ -5,12 +5,14 @@ Features:
 - FastAPI with async endpoints
 - Distributed rate limiting (Redis-backed token bucket)
 - Circuit breaker for external API resilience
+- Automatic fallback: NIM (primary) -> Cloudflare Workers AI (fallback)
 - Request/response caching with Redis TTL
 - Multi-tenant session isolation
 - Health checks and graceful shutdown
 - Structured logging with correlation IDs
 - Cost tracking and budget enforcement
 - Connection pooling for all HTTP clients
+- Cloudflare Workers AI support via OpenAI-compatible API
 """
 
 import asyncio
@@ -26,22 +28,23 @@ from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from typing import Any, Optional
 
-import aioredis
+import redis.asyncio as aioredis
 import asyncpg
 from fastapi import FastAPI, HTTPException, Request, Depends, BackgroundTasks
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
-from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 import uvicorn
 from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST
+import httpx
+from tenacity import retry, stop_after_attempt, wait_exponential, RetryError
 
 # ---------------------------------------------------------------------------
 # Configuration
 # ---------------------------------------------------------------------------
 
 REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
-DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://localhost/ml_intern")
+DATABASE_URL = os.environ.get("DATABASE_URL", "")
 MAX_CONCURRENT_REQUESTS = int(os.environ.get("MAX_CONCURRENT_REQUESTS", "100"))
 DEFAULT_RPM_LIMIT = int(os.environ.get("DEFAULT_RPM_LIMIT", "40"))
 REQUEST_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "120"))
@@ -50,6 +53,20 @@ BUDGET_USD_PER_SESSION = float(os.environ.get("BUDGET_USD_PER_SESSION", "10.0"))
 CIRCUIT_BREAKER_FAILURE_THRESHOLD = int(os.environ.get("CIRCUIT_BREAKER_FAILURE_THRESHOLD", "5"))
 CIRCUIT_BREAKER_RECOVERY_TIMEOUT = int(os.environ.get("CIRCUIT_BREAKER_RECOVERY_TIMEOUT", "60"))
 
+# Provider-specific endpoints
+NIM_API_BASE = os.environ.get("NIM_API_BASE", "https://integrate.api.nvidia.com/v1")
+CLOUDFLARE_API_KEY = os.environ.get("CLOUDFLARE_API_KEY", "")
+CLOUDFLARE_ACCOUNT_ID = os.environ.get("CLOUDFLARE_ACCOUNT_ID", "")
+
+# Fallback configuration
+FALLBACK_ENABLED = os.environ.get("FALLBACK_ENABLED", "true").lower() == "true"
+FALLBACK_PRIMARY = os.environ.get("FALLBACK_PRIMARY", "nim")
+FALLBACK_SECONDARY = os.environ.get("FALLBACK_SECONDARY", "cloudflare")
+
+# MLX (local Apple Silicon)
+MLX_API_BASE = os.environ.get("MLX_API_BASE", "http://localhost:8000/v1")
+MLX_ENABLED = os.environ.get("MLX_ENABLED", "false").lower() == "true"
+
 # ---------------------------------------------------------------------------
 # Prometheus Metrics
 # ---------------------------------------------------------------------------
@@ -89,22 +106,28 @@ CIRCUIT_BREAKER_STATE = Gauge(
     "Circuit breaker state (0=closed, 1=half-open, 2=open)",
     ["provider"],
 )
+FALLBACK_COUNT = Counter(
+    "ml_intern_fallback_total",
+    "Fallback events between providers",
+    ["from_provider", "to_provider", "reason"],
+)
 
 # ---------------------------------------------------------------------------
 # Structured Logging
 # ---------------------------------------------------------------------------
 
-class CorrelationIdFilter(logging.Filter):
-    def filter(self, record: logging.LogRecord) -> bool:
-        record.correlation_id = getattr(record, "correlation_id", "none")
-        return True
-
 logging.basicConfig(
     level=logging.INFO,
     format="%(asctime)s | %(levelname)s | correlation_id=%(correlation_id)s | %(name)s | %(message)s",
     handlers=[logging.StreamHandler(sys.stdout)],
 )
 logger = logging.getLogger("ml_intern.production")
+
+class CorrelationIdFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.correlation_id = getattr(record, "correlation_id", "none")
+        return True
+
 logger.addFilter(CorrelationIdFilter())
 
 # ---------------------------------------------------------------------------
@@ -112,17 +135,18 @@ logger.addFilter(CorrelationIdFilter())
 # ---------------------------------------------------------------------------
 
 class DatabasePool:
-    """Async PostgreSQL connection pool with prepared statements."""
-    
     def __init__(self, dsn: str):
         self.dsn = dsn
         self._pool: Optional[asyncpg.Pool] = None
     
     async def connect(self):
+        if not self.dsn:
+            logger.warning("No DATABASE_URL set — skipping database connection")
+            return
         self._pool = await asyncpg.create_pool(
             self.dsn,
-            min_size=5,
-            max_size=20,
+            min_size=2,
+            max_size=10,
             command_timeout=60,
         )
         logger.info("Database pool connected")
@@ -133,28 +157,28 @@ class DatabasePool:
             logger.info("Database pool disconnected")
     
     async def execute(self, query: str, *args):
+        if not self._pool:
+            return
         async with self._pool.acquire() as conn:
             return await conn.execute(query, *args)
     
     async def fetch(self, query: str, *args):
+        if not self._pool:
+            return []
         async with self._pool.acquire() as conn:
             return await conn.fetch(query, *args)
     
-    async def fetchrow(self, query: str, *args):
-        async with self._pool.acquire() as conn:
-            return await conn.fetchrow(query, *args)
-    
     async def fetchval(self, query: str, *args):
+        if not self._pool:
+            return None
         async with self._pool.acquire() as conn:
             return await conn.fetchval(query, *args)
 
 # ---------------------------------------------------------------------------
-# Redis Layer (Caching + Rate Limiting + Distributed State)
+# Redis Layer
 # ---------------------------------------------------------------------------
 
 class RedisManager:
-    """Redis client for caching, rate limiting, and distributed state."""
-    
     def __init__(self, url: str):
         self.url = url
         self._redis: Optional[aioredis.Redis] = None
@@ -169,8 +193,6 @@ class RedisManager:
             await self._redis.close()
             logger.info("Redis disconnected")
     
-    # --- Caching ---
-    
     async def get_cache(self, key: str) -> Optional[str]:
         val = await self._redis.get(key)
         if val:
@@ -182,26 +204,16 @@ class RedisManager:
     async def set_cache(self, key: str, value: str, ttl: int = CACHE_TTL_SECONDS):
         await self._redis.setex(key, ttl, value)
     
-    async def delete_cache(self, key: str):
-        await self._redis.delete(key)
-    
-    # --- Rate Limiting (Token Bucket) ---
-    
     async def check_rate_limit(self, key: str, rpm: int) -> tuple[bool, float]:
-        """Check if request is within rate limit. Returns (allowed, retry_after)."""
         now = time.time()
         bucket_key = f"ratelimit:{key}"
-        
-        # Lua script for atomic token bucket
         script = """
         local key = KEYS[1]
         local now = tonumber(ARGV[1])
         local rpm = tonumber(ARGV[2])
         local interval = 60.0 / rpm
-        
         local last = redis.call('hget', key, 'last')
         local tokens = redis.call('hget', key, 'tokens')
-        
         if not last then
             last = 0
             tokens = 1
@@ -209,10 +221,8 @@ class RedisManager:
             last = tonumber(last)
             tokens = tonumber(tokens)
         end
-        
         local elapsed = now - last
         tokens = math.min(1, tokens + elapsed / interval)
-        
         if tokens >= 1 then
             tokens = tokens - 1
             redis.call('hmset', key, 'last', now, 'tokens', tokens)
@@ -225,14 +235,11 @@ class RedisManager:
             return {0, retry_after}
         end
         """
-        
         result = await self._redis.eval(script, 1, bucket_key, now, rpm)
         allowed = bool(result[0])
         retry_after = float(result[1]) if not allowed else 0.0
         return allowed, retry_after
     
-    # --- Circuit Breaker State ---
-    
     async def get_circuit_state(self, provider: str) -> dict:
         key = f"circuit:{provider}"
         val = await self._redis.get(key)
@@ -249,8 +256,6 @@ class RedisManager:
 # ---------------------------------------------------------------------------
 
 class CircuitBreaker:
-    """Distributed circuit breaker using Redis."""
-    
     def __init__(self, redis: RedisManager, provider: str):
         self.redis = redis
         self.provider = provider
@@ -259,21 +264,17 @@ class CircuitBreaker:
     
     async def can_execute(self) -> bool:
         state = await self.redis.get_circuit_state(self.provider)
-        
         if state["state"] == "open":
             if time.time() - state["last_failure"] > self.recovery_timeout:
                 state["state"] = "half-open"
                 state["failures"] = 0
                 await self.redis.set_circuit_state(self.provider, state)
                 CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(1)
-                logger.info(f"Circuit breaker for {self.provider} entering half-open state")
+                logger.info(f"Circuit breaker {self.provider} half-open")
                 return True
             CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(2)
             return False
-        
-        CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(
-            0 if state["state"] == "closed" else 1
-        )
+        CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(0 if state["state"] == "closed" else 1)
         return True
     
     async def record_success(self):
@@ -283,22 +284,101 @@ class CircuitBreaker:
             state["failures"] = 0
             await self.redis.set_circuit_state(self.provider, state)
             CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(0)
-            logger.info(f"Circuit breaker for {self.provider} closed after recovery")
+            logger.info(f"Circuit breaker {self.provider} closed")
     
     async def record_failure(self):
         state = await self.redis.get_circuit_state(self.provider)
         state["failures"] += 1
         state["last_failure"] = time.time()
-        
         if state["failures"] >= self.failure_threshold:
             state["state"] = "open"
             CIRCUIT_BREAKER_STATE.labels(provider=self.provider).set(2)
-            logger.warning(
-                f"Circuit breaker for {self.provider} OPENED after "
-                f"{state['failures']} failures"
+            logger.warning(f"Circuit breaker {self.provider} OPENED after {state['failures']} failures")
+        await self.redis.set_circuit_state(self.provider, state)
+
+# ---------------------------------------------------------------------------
+# Fallback Manager
+# ---------------------------------------------------------------------------
+
+@dataclass
+class FallbackConfig:
+    primary: str = "nim"
+    secondary: str = "cloudflare"
+    enabled: bool = True
+
+class FallbackManager:
+    def __init__(self, redis: RedisManager, config: FallbackConfig = None):
+        self.redis = redis
+        self.config = config or FallbackConfig()
+        self._http_client: Optional[httpx.AsyncClient] = None
+    
+    async def init_client(self):
+        if not self._http_client:
+            self._http_client = httpx.AsyncClient(
+                limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
+                timeout=httpx.Timeout(REQUEST_TIMEOUT),
             )
+    
+    async def close_client(self):
+        if self._http_client:
+            await self._http_client.aclose()
+    
+    async def get_active_provider(self) -> tuple[str, dict]:
+        if not self.config.enabled:
+            return self.config.primary, self._get_provider_config(self.config.primary)
         
-        await self.redis.set_circuit_state(self.provider, state)
+        primary_breaker = CircuitBreaker(self.redis, self.config.primary)
+        if await primary_breaker.can_execute():
+            return self.config.primary, self._get_provider_config(self.config.primary)
+        
+        secondary_breaker = CircuitBreaker(self.redis, self.config.secondary)
+        if await secondary_breaker.can_execute():
+            FALLBACK_COUNT.labels(
+                from_provider=self.config.primary,
+                to_provider=self.config.secondary,
+                reason="circuit_open",
+            ).inc()
+            logger.warning(f"Fallback: {self.config.primary} unavailable, switching to {self.config.secondary}")
+            return self.config.secondary, self._get_provider_config(self.config.secondary)
+        
+        if MLX_ENABLED:
+            mlx_breaker = CircuitBreaker(self.redis, "mlx")
+            if await mlx_breaker.can_execute():
+                FALLBACK_COUNT.labels(
+                    from_provider=self.config.primary,
+                    to_provider="mlx",
+                    reason="both_down",
+                ).inc()
+                logger.warning("Both cloud providers down — falling back to MLX local")
+                return "mlx", self._get_provider_config("mlx")
+        
+        raise HTTPException(status_code=503, detail="All LLM providers unavailable.")
+    
+    def _get_provider_config(self, provider: str) -> dict:
+        configs = {
+            "nim": {
+                "api_base": NIM_API_BASE,
+                "api_key": os.environ.get("NVIDIA_API_KEY", "no-key"),
+                "rpm_limit": 40,
+                "cost_per_1m_input": 0.0,
+                "cost_per_1m_output": 0.0,
+            },
+            "cloudflare": {
+                "api_base": f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/v1",
+                "api_key": CLOUDFLARE_API_KEY,
+                "rpm_limit": 100,
+                "cost_per_1m_input": 0.0,
+                "cost_per_1m_output": 0.0,
+            },
+            "mlx": {
+                "api_base": MLX_API_BASE,
+                "api_key": "no-key",
+                "rpm_limit": 1000,
+                "cost_per_1m_input": 0.0,
+                "cost_per_1m_output": 0.0,
+            },
+        }
+        return configs.get(provider, configs["nim"])
 
 # ---------------------------------------------------------------------------
 # Cost Tracking
@@ -306,8 +386,6 @@ class CircuitBreaker:
 
 @dataclass
 class CostTracker:
-    """Per-session cost tracking with budget enforcement."""
-    
     session_id: str
     budget_usd: float = BUDGET_USD_PER_SESSION
     spent_usd: float = 0.0
@@ -320,18 +398,13 @@ class CostTracker:
     def record_spend(self, cost_usd: float):
         self.spent_usd += cost_usd
         LLM_COST_USD.labels(provider=self.provider, model=self.model).inc(cost_usd)
-        logger.info(
-            f"Session {self.session_id}: spent ${cost_usd:.4f}, "
-            f"total ${self.spent_usd:.4f} / ${self.budget_usd:.2f}"
-        )
+        logger.info(f"Session {self.session_id}: spent ${cost_usd:.4f}, total ${self.spent_usd:.4f} / ${self.budget_usd:.2f}")
 
 # ---------------------------------------------------------------------------
-# Semaphore for Concurrency Control
+# Concurrency Limiter
 # ---------------------------------------------------------------------------
 
 class ConcurrencyLimiter:
-    """Global concurrent request limiter."""
-    
     def __init__(self, max_concurrent: int):
         self.semaphore = asyncio.Semaphore(max_concurrent)
     
@@ -346,7 +419,7 @@ class ConcurrencyLimiter:
 # ---------------------------------------------------------------------------
 
 class ChatRequest(BaseModel):
-    model: str = Field(..., description="Model ID (e.g., nim/llama-3-8b)")
+    model: str = Field(..., description="Model ID (e.g., @cf/meta/llama-3.1-8b-instruct)")
     messages: list[dict] = Field(..., description="OpenAI-compatible messages")
     temperature: Optional[float] = 0.7
     max_tokens: Optional[int] = 4096
@@ -354,18 +427,20 @@ class ChatRequest(BaseModel):
     tools: Optional[list[dict]] = None
     tool_choice: Optional[str] = "auto"
     session_id: Optional[str] = None
-    api_key: Optional[str] = None  # Provider-specific API key override
+    provider_override: Optional[str] = None
 
 class ChatResponse(BaseModel):
     id: str
     session_id: str
     model: str
+    provider: str
     content: Optional[str] = None
     tool_calls: Optional[list[dict]] = None
     usage: dict = Field(default_factory=dict)
     cost_usd: float = 0.0
     cached: bool = False
     finish_reason: Optional[str] = None
+    fallback_used: bool = False
 
 class HealthResponse(BaseModel):
     status: str
@@ -375,17 +450,16 @@ class HealthResponse(BaseModel):
     redis_connected: bool
     db_connected: bool
     circuit_breakers: dict[str, str]
-
-class MetricsResponse(BaseModel):
-    prometheus: str
+    fallback_status: dict[str, str]
 
 # ---------------------------------------------------------------------------
-# Global State (set during lifespan)
+# Global State
 # ---------------------------------------------------------------------------
 
 db_pool: Optional[DatabasePool] = None
 redis_manager: Optional[RedisManager] = None
 concurrency_limiter: Optional[ConcurrencyLimiter] = None
+fallback_manager: Optional[FallbackManager] = None
 start_time: float = 0.0
 shutdown_event: asyncio.Event = asyncio.Event()
 
@@ -395,12 +469,10 @@ shutdown_event: asyncio.Event = asyncio.Event()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Application lifespan manager."""
-    global db_pool, redis_manager, concurrency_limiter, start_time
+    global db_pool, redis_manager, concurrency_limiter, fallback_manager, start_time
     
     start_time = time.time()
     
-    # Initialize connections
     db_pool = DatabasePool(DATABASE_URL)
     await db_pool.connect()
     
@@ -408,23 +480,25 @@ async def lifespan(app: FastAPI):
     await redis_manager.connect()
     
     concurrency_limiter = ConcurrencyLimiter(MAX_CONCURRENT_REQUESTS)
+    fallback_manager = FallbackManager(redis_manager)
+    await fallback_manager.init_client()
     
-    # Graceful shutdown handler
     loop = asyncio.get_event_loop()
     for sig in (signal.SIGTERM, signal.SIGINT):
         loop.add_signal_handler(sig, lambda: asyncio.create_task(_shutdown()))
     
-    # Initialize database schema
-    await _init_schema()
+    if DATABASE_URL:
+        await _init_schema()
     
     logger.info("ml-intern production server started")
     
     yield
     
-    # Shutdown
     logger.info("Shutting down...")
     shutdown_event.set()
     
+    if fallback_manager:
+        await fallback_manager.close_client()
     if redis_manager:
         await redis_manager.disconnect()
     if db_pool:
@@ -437,7 +511,6 @@ async def _shutdown():
     shutdown_event.set()
 
 async def _init_schema():
-    """Initialize database schema if not exists."""
     await db_pool.execute("""
         CREATE TABLE IF NOT EXISTS sessions (
             id TEXT PRIMARY KEY,
@@ -451,7 +524,7 @@ async def _init_schema():
     await db_pool.execute("""
         CREATE TABLE IF NOT EXISTS requests (
             id TEXT PRIMARY KEY,
-            session_id TEXT REFERENCES sessions(id),
+            session_id TEXT,
             model TEXT,
             provider TEXT,
             input_tokens INTEGER,
@@ -459,15 +532,7 @@ async def _init_schema():
             cost_usd NUMERIC,
             latency_ms INTEGER,
             cached BOOLEAN DEFAULT FALSE,
-            created_at TIMESTAMP DEFAULT NOW()
-        )
-    """)
-    await db_pool.execute("""
-        CREATE TABLE IF NOT EXISTS circuit_events (
-            id SERIAL PRIMARY KEY,
-            provider TEXT,
-            event_type TEXT,
-            details JSONB,
+            fallback_used BOOLEAN DEFAULT FALSE,
             created_at TIMESTAMP DEFAULT NOW()
         )
     """)
@@ -475,7 +540,7 @@ async def _init_schema():
 
 app = FastAPI(
     title="ml-intern Production API",
-    description="Production-grade API for ml-intern with rate limiting, caching, and multi-tenancy",
+    description="Production-grade API with NIM/Cloudflare fallback and MLX local support",
     version="1.0.0",
     lifespan=lifespan,
 )
@@ -495,11 +560,9 @@ app.add_middleware(
 
 @app.middleware("http")
 async def correlation_id_middleware(request: Request, call_next):
-    """Add correlation ID to all requests."""
     correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
     request.state.correlation_id = correlation_id
     
-    # Set correlation ID in logger adapter
     old_factory = logging.getLogRecordFactory()
     def record_factory(*args, **kwargs):
         record = old_factory(*args, **kwargs)
@@ -531,58 +594,12 @@ async def correlation_id_middleware(request: Request, call_next):
 # Helper Functions
 # ---------------------------------------------------------------------------
 
-def get_provider_from_model(model: str) -> tuple[str, str]:
-    """Extract provider and model name from model string."""
-    if model.startswith("anthropic/"):
-        return "anthropic", model
-    elif model.startswith("openai/"):
-        return "openai", model
-    elif model.startswith("nim/"):
-        return "nim", model.replace("nim/", "")
-    elif model.startswith("ollama/"):
-        return "ollama", model.replace("ollama/", "")
-    elif model.startswith("groq/"):
-        return "groq", model.replace("groq/", "")
-    elif model.startswith("vllm/"):
-        return "vllm", model.replace("vllm/", "")
-    elif model.startswith("llamacpp/"):
-        return "llamacpp", model.replace("llamacpp/", "")
-    elif model.startswith("lmstudio/"):
-        return "lmstudio", model.replace("lmstudio/", "")
-    elif model.startswith("mlx/"):
-        return "mlx", model.replace("mlx/", "")
-    elif model.startswith("tgi/"):
-        return "tgi", model.replace("tgi/", "")
-    elif model.startswith("local/"):
-        return "local", model.replace("local/", "")
-    else:
-        return "huggingface", model
-
-def estimate_cost(provider: str, model: str, input_tokens: int, output_tokens: int) -> float:
-    """Estimate cost in USD based on provider pricing."""
-    # Pricing per 1M tokens (rough estimates)
-    pricing = {
-        "anthropic": {"input": 15.0, "output": 75.0},  # Claude Opus 4
-        "openai": {"input": 2.5, "output": 10.0},  # GPT-4o
-        "groq": {"input": 0.0, "output": 0.0},  # Free tier
-        "nim": {"input": 0.0, "output": 0.0},  # Free tier
-        "huggingface": {"input": 0.0, "output": 0.0},  # Free credits
-        "ollama": {"input": 0.0, "output": 0.0},  # Local
-        "llamacpp": {"input": 0.0, "output": 0.0},  # Local
-        "lmstudio": {"input": 0.0, "output": 0.0},  # Local
-        "vllm": {"input": 0.0, "output": 0.0},  # Local
-        "mlx": {"input": 0.0, "output": 0.0},  # Local
-        "tgi": {"input": 0.0, "output": 0.0},  # Local
-        "local": {"input": 0.0, "output": 0.0},  # Local
-    }
-    
-    p = pricing.get(provider, {"input": 0.0, "output": 0.0})
-    cost = (input_tokens / 1_000_000) * p["input"] + (output_tokens / 1_000_000) * p["output"]
+def estimate_cost(provider_config: dict, input_tokens: int, output_tokens: int) -> float:
+    cost = (input_tokens / 1_000_000) * provider_config.get("cost_per_1m_input", 0.0)
+    cost += (output_tokens / 1_000_000) * provider_config.get("cost_per_1m_output", 0.0)
     return cost
 
 def generate_cache_key(request: ChatRequest) -> str:
-    """Generate deterministic cache key from request."""
-    # Hash of messages + model + temperature (exclude stream)
     content = json.dumps({
         "model": request.model,
         "messages": request.messages,
@@ -592,13 +609,93 @@ def generate_cache_key(request: ChatRequest) -> str:
     }, sort_keys=True)
     return f"cache:llm:{hashlib.sha256(content.encode()).hexdigest()}"
 
+# ---------------------------------------------------------------------------
+# LLM Call Implementation
+# ---------------------------------------------------------------------------
+
+async def call_llm(
+    provider: str,
+    provider_config: dict,
+    request: ChatRequest,
+    session_id: str,
+) -> ChatResponse:
+    if not fallback_manager or not fallback_manager._http_client:
+        raise HTTPException(status_code=503, detail="HTTP client not initialized")
+    
+    api_base = provider_config["api_base"]
+    api_key = provider_config["api_key"]
+    
+    payload = {
+        "model": request.model,
+        "messages": request.messages,
+        "temperature": request.temperature,
+        "max_tokens": request.max_tokens,
+        "stream": False,
+    }
+    
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}",
+    }
+    
+    if provider == "cloudflare":
+        headers["Authorization"] = f"Bearer {api_key}"
+    
+    start_time = time.time()
+    
+    try:
+        response = await fallback_manager._http_client.post(
+            f"{api_base}/chat/completions",
+            json=payload,
+            headers=headers,
+            timeout=REQUEST_TIMEOUT,
+        )
+        response.raise_for_status()
+        data = response.json()
+        
+        latency_ms = int((time.time() - start_time) * 1000)
+        
+        usage = data.get("usage", {})
+        input_tokens = usage.get("prompt_tokens", 0)
+        output_tokens = usage.get("completion_tokens", 0)
+        
+        cost = estimate_cost(provider_config, input_tokens, output_tokens)
+        
+        content = None
+        tool_calls = None
+        if "choices" in data and len(data["choices"]) > 0:
+            choice = data["choices"][0]
+            message = choice.get("message", {})
+            content = message.get("content")
+            tool_calls = message.get("tool_calls")
+        
+        return ChatResponse(
+            id=data.get("id", str(uuid.uuid4())),
+            session_id=session_id,
+            model=request.model,
+            provider=provider,
+            content=content,
+            tool_calls=tool_calls,
+            usage=usage,
+            cost_usd=cost,
+            cached=False,
+            finish_reason=data.get("choices", [{}])[0].get("finish_reason"),
+            fallback_used=False,
+        )
+        
+    except httpx.HTTPStatusError as e:
+        logger.error(f"HTTP error from {provider}: {e.response.status_code} - {e.response.text[:200]}")
+        raise HTTPException(status_code=502, detail=f"Provider {provider} returned HTTP {e.response.status_code}")
+    except httpx.RequestError as e:
+        logger.error(f"Network error calling {provider}: {e}")
+        raise HTTPException(status_code=503, detail=f"Cannot reach provider {provider}: {str(e)}")
+
 # ---------------------------------------------------------------------------
 # API Endpoints
 # ---------------------------------------------------------------------------
 
 @app.get("/health", response_model=HealthResponse)
 async def health_check():
-    """Health check endpoint for load balancers and monitoring."""
     uptime = time.time() - start_time
     
     redis_ok = False
@@ -609,81 +706,70 @@ async def health_check():
         pass
     
     db_ok = False
-    try:
-        await db_pool.fetchval("SELECT 1")
-        db_ok = True
-    except Exception:
-        pass
+    if DATABASE_URL:
+        try:
+            await db_pool.fetchval("SELECT 1")
+            db_ok = True
+        except Exception:
+            pass
     
-    # Get circuit breaker states
     circuits = {}
-    for provider in ["anthropic", "openai", "groq", "nim", "huggingface", "ollama"]:
+    fallback_status = {}
+    for provider in ["nim", "cloudflare", "mlx"]:
         try:
             state = await redis_manager.get_circuit_state(provider)
             circuits[provider] = state["state"]
+            fallback_status[provider] = "up" if state["state"] == "closed" else "down"
         except Exception:
             circuits[provider] = "unknown"
+            fallback_status[provider] = "unknown"
     
     return HealthResponse(
-        status="healthy" if redis_ok and db_ok else "degraded",
+        status="healthy" if redis_ok else "degraded",
+        version="1.0.0",
         uptime_seconds=uptime,
-        active_sessions=0,  # Would query from DB
+        active_sessions=0,
         redis_connected=redis_ok,
         db_connected=db_ok,
         circuit_breakers=circuits,
+        fallback_status=fallback_status,
     )
 
 @app.get("/metrics")
 async def metrics():
-    """Prometheus metrics endpoint."""
     from starlette.responses import Response
     return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)
 
 @app.post("/v1/chat/completions", response_model=ChatResponse)
 async def chat_completions(request: ChatRequest, background_tasks: BackgroundTasks):
-    """OpenAI-compatible chat completions endpoint with production features."""
-    
     correlation_id = getattr(request.state, "correlation_id", str(uuid.uuid4()))
     session_id = request.session_id or str(uuid.uuid4())
-    provider, model_name = get_provider_from_model(request.model)
-    request.state.provider = provider
     
-    logger.info(
-        f"Chat request: provider={provider}, model={model_name}, "
-        f"stream={request.stream}, session={session_id}"
-    )
+    logger.info(f"Chat request: model={request.model}, stream={request.stream}, session={session_id}")
     
-    # 1. Concurrency limit
     await concurrency_limiter.acquire()
     try:
-        # 2. Circuit breaker check
-        breaker = CircuitBreaker(redis_manager, provider)
-        if not await breaker.can_execute():
-            logger.warning(f"Circuit breaker OPEN for {provider}")
-            raise HTTPException(
-                status_code=503,
-                detail=f"Service temporarily unavailable for provider {provider}. "
-                       f"Circuit breaker is open. Try again later."
-            )
+        # 1. Determine provider
+        if request.provider_override:
+            provider = request.provider_override
+            provider_config = fallback_manager._get_provider_config(provider)
+            breaker = CircuitBreaker(redis_manager, provider)
+            if not await breaker.can_execute():
+                raise HTTPException(status_code=503, detail=f"Provider {provider} circuit breaker is open")
+        else:
+            provider, provider_config = await fallback_manager.get_active_provider()
         
-        # 3. Rate limiting
-        rpm = DEFAULT_RPM_LIMIT
-        if provider == "nim":
-            rpm = 40
-        elif provider == "groq":
-            rpm = 30
+        request.state.provider = provider
         
+        # 2. Rate limiting
+        rpm = provider_config.get("rpm_limit", DEFAULT_RPM_LIMIT)
         rate_limit_key = f"{provider}:{session_id}"
         allowed, retry_after = await redis_manager.check_rate_limit(rate_limit_key, rpm)
         if not allowed:
             logger.warning(f"Rate limit exceeded for {rate_limit_key}")
-            raise HTTPException(
-                status_code=429,
-                detail=f"Rate limit exceeded. Retry after {retry_after:.1f}s",
-                headers={"Retry-After": str(int(retry_after))},
-            )
+            raise HTTPException(status_code=429, detail=f"Rate limit exceeded. Retry after {retry_after:.1f}s", headers={"Retry-After": str(int(retry_after))})
         
-        # 4. Check cache for non-streaming requests
+        # 3. Check cache
         if not request.stream:
             cache_key = generate_cache_key(request)
             cached = await redis_manager.get_cache(cache_key)
@@ -694,49 +780,29 @@ async def chat_completions(request: ChatRequest, background_tasks: BackgroundTas
                     id=str(uuid.uuid4()),
                     session_id=session_id,
                     model=request.model,
+                    provider=provider,
                     content=data.get("content"),
                     tool_calls=data.get("tool_calls"),
                     usage=data.get("usage", {}),
                     cost_usd=0.0,
                     cached=True,
                     finish_reason=data.get("finish_reason"),
+                    fallback_used=False,
                 )
         
-        # 5. Budget check
-        # TODO: Get session budget from DB
-        cost_tracker = CostTracker(session_id, provider=provider, model=model_name)
-        
-        # 6. Call LLM (placeholder - would integrate with actual ml-intern agent)
-        # For now, return a mock response with proper structure
-        response_id = str(uuid.uuid4())
-        
-        # Simulate LLM call
-        input_tokens = sum(len(m.get("content", "").split()) for m in request.messages) * 1.3
-        output_tokens = request.max_tokens or 1000
+        # 4. Budget
+        cost_tracker = CostTracker(session_id, provider=provider, model=request.model)
         
-        cost = estimate_cost(provider, model_name, int(input_tokens), output_tokens)
-        cost_tracker.record_spend(cost)
+        # 5. Call LLM
+        response = await call_llm(provider, provider_config, request, session_id)
         
-        # Record success in circuit breaker
+        # Record success
+        breaker = CircuitBreaker(redis_manager, provider)
         await breaker.record_success()
         
-        # Build response
-        response = ChatResponse(
-            id=response_id,
-            session_id=session_id,
-            model=request.model,
-            content="This is a production-grade response from ml-intern.",
-            usage={
-                "prompt_tokens": int(input_tokens),
-                "completion_tokens": output_tokens,
-                "total_tokens": int(input_tokens) + output_tokens,
-            },
-            cost_usd=cost,
-            cached=False,
-            finish_reason="stop",
-        )
+        cost_tracker.record_spend(response.cost_usd)
         
-        # 7. Cache response
+        # 6. Cache response
         if not request.stream:
             cache_key = generate_cache_key(request)
             await redis_manager.set_cache(
@@ -749,8 +815,13 @@ async def chat_completions(request: ChatRequest, background_tasks: BackgroundTas
                 }),
             )
         
-        # 8. Persist to database (background)
-        background_tasks.add_task(_persist_request, session_id, request, response)
+        # 7. Persist
+        if DATABASE_URL:
+            background_tasks.add_task(_persist_request, session_id, request, response, provider)
+        
+        # Mark fallback
+        if provider != FALLBACK_PRIMARY and FALLBACK_ENABLED:
+            response.fallback_used = True
         
         return response
         
@@ -758,72 +829,90 @@ async def chat_completions(request: ChatRequest, background_tasks: BackgroundTas
         raise
     except Exception as e:
         logger.exception(f"Error processing request: {e}")
-        # Record failure in circuit breaker
-        breaker = CircuitBreaker(redis_manager, provider)
+        breaker = CircuitBreaker(redis_manager, provider if 'provider' in locals() else "unknown")
         await breaker.record_failure()
         raise HTTPException(status_code=500, detail=str(e))
     finally:
         concurrency_limiter.release()
 
-async def _persist_request(session_id: str, request: ChatRequest, response: ChatResponse):
-    """Persist request/response to database (background task)."""
+async def _persist_request(session_id: str, request: ChatRequest, response: ChatResponse, provider: str):
     try:
         await db_pool.execute(
             """
             INSERT INTO requests (id, session_id, model, provider, input_tokens, 
-                                  output_tokens, cost_usd, latency_ms, cached)
-            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+                                  output_tokens, cost_usd, latency_ms, cached, fallback_used)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
             """,
             response.id,
             session_id,
             request.model,
-            get_provider_from_model(request.model)[0],
+            provider,
             response.usage.get("prompt_tokens", 0),
             response.usage.get("completion_tokens", 0),
             response.cost_usd,
-            0,  # latency would be measured
+            0,
             response.cached,
+            response.fallback_used,
         )
     except Exception as e:
         logger.error(f"Failed to persist request: {e}")
 
 @app.get("/v1/models")
 async def list_models():
-    """List available models."""
-    return {
-        "object": "list",
-        "data": [
-            {"id": "anthropic/claude-opus-4-6", "object": "model", "owned_by": "anthropic"},
-            {"id": "anthropic/claude-opus-4-7", "object": "model", "owned_by": "anthropic"},
-            {"id": "openai/gpt-5.5", "object": "model", "owned_by": "openai"},
-            {"id": "openai/gpt-5.4", "object": "model", "owned_by": "openai"},
-            {"id": "groq/llama-3.3-70b-versatile", "object": "model", "owned_by": "groq"},
-            {"id": "groq/llama-3.1-8b-instant", "object": "model", "owned_by": "groq"},
-            {"id": "nim/llama-3-8b", "object": "model", "owned_by": "nvidia"},
+    models = []
+    
+    if os.environ.get("NVIDIA_API_KEY"):
+        models.extend([
             {"id": "nim/llama-3.1-405b-instruct", "object": "model", "owned_by": "nvidia"},
-            {"id": "ollama/llama3.1", "object": "model", "owned_by": "ollama"},
-            {"id": "vllm/llama-3-8b", "object": "model", "owned_by": "vllm"},
-            {"id": "llamacpp/llama-3-8b", "object": "model", "owned_by": "llamacpp"},
-            {"id": "lmstudio/llama-3-8b", "object": "model", "owned_by": "lmstudio"},
-            {"id": "mlx/llama-3-8b", "object": "model", "owned_by": "mlx"},
-            {"id": "tgi/llama-3-8b", "object": "model", "owned_by": "tgi"},
-            {"id": "local/llama-3-8b", "object": "model", "owned_by": "local"},
-        ],
+            {"id": "nim/llama-3.1-70b-instruct", "object": "model", "owned_by": "nvidia"},
+            {"id": "nim/llama-3.1-8b-instruct", "object": "model", "owned_by": "nvidia"},
+            {"id": "nim/mistral-7b-instruct", "object": "model", "owned_by": "nvidia"},
+        ])
+    
+    if CLOUDFLARE_API_KEY and CLOUDFLARE_ACCOUNT_ID:
+        models.extend([
+            {"id": "cloudflare/@cf/meta/llama-3.1-8b-instruct", "object": "model", "owned_by": "cloudflare"},
+            {"id": "cloudflare/@cf/meta/llama-3.1-70b-instruct", "object": "model", "owned_by": "cloudflare"},
+            {"id": "cloudflare/@cf/mistral/mistral-7b-instruct", "object": "model", "owned_by": "cloudflare"},
+            {"id": "cloudflare/@cf/qwen/qwen1.5-14b-chat-awq", "object": "model", "owned_by": "cloudflare"},
+        ])
+    
+    if MLX_ENABLED:
+        models.extend([
+            {"id": "mlx/llama-3.1-8b", "object": "model", "owned_by": "mlx"},
+            {"id": "mlx/llama-3.1-70b", "object": "model", "owned_by": "mlx"},
+        ])
+    
+    return {"object": "list", "data": models}
+
+@app.get("/v1/fallback/status")
+async def fallback_status():
+    status = {}
+    for provider in ["nim", "cloudflare", "mlx"]:
+        breaker = CircuitBreaker(redis_manager, provider)
+        can_execute = await breaker.can_execute()
+        state = await redis_manager.get_circuit_state(provider)
+        status[provider] = {
+            "circuit_state": state["state"],
+            "failures": state["failures"],
+            "available": can_execute,
+            "last_failure": state["last_failure"],
+        }
+    
+    return {
+        "fallback_enabled": FALLBACK_ENABLED,
+        "primary": FALLBACK_PRIMARY,
+        "secondary": FALLBACK_SECONDARY,
+        "providers": status,
+        "active_provider": await _get_active_provider_name(),
     }
 
-@app.delete("/v1/sessions/{session_id}")
-async def delete_session(session_id: str):
-    """Delete a session and all its data."""
-    # Clear cache entries for this session
-    pattern = f"ratelimit:*:{session_id}"
-    # Note: In production, use SCAN instead of KEYS
-    
-    await db_pool.execute(
-        "UPDATE sessions SET metadata = jsonb_set(metadata, '{deleted}', 'true') WHERE id = $1",
-        session_id,
-    )
-    
-    return {"deleted": True, "session_id": session_id}
+async def _get_active_provider_name() -> str:
+    try:
+        provider, _ = await fallback_manager.get_active_provider()
+        return provider
+    except HTTPException:
+        return "none_available"
 
 # ---------------------------------------------------------------------------
 # Main Entry Point