Create voice_assistant.py

utils/voice_assistant.py

@@ -0,0 +1,168 @@
+"""
+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 import generate, 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