app.py

import os
import asyncio
import json
import base64

from dotenv import load_dotenv
from quart import Quart, Response, request, websocket
import websockets
import google.generativeai as genai
from pinecone import Pinecone, ServerlessSpec

# Load environment variables
load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
GENAI_API_KEY = os.getenv("GENAI_API_KEY")
PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
PLIVO_AUTH_ID = os.getenv("PLIVO_AUTH_ID")
PLIVO_AUTH_TOKEN = os.getenv("PLIVO_AUTH_TOKEN")
PLIVO_FROM_NUMBER = os.getenv("PLIVO_FROM_NUMBER")

if not OPENAI_API_KEY or not GENAI_API_KEY or not PINECONE_API_KEY:
    raise ValueError("Missing required API keys (OPENAI_API_KEY, GENAI_API_KEY, PINECONE_API_KEY)")

# Configure Gemini and Pinecone
genai.configure(api_key=GENAI_API_KEY)
pc = Pinecone(api_key=PINECONE_API_KEY)
INDEX_NAME = "voice-bot-gemini-embedding-004-index"

if INDEX_NAME not in pc.list_indexes().names():
    pc.create_index(
        name=INDEX_NAME,
        dimension=768,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )

index = pc.Index(INDEX_NAME)

app = Quart(__name__)
PORT = 5000

SYSTEM_MESSAGE = "You are a helpful and friendly AI assistant with access to product-related knowledge."

# Helper: request LLM Gemini to summarize answer based on context
def request_llm_to_get_summarize(query: str, context: str) -> str:
    prompt = f"""
You are a highly accurate and detail-oriented question-answering assistant. Respond based only on these product search results:
{context}

User question: {query}

Provide a clear and concise answer using only the given info.
"""
    response = genai.GenerativeModel("gemini-1.5-flash").generate_content(prompt)
    return response.text

# Async handler: search vector DB and retrieve answer
async def get_vector_search_answer(user_question: str, namespace: str = "") -> str:
    # Get query embedding
    query_response = genai.embed_content(model="models/text-embedding-004", content=user_question)
    query_embedding = query_response["embedding"]

    # Query Pinecone
    search_results = index.query(vector=query_embedding, namespace=namespace, top_k=5, include_metadata=True)
    context_lst = []
    for match in search_results["matches"]:
        chunk_text = match["metadata"]["text"]
        context_lst.append(chunk_text)

    context = "\n-----------------\n".join(context_lst)
    answer = request_llm_to_get_summarize(user_question, context)
    return answer

@app.route("/webhook", methods=["GET", "POST"])
async def webhook():
    caller_number = request.args.get("From", "Unknown")
    print(f"Incoming call from: {caller_number}")
    stream_url = f"ws://{request.host}/media-stream"
    xml_data = f"""<?xml version="1.0" encoding="UTF-8"?>
    <Response>
        <Speak voice="Polly.Amy">Welcome to NinjaGenie! Please ask anything you want to know!</Speak>
        <Stream streamTimeout="86400" keepCallAlive="true" bidirectional="true" contentType="audio/x-mulaw;rate=8000" audioTrack="inbound">
            {stream_url}
        </Stream>
    </Response>
    """
    return Response(xml_data, mimetype="application/xml")

@app.websocket("/media-stream")
async def media_stream():
    print("Plivo client connected")
    plivo_ws = websocket

    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1",
    }

    try:
        async with websockets.connect(url, extra_headers=headers) as openai_ws:
            print("Connected to OpenAI real-time API")

            # Send initial session update i have to send the tools in this session update
            # to find where the openai is responsding. 
            # In that response i will have the function based on that function i will call the corresponding function or tool.
            #  

            session_update = {
                "type": "session.update",
                "session": {
                    "turn_detection": {"type": "server_vad"},
                    "input_audio_format": "g711_ulaw",
                    "output_audio_format": "g711_ulaw",
                    "voice": "alloy",
                    "instructions": SYSTEM_MESSAGE,
                    "modalities": ["text", "audio"],
                    "temperature": 0.8
                }
            }
            await openai_ws.send(json.dumps(session_update))

            async def receive_from_plivo():
                while True:
                    message = await plivo_ws.receive()
                    data = json.loads(message)

                    if data.get("event") == "media" and openai_ws.open:
                        audio_append = {
                            "type": "input_audio_buffer.append",
                            "audio": data["media"]["payload"]
                        }
                        await openai_ws.send(json.dumps(audio_append))

                    elif data.get("event") == "start":
                        print("Plivo Audio streaming started")
                        plivo_ws.stream_id = data["start"]["streamId"]

            async def receive_from_openai():
                async for message in openai_ws:
                    response = json.loads(message)

                    # Handle session updated
                    if response.get("type") == "session.updated":
                        print("OpenAI session updated")

                    # Handle audio delta response to send to Plivo
                    elif response.get("type") == "response.audio.delta":
                        audio_delta = {
                            "event": "playAudio",
                            "media": {
                                "contentType": "audio/x-mulaw",
                                "sampleRate": 8000,
                                "payload": base64.b64encode(base64.b64decode(response["delta"])).decode('utf-8')
                            }
                        }
                        await plivo_ws.send(json.dumps(audio_delta))

                    # Handle text transcript for product query detection
                    elif response.get("type") == "input_audio_buffer.transcript":
                        transcript = response.get("transcript", "")
                        print(f"Transcript received: {transcript}")

                        # Here you can detect product-related queries or commands and call vector search
                        if transcript.lower().startswith("product:") or "price" in transcript.lower():
                            # Remove "product:" prefix if present
                            query = transcript
                            if transcript.lower().startswith("product:"):
                                query = transcript[len("product:"):].strip()

                            # Replace with namespace or default as needed
                            namespace = "Tec Nviirons Sample data testing.xlsx 2025-05-02 09:05:24"

                            # Call vector search answer handler
                            answer = await get_vector_search_answer(query, namespace)

                            # For now, just log the answer since sending to OpenAI websocket as user prompt is invalid
                            print(f"Vector search answer: {answer}")

                            # TODO: Integrate TTS and send as audio delta to Plivo websocket for user playback

                    # Handle speech start event to clear audio buffer if needed
                    elif response.get("type") == "input_audio_buffer.speech_started":
                        print("Speech detected, clearing audio buffer")
                        clear_audio = {
                            "event": "clearAudio",
                            "stream_id": getattr(plivo_ws, "stream_id", None)
                        }
                        await plivo_ws.send(json.dumps(clear_audio))

                    # Handle response cancellation or other types as needed

            await asyncio.gather(receive_from_plivo(), receive_from_openai())

    except Exception as e:
        print(f"Error in OpenAI WebSocket: {e}")

    except websockets.ConnectionClosed:
        print("OpenAI server closed connection")

    except asyncio.CancelledError:
        print("Cancelled error")

if __name__ == "__main__":
    print("Starting server...")
    app.run(host="0.0.0.0", port=PORT)