salon_agent.py

from dotenv import load_dotenv
from livekit.agents import ChatContext, ChatMessage
import asyncio, requests, uvicorn
from fastapi import FastAPI
from backend.lookup import my_rag_lookup
from firebase.firestore_service import add_request, update_request_by_question
import os
from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions, RunContext, function_tool
from livekit.plugins import noise_cancellation, silero, groq, deepgram, cartesia
from livekit.plugins.turn_detector.multilingual import MultilingualModel
from typing import Any
from backend.pinecone import upsert_data

load_dotenv(".env")

app = FastAPI()


session: AgentSession | None = None
main_loop: asyncio.AbstractEventLoop | None = None


@app.post("/ai_webhook")
async def ai_webhook(data: dict):
    global session, main_loop
    message = f"My supervisor says: {data['answer']}"
    print(f"[AI Webhook Triggered] {data['status']}")

    if session and main_loop:
        # Use main_loop.create_task to schedule the coroutine
        # on the same loop the session is running on.
        # This is safer than awaiting it directly from the webhook.
        upsert_data(data["question"], data["answer"])
        result = update_request_by_question(data["question"], data["status"])
        print(f"[Firestore Update] {result}")
        session.say(message)

        return {"status": "AI replied"}
    else:
        return {"error": "Session or loop not ready"}, 500


# This function is no longer needed
# def start_webhook_server():
#    uvicorn.run(app, host="0.0.0.0", port=9000)


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""
            You are Emma, a friendly receptionist at Luxe Salon & Spa.

            INFORMATION YOU KNOW:
            - Business hours: Monday-Saturday 9 AM to 7 PM, Closed Sundays
            - Location: 123 Main Street, Downtown
            - Services and Prices:
              * Haircut: $50 (45 minutes)
              * Hair Coloring: $120 (2 hours)
              * Manicure: $35 (30 minutes)
              * Pedicure: $45 (45 minutes)
              * Facial: $80 (1 hour)
            - Booking: Call us or book online at luxesalon.com
            - Cancellation policy: 24 hours notice required
            - Walk-ins welcome based on availability

            VERY IMPORTANT INSTRUCTIONS:
            - Be warm, friendly, and professional.
            - NEVER answer questions outside your knowledge base.
            - If someone asks a question you don't have information about 
              (like specific stylist availability, product recommendations, 
              or anything not listed above), you MUST call the `request_help` function.
            - Never make up information.
            - After calling the `request_help` function, tell the user:
              "Let me check with my supervisor and I'll get back to you shortly."
            """
        )

    async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage,) -> None:
        
        user_text = ""
        if isinstance(new_message.content, list) and len(new_message.content) > 0:
            user_text = new_message.content[0]
        elif isinstance(new_message.content, str):
            user_text = new_message.content
        else:
            user_text = str(new_message.content)

        rag_content = await my_rag_lookup(user_text)
        print(f"[RAG Content Retrieved] {rag_content}")


    @function_tool()
    async def request_help(self, context: RunContext, question: str) -> dict[str, Any]:
        """
        Called when the assistant encounters a question it cannot answer.
        Logs the request, notifies the manager, or triggers a helpdesk API.
        """
        print(f"[Manager Alert] Assistant requested help for question: '{question}'")
        add_request({"question": question, "status": "pending"})
        return {"status": "help_requested"}


async def entrypoint(ctx: agents.JobContext):
    global session, main_loop
    main_loop = asyncio.get_event_loop()  # save the LiveKit loop

    # --- NEW: Start Uvicorn as an asyncio task ---
    config = uvicorn.Config(app, host="0.0.0.0", port=9000, loop="auto")
    server = uvicorn.Server(config)
    
    # Start the server in the background on the same event loop
    main_loop.create_task(server.serve())
    # ----------------------------------------------

    session = AgentSession(
        stt="assemblyai/universal-streaming:en",
        llm=groq.LLM(model="llama-3.1-8b-instant"),
        tts=cartesia.TTS(model="sonic-3", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02"),
        vad=silero.VAD.load(),
        turn_detection=MultilingualModel(),
    )

    # register AI webhook with backend
    # Give the server a moment to start before registering
    await asyncio.sleep(1) 
    try:
        requests.post("http://localhost:8000/register_ai", json={"webhook_url": "http://localhost:9000/ai_webhook"})
    except requests.exceptions.ConnectionError as e:
        print(f"Could not register AI webhook. Is the other server running? Error: {e}")


    await session.start(
        room=ctx.room,
        agent=Assistant(),
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
        ),
    )

    await session.generate_reply(
        instructions="Greet the user and offer your assistance."
    )


if __name__ == "__main__":
    # --- MODIFIED: Remove the Thread ---
    # We no longer start the server here
    
    # This line starts the asyncio loop and runs the entrypoint
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))