Spaces:
Sleeping
Sleeping
File size: 4,232 Bytes
cf02b2b f844095 cf02b2b f844095 cf02b2b f844095 cf02b2b |
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 |
import os
import logging
from datetime import datetime
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from websocket_handler import handle_websocket_connection
from enhanced_websocket_handler import handle_enhanced_websocket_connection
from hybrid_llm_service import HybridLLMService
from voice_service import VoiceService
from rag_service import search_documents
from lancedb_service import LanceDBService
from scenario_analysis_service import ScenarioAnalysisService
import config
from dotenv import load_dotenv
# MCP and Authentication imports
from fastapi import Depends
from pydantic import BaseModel
from typing import Optional
from auth import get_current_user
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)
# Get configuration
config_dict = {
"ALLOWED_ORIGINS": config.ALLOWED_ORIGINS,
"ENABLE_VOICE_FEATURES": config.ENABLE_VOICE_FEATURES
}
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan handler"""
# Startup
logger.info("🚀 Starting Voice Bot Application...")
logger.info("✅ Application started successfully")
yield
# Shutdown (if needed)
logger.info("🛑 Shutting down Voice Bot Application...")
# Create FastAPI application
app = FastAPI(
title="Voice Bot Government Assistant",
description="AI-powered voice assistant for government policies and services",
version="1.0.0",
lifespan=lifespan
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=config.ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Initialize services (lazy loading for HF Spaces)
llm_service = None
voice_service = None
lancedb_service = None
scenario_service = None
def get_llm_service():
global llm_service
if llm_service is None:
llm_service = HybridLLMService()
return llm_service
def get_voice_service():
global voice_service
if voice_service is None:
voice_service = VoiceService()
return voice_service
def get_lancedb_service():
global lancedb_service
if lancedb_service is None:
lancedb_service = LanceDBService()
return lancedb_service
def get_scenario_service():
global scenario_service
if scenario_service is None:
scenario_service = ScenarioAnalysisService()
return scenario_service
# Health check endpoint
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"service": "voice-bot-api",
"timestamp": datetime.now().isoformat(),
"version": "1.0.0"
}
# Root endpoint
@app.get("/")
async def root():
"""Root endpoint with service information"""
return {
"message": "Voice Bot Government Assistant API",
"status": "running",
"version": "1.0.0",
"endpoints": {
"health": "/health",
"chat": "/chat",
"websocket": "/ws",
"docs": "/docs"
}
}
# Chat endpoint
@app.post("/chat")
async def chat_endpoint(request: dict):
"""Text-based chat endpoint"""
try:
message = request.get("message", "")
if not message:
raise HTTPException(status_code=400, detail="Message is required")
llm = get_llm_service()
response = await llm.get_response(message)
return {
"response": response,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Chat error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
# WebSocket endpoint
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""WebSocket endpoint for real-time communication"""
await handle_enhanced_websocket_connection(websocket)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |