""" ElevenLabs Voice Assistant for RewardPilot Converts AI text responses to natural speech """ import os import logging from typing import Optional, List, Dict import io logger = logging.getLogger(__name__) # Check if ElevenLabs is available try: from elevenlabs.types import Voice, VoiceSettings from elevenlabs.client import ElevenLabs ELEVENLABS_AVAILABLE = True except ImportError: ELEVENLABS_AVAILABLE = False logger.warning("ElevenLabs not installed. Voice features will be disabled.") class VoiceAssistant: """Handle text-to-speech conversion using ElevenLabs""" def __init__(self): self.api_key = os.getenv("ELEVENLABS_API_KEY") self.enabled = ELEVENLABS_AVAILABLE and bool(self.api_key) if self.enabled: self.client = ElevenLabs(api_key=self.api_key) logger.info("✅ ElevenLabs Voice Assistant initialized") else: logger.warning("⚠️ ElevenLabs disabled (missing API key or library)") # Voice configurations self.voices = { "Rachel": { "voice_id": "21m00Tcm4TlvDq8ikWAM", # Professional female "description": "Clear, professional female voice" }, "Adam": { "voice_id": "pNInz6obpgDQGcFmaJgB", # Deep male "description": "Deep, authoritative male voice" }, "Bella": { "voice_id": "EXAVITQu4vr4xnSDxMaL", # Friendly female "description": "Warm, friendly female voice" }, "Antoni": { "voice_id": "ErXwobaYiN019PkySvjV", # Well-rounded male "description": "Well-rounded, versatile male voice" }, "Elli": { "voice_id": "MF3mGyEYCl7XYWbV9V6O", # Young female "description": "Young, energetic female voice" } } def text_to_speech( self, text: str, voice_name: str = "Rachel", model: str = "eleven_turbo_v2" ) -> Optional[bytes]: """ Convert text to speech audio Args: text: Text to convert voice_name: Name of voice to use model: ElevenLabs model (eleven_turbo_v2 is fastest) Returns: Audio bytes or None if failed """ if not self.enabled: logger.warning("Voice generation skipped (ElevenLabs not enabled)") return None if not text or len(text.strip()) == 0: logger.warning("Empty text provided for voice generation") return None # Limit text length to avoid API errors if len(text) > 2500: text = text[:2500] + "..." logger.info(f"Text truncated to 2500 characters for voice generation") try: voice_config = self.voices.get(voice_name, self.voices["Rachel"]) voice_id = voice_config["voice_id"] logger.info(f"🎤 Generating speech with {voice_name} ({len(text)} chars)") # Generate audio using ElevenLabs audio = self.client.generate( text=text, voice=Voice( voice_id=voice_id, settings=VoiceSettings( stability=0.5, # Balance between consistency and expressiveness similarity_boost=0.75, # How closely to match the original voice style=0.0, # Exaggeration level use_speaker_boost=True # Enhance clarity ) ), model=model ) # Convert generator to bytes audio_bytes = b"".join(audio) logger.info(f"✅ Generated {len(audio_bytes)} bytes of audio") return audio_bytes except Exception as e: logger.error(f"❌ Voice generation failed: {e}") return None def get_voice_list(self) -> List[Dict[str, str]]: """Get list of available voices""" return [ {"name": name, "description": config["description"]} for name, config in self.voices.items() ] def create_audio_summary(self, recommendation_data: dict) -> str: """ Create a concise audio-friendly summary of recommendation Args: recommendation_data: Normalized recommendation data Returns: Audio-optimized text """ card = recommendation_data.get('recommended_card', 'Unknown Card') rewards = recommendation_data.get('rewards_earned', 0) rate = recommendation_data.get('rewards_rate', 'N/A') merchant = recommendation_data.get('merchant', 'this merchant') reasoning = recommendation_data.get('reasoning', '') # Create concise, natural-sounding summary summary = f"For your purchase at {merchant}, I recommend using your {card}. " summary += f"You'll earn {rewards:.2f} dollars in rewards at a rate of {rate}. " # Add simplified reasoning (first sentence only) if reasoning: first_sentence = reasoning.split('.')[0].strip() if first_sentence and len(first_sentence) > 20: summary += f"{first_sentence}. " # Add warnings if present warnings = recommendation_data.get('warnings', []) if warnings: summary += "Important note: " + warnings[0] return summary # Global instance voice_assistant = VoiceAssistant() def get_voice_assistant() -> VoiceAssistant: """Get the global voice assistant instance""" return voice_assistant