Create gemini_realtime_service.py

services/gemini_realtime_service.py

@@ -0,0 +1,226 @@
+import asyncio
+import base64
+import json
+import os
+import pathlib
+from typing import AsyncGenerator, Callable, Literal, Optional
+import numpy as np
+from google import genai
+from google.genai.types import (
+    LiveConnectConfig,
+    PrebuiltVoiceConfig,
+    SpeechConfig,
+    VoiceConfig,
+)
+
+class GeminiRealtimeService:
+    """Dịch vụ Gemini Realtime API cho streaming chất lượng cao"""
+    
+    def __init__(self, api_key: str = None):
+        self.api_key = api_key or os.getenv("GEMINI_API_KEY")
+        self.client = None
+        self.session = None
+        self.is_active = False
+        self.callback = None
+        self.voice_name = "Puck"
+        
+    async def initialize(self):
+        """Khởi tạo client Gemini"""
+        if not self.api_key:
+            raise ValueError("Gemini API key is required")
+            
+        self.client = genai.Client(
+            api_key=self.api_key,
+            http_options={"api_version": "v1alpha"},
+        )
+    
+    async def start_session(self, voice_name: str = "Puck", callback: Callable = None):
+        """Bắt đầu session Gemini Realtime"""
+        try:
+            if not self.client:
+                await self.initialize()
+            
+            self.voice_name = voice_name
+            self.callback = callback
+            
+            config = LiveConnectConfig(
+                response_modalities=["AUDIO"],
+                speech_config=SpeechConfig(
+                    voice_config=VoiceConfig(
+                        prebuilt_voice_config=PrebuiltVoiceConfig(
+                            voice_name=voice_name,
+                        )
+                    )
+                ),
+            )
+            
+            self.session = await self.client.aio.live.connect(
+                model="gemini-2.0-flash-exp", 
+                config=config
+            )
+            
+            self.is_active = True
+            
+            # Khởi động background task để nhận responses
+            asyncio.create_task(self._response_handler())
+            
+            if self.callback:
+                self.callback({
+                    'type': 'status',
+                    'message': f'✅ Đã kết nối Gemini - Giọng: {voice_name}',
+                    'status': 'connected'
+                })
+            
+            print("✅ Gemini Realtime session started")
+            return True
+            
+        except Exception as e:
+            error_msg = f"❌ Lỗi khởi động Gemini Realtime: {e}"
+            if self.callback:
+                self.callback({
+                    'type': 'error',
+                    'message': error_msg,
+                    'status': 'error'
+                })
+            print(error_msg)
+            return False
+    
+    async def send_audio_chunk(self, audio_chunk: np.ndarray, sample_rate: int = 16000):
+        """Gửi audio chunk đến Gemini"""
+        if not self.session or not self.is_active:
+            return False
+            
+        try:
+            # Gemini expects 16kHz sample rate
+            if sample_rate != 16000:
+                audio_chunk = self._resample_audio(audio_chunk, sample_rate, 16000)
+            
+            # Encode và gửi audio
+            audio_bytes = audio_chunk.tobytes()
+            await self.session.send(audio_bytes)
+            return True
+            
+        except Exception as e:
+            print(f"❌ Lỗi gửi audio đến Gemini: {e}")
+            return False
+    
+    async def _response_handler(self):
+        """Xử lý responses từ Gemini"""
+        try:
+            async for response in self.session:
+                if hasattr(response, 'data') and response.data:
+                    # Audio response from Gemini
+                    audio_data = np.frombuffer(response.data, dtype=np.int16)
+                    
+                    if self.callback:
+                        self.callback({
+                            'type': 'audio',
+                            'audio_data': audio_data,
+                            'sample_rate': 24000,
+                            'status': 'audio_streaming'
+                        })
+                        
+                elif hasattr(response, 'text') and response.text:
+                    # Text response from Gemini
+                    if self.callback:
+                        self.callback({
+                            'type': 'text',
+                            'content': response.text,
+                            'role': 'assistant',
+                            'status': 'text_response'
+                        })
+                        
+        except Exception as e:
+            error_msg = f"❌ Lỗi nhận response từ Gemini: {e}"
+            if self.callback:
+                self.callback({
+                    'type': 'error',
+                    'message': error_msg,
+                    'status': 'error'
+                })
+            print(error_msg)
+    
+    def _resample_audio(self, audio_chunk: np.ndarray, original_rate: int, target_rate: int):
+        """Resample audio chunk (đơn giản hóa - trong thực tế dùng lib resample)"""
+        # Đây là implementation đơn giản, trong thực tế nên dùng librosa hoặc scipy
+        if original_rate == target_rate:
+            return audio_chunk
+            
+        ratio = target_rate / original_rate
+        new_length = int(len(audio_chunk) * ratio)
+        return np.interp(
+            np.linspace(0, len(audio_chunk) - 1, new_length),
+            np.arange(len(audio_chunk)),
+            audio_chunk
+        ).astype(np.int16)
+    
+    async def close(self):
+        """Đóng kết nối Gemini"""
+        if self.session:
+            await self.session.close()
+            self.is_active = False
+            
+        if self.callback:
+            self.callback({
+                'type': 'status',
+                'message': '🛑 Đã đóng kết nối Gemini',
+                'status': 'disconnected'
+            })
+            
+        print("🛑 Gemini Realtime session closed")
+
+class GeminiStreamHandler:
+    """Handler cho streaming audio với Gemini"""
+    
+    def __init__(
+        self,
+        expected_layout: Literal["mono"] = "mono",
+        output_sample_rate: int = 24000,
+        gemini_service: GeminiRealtimeService = None
+    ):
+        self.expected_layout = expected_layout
+        self.output_sample_rate = output_sample_rate
+        self.input_sample_rate = 16000
+        self.gemini_service = gemini_service
+        self.input_queue: asyncio.Queue = asyncio.Queue()
+        self.output_queue: asyncio.Queue = asyncio.Queue()
+        self.quit: asyncio.Event = asyncio.Event()
+    
+    async def start_up(self, api_key: str = None, voice_name: str = "Puck"):
+        """Khởi động Gemini service"""
+        if not self.gemini_service:
+            self.gemini_service = GeminiRealtimeService(api_key)
+            await self.gemini_service.initialize()
+        
+        # Set callback để nhận responses
+        self.gemini_service.callback = self._handle_gemini_callback
+        
+        await self.gemini_service.start_session(voice_name)
+    
+    async def _handle_gemini_callback(self, data: dict):
+        """Xử lý callback từ Gemini service"""
+        if data['type'] == 'audio':
+            self.output_queue.put_nowait(
+                (data['sample_rate'], data['audio_data'])
+            )
+    
+    async def receive(self, frame: tuple[int, np.ndarray]) -> None:
+        """Nhận audio frame và gửi đến Gemini"""
+        sample_rate, array = frame
+        array = array.squeeze()
+        
+        if self.gemini_service and self.gemini_service.is_active:
+            await self.gemini_service.send_audio_chunk(array, sample_rate)
+    
+    async def emit(self) -> tuple[int, np.ndarray] | None:
+        """Emit audio frame từ Gemini"""
+        try:
+            return await asyncio.wait_for(self.output_queue.get(), timeout=1.0)
+        except asyncio.TimeoutError:
+            return None
+    
+    def shutdown(self) -> None:
+        """Tắt handler"""
+        self.quit.set()
+        if self.gemini_service:
+            asyncio.create_task(self.gemini_service.close())