Update services/gemini_realtime_service.py

services/gemini_realtime_service.py

@@ -11,6 +11,7 @@ from google.genai.types import (
     SpeechConfig,
     VoiceConfig,
 )
+from google.api_core import exceptions as google_exceptions
 
 class GeminiRealtimeService:
     """Dịch vụ Gemini Realtime API với audio streaming thực"""
@@ -24,17 +25,21 @@ class GeminiRealtimeService:
         self.voice_name = "Puck"
         self.input_queue = asyncio.Queue()
         self.output_queue = asyncio.Queue()
+        self._session_task = None
         
     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"},
-        )
-        return True
+        try:
+            self.client = genai.Client(
+                api_key=self.api_key,
+                http_options={"api_version": "v1alpha"},
+            )
+            return True
+        except Exception as e:
+            raise Exception(f"Không thể khởi tạo Gemini client: {str(e)}")
     
     def encode_audio(self, data: np.ndarray) -> str:
         """Encode audio data to base64"""
@@ -61,8 +66,8 @@ class GeminiRealtimeService:
                 ),
             )
             
-            # Kết nối session
-            self.session = await self.client.aio.live.connect(
+            # SỬA LỖI: Sử dụng async with đúng cách
+            self.session = self.client.aio.live.connect(
                 model="gemini-2.0-flash-exp", 
                 config=config
             )
@@ -78,9 +83,8 @@ class GeminiRealtimeService:
             
             print("✅ Gemini Realtime Audio session started")
             
-            # Khởi động background tasks
-            asyncio.create_task(self._audio_sender())
-            asyncio.create_task(self._audio_receiver())
+            # Khởi động background task để xử lý session
+            self._session_task = asyncio.create_task(self._handle_session())
             
             return True
             
@@ -95,12 +99,33 @@ class GeminiRealtimeService:
             print(error_msg)
             return False
     
-    async def _audio_sender(self):
-        """Gửi audio data đến Gemini"""
+    async def _handle_session(self):
+        """Xử lý session realtime với async with đúng cách"""
         try:
+            # SỬA LỖI: Sử dụng async with đúng cách
             async with self.session as session:
-                async for audio_chunk in self._audio_stream_generator():
-                    await session.send(audio_chunk)
+                # Khởi động sender và receiver
+                sender_task = asyncio.create_task(self._audio_sender(session))
+                receiver_task = asyncio.create_task(self._audio_receiver(session))
+                
+                # Chờ cả hai task hoàn thành
+                await asyncio.gather(sender_task, receiver_task)
+                
+        except Exception as e:
+            error_msg = f"❌ Lỗi trong session: {e}"
+            if self.callback:
+                await self.callback({
+                    'type': 'error',
+                    'message': error_msg,
+                    'status': 'error'
+                })
+            print(error_msg)
+    
+    async def _audio_sender(self, session):
+        """Gửi audio data đến Gemini"""
+        try:
+            async for audio_chunk in self._audio_stream_generator():
+                await session.send(audio_chunk)
         except Exception as e:
             error_msg = f"❌ Lỗi gửi audio: {e}"
             if self.callback:
@@ -122,36 +147,35 @@ class GeminiRealtimeService:
                 print(f"❌ Lỗi audio generator: {e}")
                 break
     
-    async def _audio_receiver(self):
+    async def _audio_receiver(self, session):
         """Nhận audio response từ Gemini"""
         try:
-            async with self.session as session:
-                async for response in session:
-                    if hasattr(response, 'data') and response.data:
-                        # Audio response
-                        audio_data = np.frombuffer(response.data, dtype=np.int16)
-                        
-                        if self.callback:
-                            await self.callback({
-                                'type': 'audio',
-                                'audio_data': audio_data,
-                                'sample_rate': 24000,
-                                'status': 'audio_streaming'
-                            })
+            async for response in session:
+                if hasattr(response, 'data') and response.data:
+                    # Audio response
+                    audio_data = np.frombuffer(response.data, dtype=np.int16)
+                    
+                    if self.callback:
+                        await self.callback({
+                            'type': 'audio',
+                            'audio_data': audio_data,
+                            'sample_rate': 24000,
+                            'status': 'audio_streaming'
+                        })
+                    
+                    # Đưa vào output queue để phát ra loa
+                    self.output_queue.put_nowait((24000, audio_data))
+                    
+                elif hasattr(response, 'text') and response.text:
+                    # Text response
+                    if self.callback:
+                        await self.callback({
+                            'type': 'text',
+                            'content': response.text,
+                            'role': 'assistant',
+                            'status': 'text_response'
+                        })
                         
-                        # Đưa vào output queue để phát ra loa
-                        self.output_queue.put_nowait((24000, audio_data))
-                        
-                    elif hasattr(response, 'text') and response.text:
-                        # Text response
-                        if self.callback:
-                            await self.callback({
-                                'type': 'text',
-                                'content': response.text,
-                                'role': 'assistant',
-                                'status': 'text_response'
-                            })
-                            
         except Exception as e:
             error_msg = f"❌ Lỗi nhận audio: {e}"
             if self.callback:
@@ -221,6 +245,16 @@ class GeminiRealtimeService:
             
             return response.text
             
+        except google_exceptions.ResourceExhausted:
+            error_msg = "❌ Quota Gemini đã hết. Vui lòng kiểm tra billing."
+            if self.callback:
+                await self.callback({
+                    'type': 'error',
+                    'message': error_msg,
+                    'status': 'quota_exceeded'
+                })
+            return error_msg
+            
         except Exception as e:
             error_msg = f"❌ Lỗi gửi text: {e}"
             if self.callback:
@@ -229,12 +263,20 @@ class GeminiRealtimeService:
                     'message': error_msg,
                     'status': 'error'
                 })
-            return None
+            return error_msg
     
     async def close(self):
         """Đóng kết nối"""
         self.is_active = False
         
+        # Hủy task nếu đang chạy
+        if self._session_task:
+            self._session_task.cancel()
+            try:
+                await self._session_task
+            except asyncio.CancelledError:
+                pass
+        
         if self.callback:
             await self.callback({
                 'type': 'status',