I wish I could RM/RF my way through this, but well changes were done to the voice conversion code for elevenlabs

main.py

@@ -3,8 +3,6 @@ import io
 import json
 import re
 import tempfile
-import asyncio
-from typing import Optional
 import logging
 from contextlib import asynccontextmanager
 from fastapi import FastAPI, Request, status, Depends, Header, HTTPException
@@ -20,10 +18,15 @@ from sqlalchemy import create_engine
 # --- GRADIO ---
 import gradio as gr
 
-# --- SETUP ---
-os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
-logging.getLogger('tensorflow').setLevel(logging.ERROR)
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+# --------------------------------------------------------------------------- #
+#                              CONFIGURATION
+# --------------------------------------------------------------------------- #
+os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
+logging.getLogger("tensorflow").setLevel(logging.ERROR)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
 
 load_dotenv()
 NEON_DATABASE_URL = os.getenv("NEON_DATABASE_URL")
@@ -31,12 +34,15 @@ OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
 ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
 SHARED_SECRET = os.getenv("SHARED_SECRET")
 
-# --- CONFIG ---
 COLLECTION_NAME = "real_estate_embeddings"
 EMBEDDING_MODEL = "hkunlp/instructor-large"
-ELEVENLABS_VOICE_NAME = "Leo"
+
+# *** HARD-CODED VOICE ID (as requested) ***
+ELEVENLABS_VOICE_ID = "IvLWq57RKibBrqZGpQrC"   # <-- your voice
+
 PLANNER_MODEL = "gpt-4o-mini"
 ANSWERER_MODEL = "gpt-4o"
+
 TABLE_DESCRIPTIONS = """
 - "ongoing_projects_source": Details about projects currently under construction.
 - "upcoming_projects_source": Information on future planned projects.
@@ -46,37 +52,48 @@ TABLE_DESCRIPTIONS = """
 - "feedback_source": Customer feedback and ratings for projects.
 """
 
-# --- CLIENTS ---
+# --------------------------------------------------------------------------- #
+#                               CLIENTS
+# --------------------------------------------------------------------------- #
 embeddings = None
 vector_store = None
 client_openai = OpenAI(api_key=OPENAI_API_KEY)
-client_elevenlabs = None # Initialize as None first
+client_elevenlabs = None
 
-# --- ADDED: DETAILED ELEVENLABS INITIALIZATION LOGGING ---
+# ---- ElevenLabs init with detailed logging ---------------------------------
 try:
-    # Log the key (partially) to verify it's being read
-    key_preview = ELEVENLABS_API_KEY[:5] + "..." + ELEVENLABS_API_KEY[-4:] if ELEVENLABS_API_KEY and len(ELEVENLABS_API_KEY) > 9 else "None or too short"
-    logging.info(f"Attempting to initialize ElevenLabs client with key: {key_preview}")
+    key_preview = (
+        f"{ELEVENLABS_API_KEY[:5]}...{ELEVENLABS_API_KEY[-4:]}"
+        if ELEVENLABS_API_KEY and len(ELEVENLABS_API_KEY) > 9
+        else "None"
+    )
+    logging.info(f"Initializing ElevenLabs client with key: {key_preview}")
 
-    # Ensure key is not None or empty before initializing
     if not ELEVENLABS_API_KEY:
-        raise ValueError("ELEVENLABS_API_KEY environment variable not set or empty.")
+        raise ValueError("ELEVENLABS_API_KEY is missing or empty.")
 
     client_elevenlabs = ElevenLabs(api_key=ELEVENLABS_API_KEY)
-    logging.info(f"Initialized ElevenLabs client object. Type: {type(client_elevenlabs)}")
+    logging.info(f"ElevenLabs client created – type: {type(client_elevenlabs)}")
 
-    # Try accessing a simple attribute or method to confirm initialization
-    # Note: This might make a network call during startup
+    # Verify we can list voices (optional, but proves the key works)
     voices = client_elevenlabs.voices.get_all()
-    logging.info(f"Successfully fetched {len(voices.voices)} voices from ElevenLabs.")
+    logging.info(f"Fetched {len(voices.voices)} voices from ElevenLabs.")
 
 except Exception as e:
-    logging.error(f"Failed to initialize ElevenLabs client or fetch voices: {e}", exc_info=True)
-    client_elevenlabs = None # Ensure it's None if init failed
-# --- END ADDED LOGGING ---
+    logging.error(f"ElevenLabs init failed: {e}", exc_info=True)
+    client_elevenlabs = None
+
+# ---- Log SDK version -------------------------------------------------------
+try:
+    import elevenlabs
 
+    logging.info(f"elevenlabs SDK version: {elevenlabs.__version__}")
+except Exception:
+    logging.error("Could not import elevenlabs package.")
 
-# --- LIFESPAN ---
+# --------------------------------------------------------------------------- #
+#                               FASTAPI APP
+# --------------------------------------------------------------------------- #
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     global embeddings, vector_store
@@ -94,18 +111,12 @@ async def lifespan(app: FastAPI):
     yield
     logging.info("Shutting down.")
 
-# --- ADDED: LIBRARY VERSION LOGGING ---
-try:
-    import elevenlabs
-    logging.info(f"Found elevenlabs library version: {elevenlabs.__version__}")
-except ImportError:
-    logging.error("Could not import elevenlabs library!")
-# --- END ADDED LOGGING ---
 
 app = FastAPI(lifespan=lifespan)
 
-
-# --- PROMPTS ---
+# --------------------------------------------------------------------------- #
+#                               PROMPTS
+# --------------------------------------------------------------------------- #
 QUERY_FORMULATION_PROMPT = """
 You are a query analysis agent. Transform the user's query into a precise search query and determine the correct table to filter by.
 **Available Tables:**
