from fastapi.middleware.cors import CORSMiddleware from fastapi import FastAPI, UploadFile, File, HTTPException from pydantic import BaseModel from fastapi.responses import FileResponse from translate import translate_to_target_language, translate_input_to_english from graph_main import graph from sarvamai import SarvamAI import tempfile from pydub import AudioSegment import io from groq import Groq from deepgram import DeepgramClient, FileSource, PrerecordedOptions, LiveOptions, LiveTranscriptionEvents import base64 import os import time import string import asyncio from dotenv import load_dotenv load_dotenv() app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], allow_credentials=True ) print("CORS middleware added") class ChatRequest(BaseModel): message: str groq_client = Groq(api_key=os.getenv("GROQ_API_KEY")) sarvam_client = SarvamAI(api_subscription_key=os.getenv("SARVAM_API_KEY")) deepgram_client = DeepgramClient(os.getenv("DEEPGRAM_API_KEY")) SARVAM_LANG_MAP = { "hindi": "hi-IN", "hi": "hi-IN", "tamil": "ta-IN", "ta": "ta-IN", "marathi": "mr-IN", "mr": "mr-IN", "bengali": "bn-IN", "bn": "bn-IN", "telugu": "te-IN", "te": "te-IN", "gujarati": "gu-IN", "gu": "gu-IN", "english": "en-IN", "en": "en-IN", "kannada": "kn-IN", "kn": "kn-IN" } @app.post("/chat") async def chat(request: ChatRequest): try: user_text = request.message config = {"configurable": {"thread_id": "hotel_owner_1"}} result = await graph.ainvoke({"query": user_text}, config=config) response = result.get("response", "sorry, i couldn't process that request") lang_prompt = f"""Detect the language of this text: "{user_text}" Return ONLY one of these codes: hi, en, ta, mr, bn, te, gu, kn Return ONLY the code, nothing else.""" from agents.llm import llm lang_response = llm.invoke(lang_prompt) detected_language = lang_response.content.strip().lower() print(f"Detected language: {detected_language}") final_response = translate_to_target_language(response, detected_language) sarvam_lang = SARVAM_LANG_MAP.get(detected_language, "hi-IN") for attempt in range(3): try: tts_response = sarvam_client.text_to_speech.convert( text=final_response, target_language_code=sarvam_lang, speaker="abhilash", model="bulbul:v2" ) break except Exception as e: if attempt == 2: raise HTTPException(status_code=503, detail=f"TTS unavailable: {e}") audio_bytes = base64.b64decode(tts_response.audios[0]) audio_base64 = base64.b64encode(audio_bytes).decode("utf-8") return {"response": response, "audio": audio_base64} except Exception as e: import traceback print(f"Chat endpoint error: {e}") traceback.print_exc() return {"response": "Sorry, I encountered an error."} @app.get("/") def root(): return FileResponse("frontend/index2.html") @app.post("/voice") async def voice(audio: UploadFile = File(...)): temp_file = None try: audio_bytes = await audio.read() if len(audio_bytes) > 5 * 1024 * 1024: return HTTPException(status_code=400, detail="Audio file is too large. Max 5 MB file is allowed") try: raw_audio = AudioSegment.from_file(io.BytesIO(audio_bytes)) clean_audio = raw_audio.set_frame_rate(16000).set_channels(1) except Exception as e: raise HTTPException(status_code=400, detail=f"failed to process audio format: {e}") f = tempfile.NamedTemporaryFile(delete=False, suffix=".wav") temp_file = f.name f.close() clean_audio.export(temp_file, format="wav") total_start = time.time() with open(temp_file, "rb") as f: for attempt in range(3): try: stt_start = time.time() payload = {"buffer": f.read()} options = PrerecordedOptions( model="nova-2", language="hi-en", smart_format=True, punctuate=True ) response = deepgram_client.listen.rest.v("1").transcribe_file(payload, options) transcript_data = response["results"]["channels"][0]["alternatives"][0] raw_transcript = transcript_data["transcript"] print(f"Deepgram Heard: {raw_transcript}") print(f"STT Latency: {time.time() - stt_start:.2f}s") break except Exception as e: if attempt == 2: raise HTTPException(status_code=503, detail="STT service unavailable") try: clean_english_query = translate_input_to_english(raw_transcript) except Exception as e: raise HTTPException(status_code=503, detail="Translation not possible") detected_language = clean_english_query.get("language", "hi") english_text = clean_english_query.get("english_text", "") clean_english_query = english_text.translate(str.maketrans('', '', string.punctuation)).strip() if not clean_english_query: return {"transcript": "", "response": "Sorry, I couldn't hear you. Please try again.", "audio_base64": None} config = {"configurable": {"thread_id": "hotel_owner_1"}} try: result = await graph.ainvoke({"query": clean_english_query}, config=config) except Exception as e: import traceback traceback.print_exc() result = {"response": "I'm sorry, I encountered an internal glitch. Could you repeat that?"} graph_response = result.get("response", "Sorry, I couldn't process that request") final_response = translate_to_target_language(graph_response, detected_language) sarvam_lang = SARVAM_LANG_MAP.get(detected_language.lower(), "hi-IN") for attempt in range(3): try: tts_response = sarvam_client.text_to_speech.convert( text=final_response, target_language_code=sarvam_lang, speaker="abhilash", model="bulbul:v2" ) break except Exception as e: if attempt == 2: raise HTTPException(status_code=503, detail="TTS service unavailable") audio_bytes = base64.b64decode(tts_response.audios[0]) audio_base64 = base64.b64encode(audio_bytes).decode("utf-8") print(f"Total Latency: {time.time() - total_start:.2f}s") return { "transcript": clean_english_query, "detected_language": detected_language, "response": final_response, "audio": audio_base64, "booking_link": result.get("booking_link", "") } finally: if temp_file and os.path.exists(temp_file): os.remove(temp_file) from fastapi import WebSocket, WebSocketDisconnect async def process_and_respond(websocket: WebSocket, full_transcript: str, detected_language: str): """Takes the final transcript, runs LLM + TTS, sends audio back.""" english_text = full_transcript.strip() if not english_text: await websocket.send_json({"response": "Sorry, I couldn't hear anything. Please try again."}) return await websocket.send_json({"status": "processing", "transcript": english_text}) config = {"configurable": {"thread_id": "hotel_owner_1"}} try: graph_start = time.time() result = await graph.ainvoke({"query": english_text}, config=config) print(f"Graph Latency: {time.time() - graph_start:.2f}s") except Exception as e: import traceback traceback.print_exc() result = {"response": "I'm sorry, I encountered an internal glitch. Could you repeat that?"} graph_response = result.get("response", "Sorry, I couldn't process that request") print(f"Graph output: {graph_response}") final_response = await asyncio.to_thread(translate_to_target_language, graph_response, detected_language) sarvam_lang = SARVAM_LANG_MAP.get(detected_language.lower(), "hi-IN") for attempt in range(3): try: tts_start = time.time() tts_response = await asyncio.to_thread( sarvam_client.text_to_speech.convert, text=final_response, target_language_code=sarvam_lang, speaker="abhilash", model="bulbul:v2" ) print(f"TTS Latency: {time.time() - tts_start:.2f}s") break except Exception as e: if attempt == 2: await websocket.send_json({"response": "Audio generation failed, please try again."}) return audio_bytes = base64.b64decode(tts_response.audios[0]) audio_base64 = base64.b64encode(audio_bytes).decode("utf-8") try: await websocket.send_json({ "transcript": english_text, "detected_language": detected_language, "response": final_response, "audio": audio_base64 }) except Exception: print("Client disconnected before response could be sent") @app.websocket("/ws/voice") async def voice_ws(websocket: WebSocket): await websocket.accept() # Shared state between Deepgram callbacks and the main loop full_transcript = "" detected_language = "hi" utterance_end_event = asyncio.Event() # fires when Deepgram detects end of speech is_processing = False # prevent overlapping pipeline runs # ── Deepgram callbacks ──────────────────────────────────────────────────── async def on_transcript(self, result, **kwargs): nonlocal full_transcript, detected_language transcript = result.channel.alternatives[0].transcript if not transcript: return # Only accumulate is_final results to avoid duplicates if result.is_final: full_transcript += " " + transcript print(f"Deepgram [final]: {transcript}") # Translate on the fly so detected_language is ready when utterance ends try: translated = await asyncio.to_thread(translate_input_to_english, full_transcript.strip()) detected_language = translated.get("language", "hi") english = translated.get("english_text", full_transcript.strip()) # Replace accumulated transcript with clean english version full_transcript = english except Exception as e: print(f"Translation error: {e}") # Send live transcript to frontend so user sees what was heard try: await websocket.send_json({"live_transcript": full_transcript}) except Exception: pass else: # Send interim results so frontend can show "hearing..." feedback try: await websocket.send_json({"interim_transcript": transcript}) except Exception: pass async def on_utterance_end(self, utterance_end, **kwargs): """ Deepgram fires this after endpointing_ms of silence — means user stopped talking. We set the event here; the main loop picks it up and runs the pipeline. """ nonlocal is_processing print(f"Deepgram UtteranceEnd fired — transcript so far: '{full_transcript.strip()}'") if full_transcript.strip() and not is_processing: utterance_end_event.set() async def on_error(self, error, **kwargs): print(f"Deepgram error: {error}") try: await websocket.send_json({"error": str(error)}) except Exception: pass # ── Connect to Deepgram ─────────────────────────────────────────────────── dg_connection = deepgram_client.listen.asyncwebsocket.v("1") dg_connection.on(LiveTranscriptionEvents.Transcript, on_transcript) dg_connection.on(LiveTranscriptionEvents.UtteranceEnd, on_utterance_end) dg_connection.on(LiveTranscriptionEvents.Error, on_error) options = LiveOptions( model="nova-2", language="hi", smart_format=True, punctuate=True, encoding="linear16", sample_rate=16000, channels=1, # Server-side VAD settings: # endpointing_ms — how long Deepgram waits after last speech before # firing UtteranceEnd. 1200ms is a good balance for natural speech. endpointing=1200, # utterance_end_ms — extra buffer after endpointing before the event fires. utterance_end_ms="1500", # interim_results — send partial transcripts so we can show live feedback. interim_results=True, # vad_events — enables SpeechStarted event (optional, good for UI feedback). vad_events=True, ) started = await dg_connection.start(options) print(f"Deepgram started: {started}") if not started: print("ERROR: Deepgram WebSocket failed to start") await websocket.send_json({"error": "Speech service unavailable. Please try again."}) await websocket.close() return # ── Main loop: receive audio from browser, forward to Deepgram ─────────── # Browser now just streams raw PCM continuously — no END message needed. # Server-side VAD (via utterance_end_event) drives the pipeline. async def receive_audio(): """Reads audio chunks from browser and forwards to Deepgram.""" try: while True: message = await websocket.receive() if "bytes" in message and message["bytes"]: await dg_connection.send(message["bytes"]) elif "text" in message and message["text"] == "STOP": # Browser sends STOP only when user clicks mic off break except WebSocketDisconnect: pass try: # Run audio receiver in background so utterance_end_event can be # awaited at the same time in the foreground audio_task = asyncio.create_task(receive_audio()) while not audio_task.done(): # Wait for either Deepgram to detect end of utterance, or audio to stop try: await asyncio.wait_for(utterance_end_event.wait(), timeout=0.5) except asyncio.TimeoutError: continue if utterance_end_event.is_set(): utterance_end_event.clear() is_processing = True # Snapshot and reset transcript for next turn turn_transcript = full_transcript.strip() turn_language = detected_language full_transcript = "" if turn_transcript: await process_and_respond(websocket, turn_transcript, turn_language) is_processing = False audio_task.cancel() except WebSocketDisconnect: print("Client disconnected") finally: await dg_connection.finish() print("Deepgram connection closed")