feat: 新增 Gemini Live API 接入演示

examples/gemini-live/gemini/__init__.py

@@ -0,0 +1,4 @@
+from .gemini import Gemini
+from .xiaoai import XiaoAi
+
+__all__ = ["Gemini", "XiaoAi"]

examples/gemini-live/gemini/gemini.py

@@ -0,0 +1,102 @@
+import os
+from typing import Awaitable, Callable, Optional
+from google import genai
+from google.genai import types
+
+
+GEMINI_MODEL = "gemini-2.0-flash-live-001"
+GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") or "你的 API KEY"
+
+
+class Gemini:
+    running = False
+
+    client = genai.Client(api_key=GEMINI_API_KEY)
+
+    session: Optional[genai.live.AsyncSession] = None
+
+    config = types.LiveConnectConfig(
+        response_modalities=[types.Modality.AUDIO],
+        speech_config=types.SpeechConfig(
+            language_code="cmn-CN",
+            voice_config=types.VoiceConfig(
+                prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Kore")
+            ),
+        ),
+        context_window_compression=(
+            types.ContextWindowCompressionConfig(
+                sliding_window=types.SlidingWindow(),
+            )
+        ),
+    )
+
+    @classmethod
+    async def send_text(cls, text: str):
+        if not cls.session:
+            return
+        await cls.session.send_client_content(
+            turns={"role": "user", "parts": [{"text": text}]},
+            turn_complete=True,
+        )
+
+    @classmethod
+    async def send_audio(cls, data: bytes):
+        if not cls.session:
+            return
+        await cls.session.send_realtime_input(
+            audio=types.Blob(data=data, mime_type="audio/pcm;rate=16000")
+        )
+
+    @classmethod
+    def stop(cls):
+        cls.running = False
+        if cls.session:
+            cls.session.close()
+
+    @classmethod
+    async def start(
+        cls,
+        on_audio: Optional[Callable[[bytes], Awaitable[None]]] = None,
+        on_text: Optional[Callable[[str], Awaitable[None]]] = None,
+        set_is_speaking: Optional[Callable[[bool], Awaitable[None]]] = None,
+    ):
+        if cls.running:
+            return
+
+        cls.running = True
+
+        async with cls.client.aio.live.connect(
+            model=GEMINI_MODEL, config=cls.config
+        ) as session:
+            cls.session = session
+
+            print("🔊 AI: ", "session connected")
+
+            while True:
+                print("🔊 AI: ", "waiting for response")
+
+                if not cls.running:
+                    break
+
+                if set_is_speaking:
+                    await set_is_speaking(False)
+
+                async for response in session.receive():
+                    if response.server_content is None:
+                        continue
+
+                    if response.server_content.interrupted is True:
+                        continue
+
+                    if set_is_speaking:
+                        await set_is_speaking(True)
+
+                    if response.data is not None:
+                        print("🔊 AI: ", len(response.data))
+                        if on_audio:
+                            await on_audio(response.data)
+
+                    if response.text is not None:
+                        print("✅ AI: ", response.text)
+                        if on_text:
+                            await on_text(response.text)

examples/gemini-live/gemini/xiaoai.py

@@ -0,0 +1,51 @@
+import asyncio
+
+import numpy as np
+import open_xiaoai_server
+
+from gemini import Gemini
+
+
+class XiaoAi:
+    loop: asyncio.AbstractEventLoop
+    is_ai_speaking = False
+    speaking_count = 0
+
+    @classmethod
+    async def start(cls):
+        cls.loop = asyncio.get_event_loop()
+        open_xiaoai_server.register_fn("on_input_data", cls.input_audio)
+        await open_xiaoai_server.start_server()
+
+    @classmethod
+    async def output_audio(cls, data: bytes):
+        await open_xiaoai_server.on_output_data(data)
+
+    @classmethod
+    def input_audio(cls, data: bytes):
+        if cls.is_ai_speaking:
+            # 如果 AI 正在回答问题，则不发送音频，防止把 AI 的声音被当做输入录制进来
+            # 暂不支持中断 AI 的回复，需要等待 AI 回答完成后才能重新响应用户的语音输入
+            return
+
+        async def send_audio_task():
+            audio_array = np.frombuffer(data, dtype=np.uint16)
+            await Gemini.send_audio(audio_array.tobytes())
+
+        asyncio.run_coroutine_threadsafe(send_audio_task(), cls.loop)
+
+    @classmethod
+    async def set_is_speaking(cls, is_speaking: bool):
+        if is_speaking:
+            cls.speaking_count += 1
+            cls.is_ai_speaking = True
+            return
+
+        # 延迟 1 秒，如果 AI 还在说话，则设置为不说话
+        async def set_is_speaking_task():
+            speaking_count = cls.speaking_count
+            await asyncio.sleep(1)
+            if cls.speaking_count == speaking_count:
+                cls.is_ai_speaking = False
+
+        asyncio.run_coroutine_threadsafe(set_is_speaking_task(), cls.loop)