@@ -116,7 +127,7 @@ You are a query analysis agent. Transform the user's query into a precise search
 2. If status keywords (ongoing, completed, upcoming, etc.) are present, pick the matching table.
 3. If no status keyword, set filter_table to null.
 4. Return JSON: {{"search_query": "...", "filter_table": "table_name or null"}}
-"""
+""".strip()
 
 ANSWER_SYSTEM_PROMPT = """
 You are an expert AI assistant for a premier real estate developer.
@@ -128,158 +139,154 @@ You are an expert AI assistant for a premier real estate developer.
 1. Match user language (Hinglish → Hinglish, English → English).
 2. Use CONTEXT if available, else use core knowledge.
 3. Only answer real estate questions.
-"""
-
+""".strip()
 
-# --- AUDIO & LLM HELPERS ---
+# --------------------------------------------------------------------------- #
+#                         AUDIO & LLM HELPERS
+# --------------------------------------------------------------------------- #
 def transcribe_audio(audio_path: str, audio_bytes: bytes) -> str:
     for attempt in range(3):
         try:
             audio_file = io.BytesIO(audio_bytes)
-            filename = os.path.basename(audio_path)  # e.g., "audio.wav"
-
-            logging.info(f"Transcribing audio: {filename} ({len(audio_bytes)} bytes)")
+            filename = os.path.basename(audio_path)
 
+            logging.info(f"Transcribing {filename} ({len(audio_bytes)} bytes)")
             transcript = client_openai.audio.transcriptions.create(
                 model="whisper-1",
-                file=(filename, audio_file)  # ← Critical: gives format hint
+                file=(filename, audio_file),
             )
             text = transcript.text.strip()
 
             # Hinglish transliteration
-            if re.search(r'[\u0900-\u097F]', text):
-                response = client_openai.chat.completions.create(
+            if re.search(r"[\u0900-\u097F]", text):
+                resp = client_openai.chat.completions.create(
                     model="gpt-4o-mini",
-                    messages=[{"role": "user", "content": f"Transliterate to Roman (Hinglish): {text}"}],
-                    temperature=0.0
+                    messages=[
+                        {"role": "user", "content": f"Transliterate to Roman (Hinglish): {text}"}
+                    ],
+                    temperature=0.0,
                 )
-                text = response.choices[0].message.content.strip()
+                text = resp.choices[0].message.content.strip()
 
             logging.info(f"Transcribed: {text}")
             return text
-
         except Exception as e:
-            logging.error(f"Transcription error (attempt {attempt+1}): {e}", exc_info=True) # Added exc_info
+            logging.error(f"Transcription error (attempt {attempt + 1}): {e}", exc_info=True)
             if attempt == 2:
                 return ""
     return ""
 
-# --- UPDATED generate_elevenlabs_sync with check ---
-def generate_elevenlabs_sync(text: str, voice: str) -> bytes:
-    # --- ADDED THIS CHECK ---
+
+def generate_elevenlabs_sync(text: str) -> bytes:
+    """
+    Uses the **hard-coded voice ID** and the correct SDK method
+    `client.text_to_speech.convert`.
+    """
     if client_elevenlabs is None:
-        logging.error("ElevenLabs client is not initialized. Cannot generate audio.")
-        return b''
-    # --- END ADDED CHECK ---
+        logging.error("ElevenLabs client not initialized – skipping TTS.")
+        return b""
 
     for attempt in range(3):
         try:
-            # This call might still fail if init succeeded but key is bad at runtime
-            logging.info(f"Calling ElevenLabs generate for voice '{voice}'...")
-            audio_data = client_elevenlabs.generate(
+            logging.info("Calling ElevenLabs text_to_speech.convert...")
+            stream = client_elevenlabs.text_to_speech.convert(
+                voice_id=ELEVENLABS_VOICE_ID,
                 text=text,
-                voice=voice,
                 model="eleven_multilingual_v2",
-                output_format="mp3_44100_128"
+                output_format="mp3_44100_128",
             )
-            # Check if generate returns bytes directly or needs iteration (depends on exact version/method)
-            if isinstance(audio_data, bytes):
-                 logging.info(f"ElevenLabs generate returned {len(audio_data)} bytes.")
-                 return audio_data
-            else:
-                 # Handle streaming iterator if necessary
-                 chunks = b""
-                 for chunk in audio_data:
-                     chunks += chunk
-                 logging.info(f"ElevenLabs generate streamed {len(chunks)} bytes.")
-                 return chunks
-
+            # The SDK returns a generator of bytes – collect everything
+            audio_bytes = b""
+            for chunk in stream:
+                if chunk:
+                    audio_bytes += chunk
+            logging.info(f"TTS returned {len(audio_bytes)} bytes.")
+            return audio_bytes
         except Exception as e:
-            logging.error(f"ElevenLabs error during generate (attempt {attempt+1}): {e}", exc_info=True) # Added exc_info
+            logging.error(
+                f"ElevenLabs TTS error (attempt {attempt + 1}): {e}", exc_info=True
+            )
             if attempt == 2:
-                return b''
-    return b''
-# --- END UPDATED FUNCTION ---
+                return b""
+    return b""
+
 
 async def formulate_search_plan(user_query: str) -> dict:
-    logging.info(f"Formulating search plan for query: {user_query}")
+    logging.info(f"Formulating search plan for: {user_query}")
     for attempt in range(3):
         try:
-            # Format the prompt here with BOTH variables
-            formatted_prompt = QUERY_FORMULATION_PROMPT.format(
-                table_descriptions=TABLE_DESCRIPTIONS,
-                user_query=user_query
+            formatted = QUERY_FORMULATION_PROMPT.format(
+                table_descriptions=TABLE_DESCRIPTIONS, user_query=user_query
             )
-
-            response = await run_in_threadpool(
+            resp = await run_in_threadpool(
                 client_openai.chat.completions.create,
                 model=PLANNER_MODEL,
-                messages=[{"role": "user", "content": formatted_prompt}], # Use the fully formatted prompt
+                messages=[{"role": "user", "content": formatted}],
                 response_format={"type": "json_object"},
-                temperature=0.0
+                temperature=0.0,
             )
-            # Log the raw response BEFORE trying to parse
-            raw_response_content = response.choices[0].message.content
-            logging.info(f"Raw Planner LLM response content: {raw_response_content}")
-
-            # Try parsing
-            plan = json.loads(raw_response_content)
-            logging.info(f"Successfully parsed search plan: {plan}")
+            raw = resp.choices[0].message.content
+            logging.info(f"Planner raw response: {raw}")
+            plan = json.loads(raw)
+            logging.info(f"Parsed plan: {plan}")
             return plan
         except Exception as e:
-            # Log the specific error during parsing or API call, with traceback
-            logging.error(f"Planner error (attempt {attempt+1}): {e}", exc_info=True)
+            logging.error(f"Planner error (attempt {attempt + 1}): {e}", exc_info=True)
             if attempt == 2:
-                logging.warning("Planner failed after 3 attempts. Using fallback.")
                 return {"search_query": user_query, "filter_table": None}
-    # Fallback if loop finishes unexpectedly
-    logging.error("Planner loop finished unexpectedly. Using fallback.")
     return {"search_query": user_query, "filter_table": None}
 
+
 async def get_agent_response(user_text: str) -> str:
     for attempt in range(3):
         try:
             plan = await formulate_search_plan(user_text)
-            search_query = plan.get("search_query", user_text)
-            filter_table = plan.get("filter_table")
-            search_filter = {"source_table": filter_table} if filter_table else {}
+            search_q = plan.get("search_query", user_text)
+            filter_tbl = plan.get("filter_table")
+            search_filter = {"source_table": filter_tbl} if filter_tbl else {}
 
             docs = await run_in_threadpool(
                 vector_store.similarity_search,
-                search_query, k=3, filter=search_filter
+                search_q,
+                k=3,
+                filter=search_filter,
             )
             if not docs:
-                docs = await run_in_threadpool(vector_store.similarity_search, search_query, k=3)
+                docs = await run_in_threadpool(vector_store.similarity_search, search_q, k=3)
 
-            context = "\n\n".join([d.page_content for d in docs])
+            context = "\n\n".join(d.page_content for d in docs)
 
-            response = await run_in_threadpool(
+            resp = await run_in_threadpool(
                 client_openai.chat.completions.create,
                 model=ANSWERER_MODEL,
                 messages=[
                     {"role": "system", "content": ANSWER_SYSTEM_PROMPT},
                     {"role": "system", "content": f"CONTEXT:\n{context}"},
-                    {"role": "user", "content": f"Question: {user_text}"}
-                ]
+                    {"role": "user", "content": f"Question: {user_text}"},
+                ],
             )
-            return response.choices[0].message.content.strip()
+            return resp.choices[0].message.content.strip()
         except Exception as e:
-            logging.error(f"RAG error (attempt {attempt+1}): {e}", exc_info=True) # Added exc_info
+            logging.error(f"RAG error (attempt {attempt + 1}): {e}", exc_info=True)
             if attempt == 2:
                 return "Sorry, I couldn't respond. Please try again."
     return "Sorry, I couldn't respond."
 
 
-# --- AUTH ENDPOINT ---
+# --------------------------------------------------------------------------- #
+#                               AUTH ENDPOINT
+# --------------------------------------------------------------------------- #
 class TextQuery(BaseModel):
     query: str
 
+
 async def verify_token(x_auth_token: str = Header(...)):
     if not SHARED_SECRET or x_auth_token != SHARED_SECRET:
         logging.warning("Auth failed for /test-text-query")
         raise HTTPException(status_code=401, detail="Invalid token")
     logging.info("Auth passed")
 
+
 @app.post("/test-text-query", dependencies=[Depends(verify_token)])
 async def test_text_query_endpoint(query: TextQuery):
     logging.info(f"Text query: {query.query}")
@@ -287,58 +294,59 @@ async def test_text_query_endpoint(query: TextQuery):
     return {"response": response}
 
 
-# --- GRADIO AUDIO PROCESSING ---
+# --------------------------------------------------------------------------- #
+#                               GRADIO PIPELINE
+# --------------------------------------------------------------------------- #
 async def process_audio(audio_path):
     if not audio_path or not os.path.exists(audio_path):
         return None, "No valid audio file received."
 
     try:
-        # Read raw bytes
+        # ---- 1. READ RAW BYTES ------------------------------------------------
         with open(audio_path, "rb") as f:
             audio_bytes = f.read()
-
-        if len(audio_bytes) == 0:
+        if not audio_bytes:
             return None, "Empty audio file."
 
-        # 1. Transcribe — pass path + bytes
+        # ---- 2. TRANSCRIBE ----------------------------------------------------
         user_text = await run_in_threadpool(transcribe_audio, audio_path, audio_bytes)
         if not user_text:
             return None, "Couldn't understand audio. Try again."
 
         logging.info(f"User: {user_text}")
 
-        # 2. AI Response
+        # ---- 3. GET AI RESPONSE -----------------------------------------------
         agent_response = await get_agent_response(user_text)
         if not agent_response:
             return None, "No response generated."
 
         logging.info(f"AI: {agent_response[:100]}...")
 
-        # 3. Generate Speech
-        ai_audio_bytes = await run_in_threadpool(
-            generate_elevenlabs_sync, agent_response, ELEVENLABS_VOICE_NAME
-        )
+        # ---- 4. TEXT-TO-SPEECH ------------------------------------------------
+        ai_audio_bytes = await run_in_threadpool(generate_elevenlabs_sync, agent_response)
         if not ai_audio_bytes:
-            # Return the text response even if TTS fails
-            logging.error("Failed to generate voice. Returning text only.")
-            return None, f"**You:** {user_text}\n\n**AI:** {agent_response}\n\n_(Audio generation failed)_"
-
+            logging.error("TTS failed – returning text only.")
+            return (
+                None,
+                f"**You:** {user_text}\n\n**AI:** {agent_response}\n\n_(Audio generation failed)_",
+            )
 
-        # Save to temp file
+        # Save to a temporary file for Gradio
         with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as f:
             f.write(ai_audio_bytes)
             out_path = f.name
-            logging.info(f"Saved generated audio to temp file: {out_path}")
-
+            logging.info(f"Saved TTS audio to {out_path}")
 
         return out_path, f"**You:** {user_text}\n\n**AI:** {agent_response}"
 
     except Exception as e:
-        logging.error(f"Audio processing error: {e}", exc_info=True) # Added exc_info
+        logging.error(f"Audio processing error: {e}", exc_info=True)
         return None, f"Error: {str(e)}"
 
 
-# --- GRADIO UI ---
+# --------------------------------------------------------------------------- #
+#                               GRADIO UI
+# --------------------------------------------------------------------------- #
 with gr.Blocks(title="Real Estate AI") as demo:
     gr.Markdown("# Real Estate Voice Assistant")
     gr.Markdown("Ask about projects in Pune, Mumbai, Bengaluru, etc.")
@@ -349,11 +357,12 @@ with gr.Blocks(title="Real Estate AI") as demo:
 
     out_text = gr.Textbox(label="Conversation", lines=8)
 
-    inp.change(process_audio, inp, [out_audio, out_text])
+    inp.change(process_audio, inputs=inp, outputs=[out_audio, out_text])
 
-    # Removed examples to avoid FileNotFoundError with text inputs
-    # gr.Examples(examples=[], inputs=inp)
+    # No examples – they caused FileNotFound errors when clicking text.
 
 
-# --- MOUNT GRADIO ---
+# --------------------------------------------------------------------------- #
+#                               MOUNT GRADIO
+# --------------------------------------------------------------------------- #
 app = gr.mount_gradio_app(app, demo, path="/")