hina625/agent-Backend
0
1import logging2import os3import json4import time5from dotenv import load_dotenv6 7# LiveKit import statements8from livekit.agents import (9 Agent,10 AgentSession,11 JobContext,12 JobProcess,13 WorkerOptions,14 cli,15 TurnHandlingOptions,16 llm,17 BackgroundAudioPlayer,18 AudioConfig,19 BuiltinAudioClip,20)21from livekit.agents.voice.turn import EndpointingOptions, PreemptiveGenerationOptions22from livekit.agents.metrics import LLMMetrics23from livekit.plugins import openai, silero, deepgram, cartesia, elevenlabs24 25# Local custom STT and TTS imports26from stt import FasterWhisperSTT27from tts import PiperTTS28from dynamic_rag import DynamicRAG29from twilio.rest import Client30from google_calendar import create_event31 32# Load environment configurations (.env file)33load_dotenv()34 35# Setup logger for printing information in console36logger = logging.getLogger("outbound-agent")37logger.setLevel(logging.INFO)38 39# Initialize Dynamic RAG Manager40rag_manager = DynamicRAG()41 42# Configuration settings (can be overridden via .env file)43WHISPER_DEVICE = os.getenv("WHISPER_DEVICE", "cpu")44PIPER_USE_CUDA = os.getenv("PIPER_USE_CUDA", "false").lower() == "true"45OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1")46 47 48def prewarm(proc: JobProcess):49 """Load Voice Activity Detection (VAD) model into memory before call starts"""50 proc.userdata["vad"] = silero.VAD.load()51 52 53async def entrypoint(ctx: JobContext):54 """Main function called when a new participant connects to the voice room"""55 logger.info(f"Connecting to room: {ctx.room.name}")56 await ctx.connect()57 58 # Wait for the user to join the call59 participant = await ctx.wait_for_participant()60 logger.info(f"Started voice assistant for user: {participant.identity}")61 62 # 1. Default settings for LLM, STT, Language, and Voice63 call_context = {64 "prompt": "You are Astra, an extremely empathetic, highly intelligent, and friendly AI companion. Your goal is to be a true friend to the caller. You listen carefully to their problems, offer thoughtful advice, and solve their issues in any domain (life, tech, knowledge). Talk like a caring human friend, not a robotic AI. Use a warm, natural, and conversational tone. Rules: 1. Keep responses concise (1-3 sentences max). 2. Speak in plain text, NO markdown. 3. Always reply in the user's exact language.",65 "provider": "ollama",66 "model": "qwen3.5:9b",67 "stt_model": "base",68 "language": "hi",69 "voice": "voices/pratham/medium/hi_IN-pratham-medium.onnx",70 }71 72 # 2. Read custom user choices from participant metadata (if sent by frontend)73 try:74 raw_meta = participant.metadata75 if raw_meta:76 parsed_meta = json.loads(raw_meta)77 call_context.update(parsed_meta)78 logger.info(f"Loaded customized call context: {call_context}")79 except Exception as e:80 logger.error(f"Error parsing metadata: {e}")81 82 # 3. Setup LLM Plugin (Groq, OpenAI, or Ollama)83 provider = call_context.get("provider", "ollama")84 model_name = call_context.get("model", "qwen3.5:9b")85 openai_key = os.getenv("OPENAI_API_KEY", "")86 groq_key = os.getenv("GROQ_API_KEY", "")87 88 llm_plugin = None89 90 if openai_key and not openai_key.startswith("gsk_"):91 # Quick check if OpenAI has quota by making a tiny chat request92 import requests93 try:94 res = requests.post(95 "https://api.openai.com/v1/chat/completions",96 headers={97 "Authorization": f"Bearer {openai_key}",98 "Content-Type": "application/json"99 },100 json={101 "model": "gpt-4o-mini",102 "messages": [{"role": "user", "content": "hello"}],103 "max_tokens": 1104 },105 timeout=3106 )107 if res.status_code == 200:108 logger.info("Using OpenAI Cloud LLM (gpt-4o-mini)")109 llm_plugin = openai.LLM(model="gpt-4o-mini", api_key=openai_key)110 else:111 logger.warning(f"OpenAI API check failed ({res.status_code}). Attempting Groq fallback...")112 except Exception as e:113 logger.warning(f"OpenAI API check exception ({e}). Attempting Groq fallback...")114 115 if not llm_plugin and (groq_key or openai_key.startswith("gsk_")):116 fallback_key = groq_key if groq_key else openai_key117 logger.info("Using Groq API LLM (llama-3.3-70b-versatile) as fallback")118 llm_plugin = openai.LLM(119 model="llama-3.3-70b-versatile",120 api_key=fallback_key,121 base_url="https://api.groq.com/openai/v1",122 )123 124 if not llm_plugin:125 logger.info(f"Using Ollama Local LLM ({model_name})")126 llm_plugin = openai.LLM.with_ollama(127 model=model_name,128 base_url=OLLAMA_BASE_URL,129 )130 131 # 4. Setup Speech-to-Text (STT) - Cloud (Deepgram/Groq) or Local Faster-Whisper132 deepgram_key = os.getenv("DEEPGRAM_API_KEY", "")133 134 if deepgram_key:135 logger.info("Using Deepgram Cloud STT")136 stt_plugin = deepgram.STT()137 elif groq_key or openai_key.startswith("gsk_"):138 fallback_key = groq_key if groq_key else openai_key139 logger.info("Using Groq Cloud Whisper STT")140 stt_plugin = openai.STT(141 model="whisper-large-v3",142 api_key=fallback_key,143 base_url="https://api.groq.com/openai/v1",144 )145 else:146 logger.info(f"Using local Faster-Whisper STT (Device: {WHISPER_DEVICE})")147 stt_plugin = FasterWhisperSTT(148 model_size=call_context.get("stt_model", "base"),149 device=WHISPER_DEVICE,150 compute_type="float16" if WHISPER_DEVICE == "cuda" else "int8",151 )152 153 # 5. Setup Text-to-Speech (TTS) - Cloud (ElevenLabs/Cartesia/OpenAI) or Local Piper154 eleven_key = os.getenv("ELEVEN_API_KEY", "")155 cartesia_key = os.getenv("CARTESIA_API_KEY", "")156 157 if eleven_key:158 logger.info("Using ElevenLabs Cloud TTS")159 tts_plugin = elevenlabs.TTS(api_key=eleven_key)160 elif cartesia_key:161 logger.info("Using Cartesia Cloud TTS")162 tts_plugin = cartesia.TTS(api_key=cartesia_key)163 elif openai_key and not openai_key.startswith("gsk_"):164 logger.info("Using OpenAI Cloud TTS")165 tts_plugin = openai.TTS(api_key=openai_key)166 else:167 voice_file = call_context.get("voice", "voices/pratham/medium/hi_IN-pratham-medium.onnx")168 169 # Clean up leading slash to resolve path correctly on Windows170 if voice_file.startswith("/") or voice_file.startswith("\\"):171 voice_file = voice_file.lstrip("/\\")172 173 base_dir = os.path.dirname(os.path.abspath(__file__))174 absolute_voice_path = os.path.join(base_dir, voice_file)175 176 logger.info(f"Using local Piper TTS (CPU fallback): {absolute_voice_path}")177 tts_plugin = PiperTTS(178 model_path=absolute_voice_path,179 use_cuda=PIPER_USE_CUDA,180 )181 182 class AssistantFnc(llm.FunctionContext):183 @llm.ai_callable(description="Book a meeting on the user's Google Calendar. Call this when the user asks to schedule or book a meeting.")184 async def book_meeting(self, date: str, time: str, name: str):185 """186 This function is called by the LLM when the user wants to book a meeting.187 """188 logger.info(f"AI requested to book a meeting for {name} on {date} at {time}")189 success, message = create_event(date, time, f"Meeting with {name}")190 return message191 192 # 6. Define how our Voice Agent behaves193 class OutboundAgent(Agent):194 def __init__(self) -> None:195 super().__init__(196 instructions=call_context.get("prompt", "You are a helpful assistant."),197 )198 199 async def on_enter(self) -> None:200 """GREET the user as soon as they connect"""201 identity = participant.identity202 if identity.startswith("sip:"):203 phone_number = identity.replace("sip:", "")204 logger.info(f"SIP Call received from phone: {phone_number}")205 206 # --- SEND TWILIO SMS START ---207 try:208 twilio_sid = os.getenv("TWILIO_ACCOUNT_SID")209 twilio_token = os.getenv("TWILIO_AUTH_TOKEN")210 twilio_from = os.getenv("TWILIO_PHONE_NUMBER")211 if twilio_sid and twilio_token and twilio_from:212 twilio_client = Client(twilio_sid, twilio_token)213 twilio_client.messages.create(214 body="Hello! Thanks for calling Astra. We're on the line now. Feel free to ask me to book a meeting for you!",215 from_=twilio_from,216 to="+" + phone_number.lstrip("+")217 )218 except Exception as e:219 logger.error(f"Failed to send Twilio start SMS: {e}")220 # --- SEND TWILIO SMS END ---221 222 await self.session.say(223 "Hello! Thanks for calling the Astra AI hotline. I am connected and ready to help. How can I assist you today?", 224 allow_interruptions=True225 )226 else:227 await self.session.say("Hello. I am connected and ready to help!", allow_interruptions=True)228 229 async def on_user_turn_completed(230 self, turn_ctx: llm.ChatContext, new_message: llm.ChatMessage,231 ) -> None:232 user_text = new_message.text_content233 logger.info(f"User turn completed. Query: {user_text}")234 235 import re236 # Extract URLs and PDF paths from text237 urls = re.findall(r'(https?://\S+)', user_text)238 pdfs = re.findall(r'(\b\S+\.pdf\b)', user_text)239 240 processed_any = False241 242 for url in urls:243 logger.info(f"Detected URL to index: {url}")244 await self.session.say(f"Reading the website content...", allow_interruptions=True)245 246 # Fetch and index URL text247 success = rag_manager.add_url(url)248 if success:249 await self.session.say("Finished reading website content. You can now ask questions about it!", allow_interruptions=True)250 else:251 await self.session.say("Sorry, I failed to load that website.", allow_interruptions=True)252 processed_any = True253 254 for pdf in pdfs:255 logger.info(f"Detected PDF to index: {pdf}")256 await self.session.say(f"Reading the PDF file...", allow_interruptions=True)257 258 success = rag_manager.add_pdf(pdf)259 if success:260 await self.session.say("Finished reading the PDF. You can now ask questions about it!", allow_interruptions=True)261 else:262 await self.session.say("Sorry, I could not read that PDF file. Make sure the file exists.", allow_interruptions=True)263 processed_any = True264 265 # Retrieve relevant context for the LLM266 rag_content = rag_manager.search(user_text)267 logger.info(f"Retrieved RAG context: {rag_content}")268 269 # Inject context into LLM context if we have valid context270 if rag_content and "No matching information" not in rag_content and "No documents" not in rag_content:271 turn_ctx.add_message(272 role="assistant",273 content=f"Relevant context from documents/websites to help answer: {rag_content}"274 )275 276 # 7. Print latency statistics when LLM generates response277 turn_counter = 0278 @llm_plugin.on("metrics_collected")279 def on_llm_metrics(metrics: LLMMetrics):280 nonlocal turn_counter281 turn_counter += 1282 logger.info(283 f"📊 [LLM Stats] Turn: {turn_counter} | Reply Time: {metrics.ttft * 1000:.0f}ms | Output Tokens: {metrics.completion_tokens}"284 )285 286 # 8. Create and start the Agent Session with optimized options for fast response times287 session = AgentSession(288 vad=ctx.proc.userdata["vad"],289 stt=stt_plugin,290 llm=llm_plugin,291 tts=tts_plugin,292 fnc_ctx=AssistantFnc(),293 turn_handling=TurnHandlingOptions(294 endpointing=EndpointingOptions(295 min_delay=0.5,296 max_delay=1.0,297 ),298 preemptive_generation=PreemptiveGenerationOptions(299 enabled=True,300 )301 )302 )303 304 @ctx.room.on("disconnected")305 def on_disconnected(reason):306 logger.info(f"Room disconnected: {reason}")307 identity = participant.identity308 if identity.startswith("sip:"):309 phone_number = identity.replace("sip:", "")310 try:311 twilio_sid = os.getenv("TWILIO_ACCOUNT_SID")312 twilio_token = os.getenv("TWILIO_AUTH_TOKEN")313 twilio_from = os.getenv("TWILIO_PHONE_NUMBER")314 if twilio_sid and twilio_token and twilio_from:315 twilio_client = Client(twilio_sid, twilio_token)316 twilio_client.messages.create(317 body="Thank you for speaking with Astra today! If you booked a meeting, your calendar has been updated.",318 from_=twilio_from,319 to="+" + phone_number.lstrip("+")320 )321 except Exception as e:322 logger.error(f"Failed to send Twilio end SMS: {e}")323 324 await session.start(room=ctx.room, agent=OutboundAgent())325 326 327if __name__ == "__main__":328 # Start the LiveKit agent worker application329 cli.run_app(330 WorkerOptions(331 entrypoint_fnc=entrypoint,332 prewarm_fnc=prewarm,333 job_memory_warn_mb=1500,334 )335 )