chat-api-added requirements.txt

Dockerfile

@@ -0,0 +1,33 @@
+FROM python:3.10-slim-bullseye
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies first (including curl)
+RUN apt-get update && apt-get install -y \
+    gcc \
+    g++ \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements file first to leverage Docker cache
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Set environment variables
+ENV HF_HOME=/tmp/cache
+ENV PORT=7860
+
+# Create cache directory (if still needed)
+RUN mkdir -p ${HF_HOME} && chmod 777 ${HF_HOME}
+
+# Expose port
+EXPOSE $PORT
+
+# Command to run the FastAPI app
+CMD bash -c "while true; do curl -s https://xce009-inference-test.hf.space/ping >/dev/null && sleep 300 || sleep 300; done & uvicorn chat_api:app --host 0.0.0.0 --port $PORT"

README.md

@@ -0,0 +1,9 @@
+---
+title: My Inference Space
+emoji: 🚀
+colorFrom: blue
+colorTo: purple
+sdk: docker
+app_file: main.py
+pinned: false
+---

chat_api.py

@@ -0,0 +1,540 @@
+import asyncio
+import aiohttp
+import logging
+import os
+import random
+from typing import List, Dict, Optional, Any
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi.responses import StreamingResponse, JSONResponse
+from pydantic import BaseModel, Field
+import uvicorn
+from datetime import datetime
+import json
+import time
+from collections import defaultdict, deque
+import threading
+from contextlib import asynccontextmanager
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+# Pydantic models
+class Message(BaseModel):
+    role: str = Field(..., description="Role: 'system', 'user', or 'assistant'")
+    content: str = Field(..., description="Message content")
+
+class ProviderPreferences(BaseModel):
+    sort: Optional[str] = Field(None, description="Sort by 'price', 'throughput', or 'latency'")
+    order: Optional[List[str]] = Field(None, description="Specific provider order")
+    allow_fallbacks: Optional[bool] = Field(True, description="Allow fallback providers")
+    require_parameters: Optional[bool] = Field(False, description="Require all parameters support")
+    data_collection: Optional[str] = Field("allow", description="'allow' or 'deny' data collection")
+    only: Optional[List[str]] = Field(None, description="Only use these providers")
+    ignore: Optional[List[str]] = Field(None, description="Ignore these providers")
+    quantizations: Optional[List[str]] = Field(None, description="Required quantization levels")
+    max_price: Optional[Dict[str, float]] = Field(None, description="Maximum pricing constraints")
+
+class ChatRequest(BaseModel):
+    model: str = Field(..., description="Model ID (e.g., 'openai/gpt-3.5-turbo')")
+    messages: List[Message] = Field(..., description="List of messages")
+    system_prompt: Optional[str] = Field(None, description="System prompt (will be added as system message)")
+    max_tokens: Optional[int] = Field(1000, description="Maximum tokens to generate")
+    temperature: Optional[float] = Field(0.7, description="Temperature (0-2)")
+    top_p: Optional[float] = Field(1.0, description="Top-p sampling")
+    frequency_penalty: Optional[float] = Field(0.0, description="Frequency penalty")
+    presence_penalty: Optional[float] = Field(0.0, description="Presence penalty")
+    stream: Optional[bool] = Field(False, description="Enable streaming response")
+    provider: Optional[ProviderPreferences] = Field(None, description="Provider routing preferences")
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "model": "openai/gpt-3.5-turbo",
+                "messages": [
+                    {"role": "user", "content": "Hello, how are you?"}
+                ],
+                "system_prompt": "You are a helpful assistant.",
+                "max_tokens": 1000,
+                "temperature": 0.7,
+                "stream": False
+            }
+        }
+
+class ChatResponse(BaseModel):
+    success: bool
+    model: str
+    choices: List[Dict[str, Any]]
+    usage: Optional[Dict[str, Any]]
+    response_time: float
+    provider_used: Optional[str] = None
+    timestamp: str
+
+class APIKeyManager:
+    """Manages multiple API keys with rotation and rate limiting"""
+    
+    def __init__(self, api_keys: List[str]):
+        if not api_keys:
+            raise ValueError("At least one API key is required")
+        
+        self.api_keys = api_keys
+        self.key_stats = {key: {"requests": 0, "errors": 0, "last_used": 0} for key in api_keys}
+        self.current_index = 0
+        self.lock = threading.Lock()
+        
+        # Rate limiting per key (rough estimate)
+        self.rate_limits = {key: deque() for key in api_keys}
+        self.max_requests_per_minute = 60  # Conservative estimate
+        
+        logger.info(f"Initialized API key manager with {len(api_keys)} keys")
+    
+    def get_next_key(self) -> str:
+        """Get the next available API key using round-robin with rate limiting"""
+        with self.lock:
+            current_time = time.time()
+            
+            # Try to find a key that's not rate limited
+            for _ in range(len(self.api_keys)):
+                key = self.api_keys[self.current_index]
+                
+                # Clean old requests from rate limit tracker
+                while (self.rate_limits[key] and 
+                       current_time - self.rate_limits[key][0] > 60):
+                    self.rate_limits[key].popleft()
+                
+                # Check if this key can handle more requests
+                if len(self.rate_limits[key]) < self.max_requests_per_minute:
+                    self.rate_limits[key].append(current_time)
+                    self.key_stats[key]["requests"] += 1
+                    self.key_stats[key]["last_used"] = current_time
+                    
+                    # Move to next key for next request
+                    self.current_index = (self.current_index + 1) % len(self.api_keys)
+                    return key
+                
+                # Try next key
+                self.current_index = (self.current_index + 1) % len(self.api_keys)
+            
+            # If all keys are rate limited, use the one with the oldest request
+            oldest_key = min(self.api_keys, 
+                           key=lambda k: self.key_stats[k]["last_used"])
+            self.key_stats[oldest_key]["requests"] += 1
+            self.key_stats[oldest_key]["last_used"] = current_time
+            return oldest_key
+    
+    def record_error(self, api_key: str):
+        """Record an error for an API key"""
+        with self.lock:
+            if api_key in self.key_stats:
+                self.key_stats[api_key]["errors"] += 1
+    
+    def get_stats(self) -> Dict:
+        """Get statistics for all API keys"""
+        with self.lock:
+            return dict(self.key_stats)
+
+class OpenRouterClient:
+    """High-performance OpenRouter client with connection pooling"""
+    
+    def __init__(self, key_manager: APIKeyManager):
+        self.key_manager = key_manager
+        self.base_url = "https://openrouter.ai/api/v1"
+        self.session_pool = {}
+        self.max_connections = 100  # Total connection pool
+        self.max_connections_per_host = 20
+    
+    async def get_session(self, api_key: str) -> aiohttp.ClientSession:
+        """Get or create a session for the API key"""
+        if api_key not in self.session_pool:
+            connector = aiohttp.TCPConnector(
+                limit=self.max_connections,
+                limit_per_host=self.max_connections_per_host,
+                keepalive_timeout=30,
+                enable_cleanup_closed=True,
+                ttl_dns_cache=300,  # DNS cache TTL
+                use_dns_cache=True
+            )
+            
+            timeout = aiohttp.ClientTimeout(
+                total=60,  # Total timeout
+                connect=10,  # Connection timeout
+                sock_read=30  # Socket read timeout
+            )
+            
+            headers = {
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": "application/json",
+                "HTTP-Referer": "https://your-app.com",  # Optional: for analytics
+                "X-Title": "High-Performance Chat API"  # Optional: for analytics
+            }
+            
+            self.session_pool[api_key] = aiohttp.ClientSession(
+                connector=connector,
+                timeout=timeout,
+                headers=headers,
+                raise_for_status=False
+            )
+        
+        return self.session_pool[api_key]
+    
+    async def chat_completion(self, request: ChatRequest) -> Dict[str, Any]:
+        """Send chat completion request with automatic key rotation"""
+        start_time = time.time()
+        api_key = self.key_manager.get_next_key()
+        
+        try:
+            session = await self.get_session(api_key)
+            
+            # Prepare messages
+            messages = []
+            
+            # Add system prompt if provided
+            if request.system_prompt:
+                messages.append({"role": "system", "content": request.system_prompt})
+            
+            # Add user messages
+            messages.extend([msg.dict() for msg in request.messages])
+            
+            # Prepare request payload
+            payload = {
+                "model": request.model,
+                "messages": messages,
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature,
+                "top_p": request.top_p,
+                "frequency_penalty": request.frequency_penalty,
+                "presence_penalty": request.presence_penalty,
+                "stream": request.stream
+            }
+            
+            # Add provider preferences if specified
+            if request.provider:
+                provider_dict = request.provider.dict(exclude_none=True)
+                if provider_dict:
+                    payload["provider"] = provider_dict
+            
+            logger.debug(f"Sending request to {request.model} with key ending in ...{api_key[-4:]}")
+            
+            async with session.post(f"{self.base_url}/chat/completions", json=payload) as response:
+                response_time = time.time() - start_time
+                
+                if response.status == 200:
+                    result = await response.json()
+                    
+                    # Extract provider information if available
+                    provider_used = None
+                    if "model" in result and "/" in result["model"]:
+                        # Sometimes the response model includes provider info
+                        provider_used = result["model"].split("/")[0]
+                    
+                    return {
+                        "success": True,
+                        "data": result,
+                        "response_time": response_time,
+                        "provider_used": provider_used,
+                        "api_key_used": api_key[-4:]  # Last 4 characters for debugging
+                    }
+                else:
+                    error_data = await response.text()
+                    logger.error(f"OpenRouter API error {response.status}: {error_data}")
+                    
+                    # Record error for this key
+                    self.key_manager.record_error(api_key)
+                    
+                    return {
+                        "success": False,
+                        "error": f"API error {response.status}: {error_data}",
+                        "response_time": response_time
+                    }
+        
+        except Exception as e:
+            response_time = time.time() - start_time
+            logger.error(f"Request failed with key ...{api_key[-4:]}: {str(e)}")
+            
+            # Record error for this key
+            self.key_manager.record_error(api_key)
+            
+            return {
+                "success": False,
+                "error": str(e),
+                "response_time": response_time
+            }
+    
+    async def stream_chat_completion(self, request: ChatRequest):
+        """Stream chat completion response"""
+        api_key = self.key_manager.get_next_key()
+        
+        try:
+            session = await self.get_session(api_key)
+            
+            # Prepare messages
+            messages = []
+            if request.system_prompt:
+                messages.append({"role": "system", "content": request.system_prompt})
+            messages.extend([msg.dict() for msg in request.messages])
+            
+            # Prepare request payload
+            payload = {
+                "model": request.model,
+                "messages": messages,
+                "max_tokens": request.max_tokens,
+                "temperature": request.temperature,
+                "top_p": request.top_p,
+                "frequency_penalty": request.frequency_penalty,
+                "presence_penalty": request.presence_penalty,
+                "stream": True
+            }
+            
+            if request.provider:
+                provider_dict = request.provider.dict(exclude_none=True)
+                if provider_dict:
+                    payload["provider"] = provider_dict
+            
+            async with session.post(f"{self.base_url}/chat/completions", json=payload) as response:
+                if response.status == 200:
+                    async for chunk in response.content.iter_chunked(1024):
+                        if chunk:
+                            yield chunk
+                else:
+                    error_data = await response.text()
+                    self.key_manager.record_error(api_key)
+                    yield f"data: {json.dumps({'error': f'API error {response.status}: {error_data}'})}\n\n".encode()
+        
+        except Exception as e:
+            logger.error(f"Streaming failed with key ...{api_key[-4:]}: {str(e)}")
+            self.key_manager.record_error(api_key)
+            yield f"data: {json.dumps({'error': str(e)})}\n\n".encode()
+    
+    async def close_all_sessions(self):
+        """Close all aiohttp sessions"""
+        for session in self.session_pool.values():
+            await session.close()
+        self.session_pool.clear()
+
+# Global variables
+client: Optional[OpenRouterClient] = None
+key_manager: Optional[APIKeyManager] = None
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Startup and shutdown events"""
+    global client, key_manager
+    
+    # Startup
+    logger.info("Starting OpenRouter Chat API...")
+    
+    # Load API keys from environment
+    api_keys_str = os.getenv("OPENROUTER_API_KEYS", "")
+    if not api_keys_str:
+        raise ValueError("OPENROUTER_API_KEYS environment variable is required")
+    
+    api_keys = [key.strip() for key in api_keys_str.split(",") if key.strip()]
+    if not api_keys:
+        raise ValueError("No valid API keys found in OPENROUTER_API_KEYS")
+    
+    # Initialize key manager and client
+    key_manager = APIKeyManager(api_keys)
+    client = OpenRouterClient(key_manager)
+    
+    logger.info(f"✅ API initialized with {len(api_keys)} keys")
+    
+    yield
+    
+    # Shutdown
+    logger.info("Shutting down...")
+    if client:
+        await client.close_all_sessions()
+
+# Create FastAPI app
+app = FastAPI(
+    title="High-Performance OpenRouter Chat API",
+    description="Scalable chat completions API with multiple key rotation and parallel processing",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+@app.get("/", response_model=Dict)
+async def root():
+    """Root endpoint with API information"""
+    return {
+        "message": "High-Performance OpenRouter Chat API",
+        "version": "1.0.0",
+        "endpoints": {
+            "chat": "/api/chat",
+            "chat_stream": "/api/chat (with stream=true)",
+            "stats": "/api/stats",
+            "health": "/health"
+        },
+        "features": [
+            "Multiple API key rotation",
+            "Connection pooling",
+            "Parallel processing",
+            "Provider routing",
+            "Streaming support",
+            "Rate limiting"
+        ]
+    }
+
+@app.post("/api/chat", response_model=ChatResponse)
+async def chat_completion(request: ChatRequest):
+    """Send chat completion request"""
+    if not client:
+        raise HTTPException(status_code=503, detail="Service not initialized")
+    
+    try:
+        # Handle streaming requests
+        if request.stream:
+            return StreamingResponse(
+                client.stream_chat_completion(request),
+                media_type="text/plain",
+                headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
+            )
+        
+        # Handle regular requests
+        result = await client.chat_completion(request)
+        
+        if result["success"]:
+            return ChatResponse(
+                success=True,
+                model=request.model,
+                choices=result["data"].get("choices", []),
+                usage=result["data"].get("usage"),
+                response_time=result["response_time"],
+                provider_used=result.get("provider_used"),
+                timestamp=datetime.now().isoformat()
+            )
+        else:
+            raise HTTPException(
+                status_code=500, 
+                detail=f"Chat completion failed: {result['error']}"
+            )
+    
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Unexpected error in chat_completion: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/api/stats", response_model=Dict)
+async def get_api_stats():
+    """Get API key usage statistics"""
+    if not key_manager:
+        raise HTTPException(status_code=503, detail="Service not initialized")
+    
+    stats = key_manager.get_stats()
+    
+    # Calculate summary statistics
+    total_requests = sum(stat["requests"] for stat in stats.values())
+    total_errors = sum(stat["errors"] for stat in stats.values())
+    error_rate = (total_errors / total_requests * 100) if total_requests > 0 else 0
+    
+    return {
+        "summary": {
+            "total_keys": len(stats),
+            "total_requests": total_requests,
+            "total_errors": total_errors,
+            "error_rate_percent": round(error_rate, 2)
+        },
+        "key_stats": {
+            f"key_...{key[-4:]}": {
+                "requests": stat["requests"],
+                "errors": stat["errors"],
+                "error_rate": round((stat["errors"] / stat["requests"] * 100) if stat["requests"] > 0 else 0, 2),
+                "last_used": datetime.fromtimestamp(stat["last_used"]).isoformat() if stat["last_used"] > 0 else "Never"
+            }
+            for key, stat in stats.items()
+        }
+    }
+
+@app.get("/health")
+async def health_check():
+    """Health check endpoint"""
+    if not client or not key_manager:
+        return JSONResponse(
+            status_code=503,
+            content={
+                "status": "unhealthy",
+                "message": "Service not initialized",
+                "timestamp": datetime.now().isoformat()
+            }
+        )
+    
+    try:
+        stats = key_manager.get_stats()
+        return {
+            "status": "healthy",
+            "api_keys_loaded": len(stats),
+            "total_requests": sum(stat["requests"] for stat in stats.values()),
+            "timestamp": datetime.now().isoformat()
+        }
+    
+    except Exception as e:
+        return JSONResponse(
+            status_code=503,
+            content={
+                "status": "unhealthy",
+                "error": str(e),
+                "timestamp": datetime.now().isoformat()
+            }
+        )
+
+# Batch processing endpoint for high throughput
+@app.post("/api/chat/batch")
+async def batch_chat_completions(requests: List[ChatRequest]):
+    """Process multiple chat requests in parallel"""
+    if not client:
+        raise HTTPException(status_code=503, detail="Service not initialized")
+    
+    if len(requests) > 50:  # Limit batch size
+        raise HTTPException(status_code=400, detail="Batch size limited to 50 requests")
+    
+    try:
+        # Process all requests in parallel
+        tasks = [client.chat_completion(req) for req in requests]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        
+        # Format results
+        responses = []
+        for i, (request, result) in enumerate(zip(requests, results)):
+            if isinstance(result, Exception):
+                responses.append({
+                    "request_index": i,
+                    "success": False,
+                    "error": str(result)
+                })
+            elif result["success"]:
+                responses.append({
+                    "request_index": i,
+                    "success": True,
+                    "model": request.model,
+                    "choices": result["data"].get("choices", []),
+                    "usage": result["data"].get("usage"),
+                    "response_time": result["response_time"],
+                    "provider_used": result.get("provider_used")
+                })
+            else:
+                responses.append({
+                    "request_index": i,
+                    "success": False,
+                    "error": result["error"]
+                })
+        
+        return {
+            "success": True,
+            "batch_size": len(requests),
+            "results": responses,
+            "timestamp": datetime.now().isoformat()
+        }
+    
+    except Exception as e:
+        logger.error(f"Batch processing failed: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+