""" FastAPI server for OpenAI Realtime API integration with RAG system. Provides endpoints for session management and RAG tool calls. Directory structure: /data/ # Original PDFs, HTML /embeddings/ # FAISS, Chroma, DPR vector stores /graph/ # Graph database files /metadata/ # Image metadata (SQLite or MongoDB) """ import json import logging import os import time from typing import Dict, Any, Optional from fastapi import FastAPI, HTTPException, Request, Response, status from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError from starlette.exceptions import HTTPException as StarletteHTTPException from pydantic import BaseModel import uvicorn from openai import OpenAI # Import all query modules from query_graph import query as graph_query from query_vanilla import query as vanilla_query from query_dpr import query as dpr_query from query_bm25 import query as bm25_query from query_context import query as context_query from query_vision import query as vision_query from config import OPENAI_API_KEY, OPENAI_CHAT_MODEL, OPENAI_REALTIME_MODEL, REALTIME_VOICE, REALTIME_INSTRUCTIONS, DEFAULT_METHOD from analytics_db import log_query logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI(title="SIGHT Realtime API Server", version="1.0.0") # CORS middleware for frontend integration app.add_middleware( CORSMiddleware, allow_origins=["*"], # In production, restrict to your domain allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") async def log_requests(request: Request, call_next): """Log all incoming requests for debugging.""" logger.info(f"Incoming request: {request.method} {request.url}") try: response = await call_next(request) logger.info(f"Response status: {response.status_code}") return response except Exception as e: logger.error(f"Request processing error: {e}") return JSONResponse( content={"error": "Internal server error"}, status_code=500 ) # Exception handlers @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): logger.warning(f"Validation error for {request.url}: {exc}") return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={"error": "Invalid request format", "details": str(exc)} ) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): logger.warning(f"HTTP error for {request.url}: {exc.status_code} - {exc.detail}") return JSONResponse( status_code=exc.status_code, content={"error": exc.detail} ) @app.exception_handler(Exception) async def general_exception_handler(request: Request, exc: Exception): logger.error(f"Unhandled error for {request.url}: {exc}") return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": "Internal server error"} ) # Initialize OpenAI client client = OpenAI(api_key=OPENAI_API_KEY) # Query method dispatch QUERY_DISPATCH = { 'graph': graph_query, 'vanilla': vanilla_query, 'dpr': dpr_query, 'bm25': bm25_query, 'context': context_query, 'vision': vision_query } # Use configuration from config.py with environment variable overrides REALTIME_MODEL = os.getenv("REALTIME_MODEL", OPENAI_REALTIME_MODEL) VOICE = os.getenv("REALTIME_VOICE", REALTIME_VOICE) INSTRUCTIONS = os.getenv("REALTIME_INSTRUCTIONS", REALTIME_INSTRUCTIONS) # Pydantic models for request/response class SessionRequest(BaseModel): """Request model for creating ephemeral sessions.""" model: Optional[str] = "gpt-4o-realtime-preview" instructions: Optional[str] = None voice: Optional[str] = None class RAGRequest(BaseModel): """Request model for RAG queries.""" query: str method: str = "graph" top_k: int = 5 image_path: Optional[str] = None class RAGResponse(BaseModel): """Response model for RAG queries.""" answer: str citations: list method: str citations_html: Optional[str] = None @app.post("/session") async def create_ephemeral_session(request: SessionRequest) -> JSONResponse: """ Create an ephemeral session token for OpenAI Realtime API. This token will be used by the frontend WebRTC client. """ try: logger.info(f"Creating ephemeral session with model: {request.model or REALTIME_MODEL}") # Create ephemeral token using direct HTTP call to OpenAI API # Since the Python SDK doesn't support realtime sessions yet import requests session_data = { "model": request.model or REALTIME_MODEL, "voice": request.voice or VOICE, "modalities": ["audio", "text"], "instructions": request.instructions or INSTRUCTIONS, } headers = { "Authorization": f"Bearer {OPENAI_API_KEY}", "Content-Type": "application/json" } # Make direct HTTP request to OpenAI's realtime sessions endpoint response = requests.post( "https://api.openai.com/v1/realtime/sessions", json=session_data, headers=headers, timeout=30 ) if response.status_code == 200: session_result = response.json() response_data = { "client_secret": session_result.get("client_secret", {}).get("value") or session_result.get("client_secret"), "model": request.model or REALTIME_MODEL, "session_id": session_result.get("id") } logger.info("Ephemeral session created successfully") return JSONResponse(content=response_data, status_code=200) else: logger.error(f"OpenAI API error: {response.status_code} - {response.text}") return JSONResponse( content={"error": f"OpenAI API error: {response.status_code} - {response.text}"}, status_code=response.status_code ) except requests.exceptions.RequestException as e: logger.error(f"Network error creating ephemeral session: {e}") return JSONResponse( content={"error": f"Network error: {str(e)}"}, status_code=500 ) except Exception as e: logger.error(f"Error creating ephemeral session: {e}") return JSONResponse( content={"error": f"Session creation failed: {str(e)}"}, status_code=500 ) @app.post("/rag", response_model=RAGResponse) async def rag_query(request: RAGRequest) -> RAGResponse: """ Handle RAG queries from the realtime interface. This endpoint is called by the JavaScript frontend when the model requests the ask_rag function. """ try: logger.info(f"RAG query: {request.query} using method: {request.method}") # Validate and default method if needed method = request.method if method not in QUERY_DISPATCH: logger.warning(f"Invalid method '{method}', using default '{DEFAULT_METHOD}'") method = DEFAULT_METHOD # Get the appropriate query function query_func = QUERY_DISPATCH[method] # Execute the query start_time = time.time() answer, citations = query_func( question=request.query, image_path=request.image_path, top_k=request.top_k ) response_time = (time.time() - start_time) * 1000 # Convert to ms # Format citations for HTML display (optional) citations_html = format_citations_html(citations, method) # Log to analytics database (mark as voice interaction) try: # Generate unique session ID for each voice interaction import uuid voice_session_id = f"voice_{uuid.uuid4().hex[:8]}" log_query( user_query=request.query, method=method, answer=answer, citations=citations, response_time=response_time, image_path=request.image_path, top_k=request.top_k, session_id=voice_session_id, additional_settings={'voice_interaction': True, 'interaction_type': 'speech_to_speech'} ) logger.info(f"Voice interaction logged: {request.query[:50]}...") except Exception as log_error: logger.error(f"Failed to log voice query: {log_error}") logger.info(f"RAG query completed: {len(answer)} chars, {len(citations)} citations") return RAGResponse( answer=answer, citations=citations, method=method, citations_html=citations_html ) except Exception as e: logger.error(f"Error processing RAG query: {e}") raise HTTPException(status_code=500, detail=f"RAG query failed: {str(e)}") def format_citations_html(citations: list, method: str) -> str: """Format citations as HTML for display.""" if not citations: return "
No citations available
" html_parts = ["