import os import time from groq import Groq import deepgram_tts GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "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") PERSONA_PROMPTS = { "Medical Telehealth Assistant": ( "You are Dr. Thalia, an empathetic and professional AI Telehealth Calling Assistant calling a patient. " "Your goal is to communicate lab results, answer patient medical concerns clearly, and schedule follow-ups. " "CRITICAL INSTRUCTIONS FOR VOICE SYNTHESIS:\n" "1. Speak naturally as if on a live phone call.\n" "2. Keep responses brief (1 to 3 short spoken sentences).\n" "3. Do NOT use bullet points, markdown bold, lists, or symbols like # or *.\n" "4. Spell out medical numbers clearly (e.g. 'two hundred forty milligrams per deciliter')." ), "Customer Support Specialist": ( "You are Alex, a helpful AI Customer Service Calling Agent assisting a customer over the phone. " "Keep your tone polite, natural, and conversational. Keep responses to 1-3 spoken sentences without markdown formatting." ), "Outbound Sales & Qualification": ( "You are Jordan, a friendly AI Outbound Account Manager calling a potential business client. " "Speak concisely, ask engaging follow-up questions, and maintain a warm phone persona." ), "Custom Assistant": ( "You are an AI Voice Calling Agent on a live phone call. Speak naturally, warmly, and concisely." ) } class AICallingAgent: def __init__(self): self.client = Groq(api_key=GROQ_API_KEY) self.model = "llama-3.3-70b-versatile" def process_call_turn( self, user_input: str, conversation_history: list, persona: str = "Medical Telehealth Assistant", voice_model: str = "aura-2-thalia-en" ) -> dict: """ Processes a phone call conversational turn: 1. Generates conversational LLM text response via Groq API. 2. Synthesizes voice audio via Deepgram Aura-2 API. Returns dictionary with text, audio file path, latency, and updated history. """ start_time = time.time() system_prompt = PERSONA_PROMPTS.get(persona, PERSONA_PROMPTS["Medical Telehealth Assistant"]) messages = [{"role": "system", "content": system_prompt}] for msg in conversation_history: messages.append({"role": msg["role"], "content": msg["content"]}) messages.append({"role": "user", "content": user_input}) try: completion = self.client.chat.completions.create( model=self.model, messages=messages, temperature=0.7, max_tokens=250 ) agent_text = completion.choices[0].message.content.strip() except Exception as e: print(f"[AICallingAgent] Groq API Error: {e}") agent_text = "I apologize, I am experiencing a brief connection drop. Could you please repeat that?" # Generate Voice Audio via Deepgram audio_filepath = deepgram_tts.generate_voice_audio(agent_text, voice_model=voice_model) latency_ms = int((time.time() - start_time) * 1000) updated_history = list(conversation_history) updated_history.append({"role": "user", "content": user_input}) updated_history.append({"role": "assistant", "content": agent_text}) return { "agent_text": agent_text, "audio_filepath": audio_filepath, "latency_ms": latency_ms, "history": updated_history }