Spaces:
Paused
Paused
File size: 7,593 Bytes
9f6ffb8 83dc8bf 9825db3 9f6ffb8 b384948 9f6ffb8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """
FastAPI Serving Layer for Voice-Enabled Indic RAG System.
Endpoints:
- POST /query: Accepts audio file upload (multipart/form-data) OR JSON text bypass. Returns QueryResponse.
- GET /health: Healthcheck reporting active languages, index vector counts, and model status.
- GET /languages: Returns active languages strictly derived from config.LANGUAGES.
- GET /: Web Demo UI.
"""
import asyncio
import logging
import os
import shutil
import tempfile
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, Dict, Optional
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
import config
from rag_pipeline.orchestrator import get_orchestrator, RAGPipelineOrchestrator
from rag_pipeline.schemas import QueryRequest, QueryResponse
from vector_search.embed import get_embedder
from vector_search import get_index_manager
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Lifespan event handler: Preload embedding models, FAISS indexes, and
eagerly initialize and warm up the full RAG pipeline orchestrator ONCE at startup.
Never re-loads per request. Self-heals if indexes are missing or empty.
"""
print("[API Lifespan] Initializing and pre-loading embedding model...")
embedder = get_embedder()
print("[API Lifespan] Loading FAISS HNSW indexes and corpus centroids into memory...")
index_mgr = get_index_manager()
# Ensure indexes are populated
for name, idx in index_mgr.indexes.items():
print(f"[API Lifespan] Index '{name}': {idx.index.ntotal} vectors loaded (configured cap: {config.MAX_INDEX_PASSAGES_PER_LANG}).")
print("[API Lifespan] Eagerly initializing and warming up RAG orchestrator & ONNX models...")
orchestrator = get_orchestrator()
print("[API Lifespan] RAG orchestrator warmup complete.")
print(f"[API Lifespan] Initialized successfully. Active languages: {config.LANGUAGES}")
print(f"[API Lifespan] Allow Network Calls Switch: {config.ALLOW_NETWORK_CALLS_IN_PIPELINE}")
print(f"[API Lifespan] Request Timeout Deadline: {config.REQUEST_TIMEOUT_SECONDS}s")
yield
print("[API Lifespan] Shutting down RAG service.")
app = FastAPI(
title="⚡ VECTOR Voice Indic RAG API",
description="Instrumented low-latency Voice RAG pipeline for Indic languages (English, Hindi, Marathi)",
version="1.0.0",
lifespan=lifespan,
)
# CORS middleware for demo interface
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/health", response_class=JSONResponse)
async def health_check() -> Dict[str, Any]:
"""Health check reporting system and index readiness."""
index_mgr = get_index_manager()
index_stats = {
name: idx.index.ntotal for name, idx in index_mgr.indexes.items()
}
return {
"status": "healthy",
"configured_languages": config.LANGUAGES,
"embedding_model": config.EMBEDDING_MODEL_NAME,
"indexes_loaded": index_stats,
"centroids_available": list(index_mgr.centroids.keys()),
"allow_network_calls": config.ALLOW_NETWORK_CALLS_IN_PIPELINE,
"sarvam_stt_configured": bool(config.SARVAM_API_KEY),
"llm_fallback_configured": bool(config.LLM_API_KEY and config.ALLOW_NETWORK_CALLS_IN_PIPELINE),
"semantic_answer_cache_configured": config.SEMANTIC_ANSWER_CACHE_ENABLED,
"request_timeout_seconds": config.REQUEST_TIMEOUT_SECONDS,
"query_intent_filter_enabled": config.ENABLE_QUERY_INTENT_FILTER,
}
@app.get("/languages", response_class=JSONResponse)
async def get_supported_languages() -> Dict[str, Any]:
"""Returns metadata for all currently configured active languages."""
lang_details = [
{"code": l, **config.get_language_info(l)} for l in config.LANGUAGES
]
return {
"active_languages": config.LANGUAGES,
"language_details": lang_details,
}
@app.post("/query", response_model=QueryResponse)
async def query_pipeline(
file: Optional[UploadFile] = File(None),
text: Optional[str] = Form(None),
language_hint: Optional[str] = Form(None),
cross_lingual: Optional[bool] = Form(False),
request_body: Optional[QueryRequest] = None,
) -> QueryResponse:
"""
Execute end-to-end Voice RAG query with strict deadline timeout protection.
Accepts:
1. Multipart file upload ('file') with optional 'language_hint' and 'cross_lingual'
2. Multipart form text ('text') with optional 'language_hint' and 'cross_lingual'
3. JSON body with text / audio_path / language_hint / cross_lingual
"""
orchestrator = get_orchestrator()
temp_audio_path = None
try:
# Build query request object
req: Optional[QueryRequest] = None
# 1. Handle JSON request body
if request_body and (request_body.text or request_body.audio_path):
req = request_body
# 2. Handle Multipart audio upload
elif file and file.filename:
suffix = Path(file.filename).suffix or ".wav"
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
shutil.copyfileobj(file.file, tmp)
temp_audio_path = tmp.name
req = QueryRequest(
audio_path=temp_audio_path,
language_hint=language_hint,
cross_lingual=False if cross_lingual is None else cross_lingual,
)
# 3. Handle Multipart Form text bypass
elif text and text.strip():
req = QueryRequest(
text=text.strip(),
language_hint=language_hint,
cross_lingual=False if cross_lingual is None else cross_lingual,
)
if not req:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Either 'file' audio upload or 'text' query must be provided.",
)
# Enforce request deadline with asyncio.wait_for
try:
return await asyncio.wait_for(
orchestrator.execute(req),
timeout=config.REQUEST_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
logger.error(
f"Query execution exceeded request deadline timeout ({config.REQUEST_TIMEOUT_SECONDS}s)"
)
raise HTTPException(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
detail=f"Request execution exceeded {config.REQUEST_TIMEOUT_SECONDS}s deadline.",
)
finally:
# Clean up temporary audio file if created
if temp_audio_path and os.path.exists(temp_audio_path):
try:
os.remove(temp_audio_path)
except Exception:
pass
@app.get("/", response_class=HTMLResponse)
async def serve_demo_ui():
"""Serves the interactive voice-enabled web demo."""
demo_file = config.BASE_DIR / "demo" / "index.html"
if demo_file.exists():
with open(demo_file, "r", encoding="utf-8") as f:
return f.read()
return "<h1>Voice-Enabled Indic RAG Service</h1><p>API is running. Visit <a href='/docs'>/docs</a> for API specification.</p>"
|