from __future__ import annotations import logging import os import time from datetime import datetime from typing import Dict, List, Union from fastapi import APIRouter, BackgroundTasks, HTTPException, Header, Request from pydantic import BaseModel, Field from .analytics import Analytics from .service import ( ModelConfig, TextModelType, EmbeddingsService, ModelKind, detect_model_kind, ) logger = logging.getLogger(__name__) router = APIRouter( tags=["v1"], responses={404: {"description": "Not found"}}, ) class EmbeddingRequest(BaseModel): """ Request model for generating embeddings. """ model: str = Field( default=TextModelType.MULTILINGUAL_E5_SMALL.value, description=( "Which model ID to use? " "Text options: ['multilingual-e5-small', 'multilingual-e5-base', 'multilingual-e5-large', " "'snowflake-arctic-embed-l-v2.0', 'paraphrase-multilingual-MiniLM-L12-v2', " "'paraphrase-multilingual-mpnet-base-v2', 'bge-m3']. " "Image option: ['siglip-base-patch16-256-multilingual']." ), ) input: Union[str, List[str]] = Field( ..., description="Text(s) or image URL(s)/path(s)." ) class RankRequest(BaseModel): """ Request model for ranking candidates. """ model: str = Field( default=TextModelType.MULTILINGUAL_E5_SMALL.value, description=( "Model ID for the queries. " "Can be a text or image model (e.g. 'siglip-base-patch16-256-multilingual' for images)." ), ) queries: Union[str, List[str]] = Field( ..., description="Query text(s) or image(s) depending on the model type." ) candidates: List[str] = Field(..., description="Candidate texts to rank.") class EmbeddingResponse(BaseModel): """ Response model for embeddings. """ object: str data: List[dict] model: str usage: dict class RankResponse(BaseModel): """ Response model for ranking results. """ probabilities: List[List[float]] cosine_similarities: List[List[float]] class StatsBucket(BaseModel): """ Model for daily/weekly/monthly/yearly stats. """ total: Dict[str, int] daily: Dict[str, int] weekly: Dict[str, int] monthly: Dict[str, int] yearly: Dict[str, int] class StatsResponse(BaseModel): """ Analytics stats response model, including both access and token counts. """ access: StatsBucket tokens: StatsBucket # Initialize the embeddings service and analytics. service_config = ModelConfig() embeddings_service = EmbeddingsService(config=service_config) analytics = Analytics( url=os.environ.get("REDIS_URL", "redis://localhost:6379/0"), token=os.environ.get("REDIS_TOKEN", "***"), sync_interval=30 * 60, # 30 minutes ) # Rate limiting cache: {ip: [timestamp1, timestamp2, ...]} rate_limit_cache: Dict[str, List[float]] = {} def check_rate_limit( client_ip: str, max_requests: int = 4, window_seconds: int = 60 ) -> bool: """ Check if the client IP has exceeded the rate limit. Returns True if request is allowed, False if rate limited. """ current_time = time.time() # Clean up old entries and get current requests if client_ip in rate_limit_cache: # Remove requests older than the window rate_limit_cache[client_ip] = [ timestamp for timestamp in rate_limit_cache[client_ip] if current_time - timestamp < window_seconds ] else: rate_limit_cache[client_ip] = [] # Check if under limit if len(rate_limit_cache[client_ip]) < max_requests: # Add current request timestamp rate_limit_cache[client_ip].append(current_time) return True return False @router.post("/embeddings", response_model=EmbeddingResponse, tags=["embeddings"]) async def create_embeddings( request: EmbeddingRequest, background_tasks: BackgroundTasks, fastapi_request: Request, authorization: str = Header(None), ): """ Generate embeddings for the given text or image inputs. """ # Check authorization expected_token = os.environ.get("ACCESS_TOKEN") is_authenticated = False if expected_token: if authorization: # Support both "Bearer " and plain token formats token = authorization if authorization.startswith("Bearer "): token = authorization[7:] # Remove "Bearer " prefix if token == expected_token: is_authenticated = True # If not authenticated, check rate limit if not is_authenticated: # Get client IP client_ip = fastapi_request.client.host if hasattr(fastapi_request.headers, "get"): # Check for forwarded IP (in case of proxy) forwarded_for = fastapi_request.headers.get("X-Forwarded-For") if forwarded_for: client_ip = forwarded_for.split(",")[0].strip() real_ip = fastapi_request.headers.get("X-Real-IP") if real_ip: client_ip = real_ip.strip() # Check rate limit (4 requests per minute) if not check_rate_limit(client_ip): raise HTTPException( status_code=429, detail="Rate limit exceeded. Maximum 4 requests per minute for unauthenticated users.", ) # If no authorization header was provided when ACCESS_TOKEN is set if not authorization: raise HTTPException( status_code=401, detail="Authorization header required" ) else: raise HTTPException( status_code=401, detail="Invalid authorization token" ) try: modality = detect_model_kind(request.model) embeddings = await embeddings_service.generate_embeddings( model=request.model, inputs=request.input, ) # Estimate tokens if using a text model. total_tokens = 0 if modality == ModelKind.TEXT: total_tokens = embeddings_service.estimate_tokens(request.input) resp = { "object": "list", "data": [], "model": request.model, "usage": { "prompt_tokens": total_tokens, "total_tokens": total_tokens, }, } for idx, emb in enumerate(embeddings): resp["data"].append( { "object": "embedding", "index": idx, "embedding": emb.tolist(), } ) # Record analytics in the background. background_tasks.add_task( analytics.access, request.model, resp["usage"]["total_tokens"] ) return resp except Exception as e: msg = ( "Failed to generate embeddings. Check model ID, inputs, etc.\n" f"Details: {str(e)}" ) logger.error(msg) raise HTTPException(status_code=500, detail=msg) @router.post("/rank", response_model=RankResponse, tags=["rank"]) async def rank_candidates(request: RankRequest, background_tasks: BackgroundTasks): """ Rank candidate texts against the given queries. """ try: results = await embeddings_service.rank( model=request.model, queries=request.queries, candidates=request.candidates, ) # Record analytics in the background. background_tasks.add_task( analytics.access, request.model, results["usage"]["total_tokens"] ) return results except Exception as e: msg = ( "Failed to rank candidates. Check model ID, inputs, etc.\n" f"Details: {str(e)}" ) logger.error(msg) raise HTTPException(status_code=500, detail=msg) @router.get("/stats", response_model=StatsResponse, tags=["stats"]) async def get_stats(): """ Retrieve usage statistics for all models, including access counts and token usage. """ try: day_key = datetime.utcnow().strftime("%Y-%m-%d") week_key = f"{datetime.utcnow().year}-W{datetime.utcnow().strftime('%U')}" month_key = datetime.utcnow().strftime("%Y-%m") year_key = datetime.utcnow().strftime("%Y") stats_data = ( await analytics.stats() ) # Expected to return a dict with 'access' and 'tokens' keys return { "access": { "total": stats_data["access"].get("total", {}), "daily": stats_data["access"].get(day_key, {}), "weekly": stats_data["access"].get(week_key, {}), "monthly": stats_data["access"].get(month_key, {}), "yearly": stats_data["access"].get(year_key, {}), }, "tokens": { "total": stats_data["tokens"].get("total", {}), "daily": stats_data["tokens"].get(day_key, {}), "weekly": stats_data["tokens"].get(week_key, {}), "monthly": stats_data["tokens"].get(month_key, {}), "yearly": stats_data["tokens"].get(year_key, {}), }, } except Exception as e: msg = f"Failed to fetch analytics stats: {str(e)}" logger.error(msg) raise HTTPException(status_code=500, detail=msg)