Gemini Live

1
import asyncio
2
import logging
3
import os
4
from typing import Optional
5
6
from dotenv import load_dotenv
7
from google import genai
8
from google.genai import errors as genai_errors
9
from google.genai import types
10
from google.genai.live import AsyncSession
11
from websockets import ConnectionClosed
12
13
from agentduet import (
14
    BufferFullError,
15
    Call,
16
    CallAudioConfig,
17
    CallClosedError,
18
    IncomingCallNotification,
19
    SessionManager,
20
    SessionManagerConfig,
21
    new_session_id,
22
)
23
24
load_dotenv()
25
26
logging.basicConfig(
27
    level=logging.INFO,
28
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
29
)
30
logger = logging.getLogger(__name__)
31
32
genai_client = genai.Client(vertexai=False, api_key=os.getenv("GEMINI_API_KEY"))
33
MODEL = "models/gemini-3.1-flash-live-preview"
34
CONFIG = types.LiveConnectConfig(
35
    response_modalities=[types.Modality.AUDIO],
36
    speech_config=types.SpeechConfig(
37
        voice_config=types.VoiceConfig(
38
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr")
39
        )
40
    ),
41
    system_instruction="You are a helpful and friendly AI assistant.",
42
)
43
44
45
class GeminiLiveIntegration:
46
    def __init__(self, call: Call, gemini_session: AsyncSession):
47
        self._call = call
48
        self._gemini_session = gemini_session
49
        self._send_to_gemini_task: Optional[asyncio.Task] = None
50
        self._recv_from_gemini_task: Optional[asyncio.Task] = None
51
        self._terminated = False
52
53
    async def _on_hangup(self, evt):
54
        logger.info("Call terminated - closing Gemini session")
55
        self._terminated = True
56
        try:
57
            await self._gemini_session.close()
58
        except Exception:
59
            logger.exception("Error closing Gemini session")
60
61
        for task in (self._send_to_gemini_task, self._recv_from_gemini_task):
62
            if task:
63
                task.cancel()
64
                try:
65
                    await task
66
                except (asyncio.CancelledError, genai_errors.APIError):
67
                    pass
68
69
    async def run(self):
70
        self._call.on_hangup(self._on_hangup)
71
        self._send_to_gemini_task = asyncio.create_task(self.stream_to_gemini())
72
        self._recv_from_gemini_task = asyncio.create_task(
73
            self.receive_audio_from_gemini()
74
        )
75
        await asyncio.gather(
76
            self._send_to_gemini_task,
77
            self._recv_from_gemini_task,
78
            return_exceptions=True,
79
        )
80
81
    async def stream_to_gemini(self):
82
        try:
83
            await self._gemini_session.send_realtime_input(text="Hello, how are you?")
84
            async for audio_chunk in self._call.caller.audio_stream():
85
                await self._gemini_session.send_realtime_input(
86
                    audio=types.Blob(
87
                        data=audio_chunk, mime_type="audio/pcm;rate=24000"
88
                    )
89
                )
90
        except ConnectionClosed:
91
            pass
92
        except CallClosedError:
93
            logger.debug("Call closed; stopping stream to Gemini")
94
        except asyncio.CancelledError:
95
            raise
96
        except Exception:
97
            logger.exception("Error in stream to Gemini")
98
            raise
99
100
    async def receive_audio_from_gemini(self):
101
        try:
102
            while True:
103
                async for response in self._gemini_session.receive():
104
                    if server_content := response.server_content:
105
                        if server_content.interrupted:
106
                            # Public barge-in API - clears queued outbound PCM
107
                            await self._call.clear_send_audio_buffer()
108
                            logger.debug("Gemini interrupted - cleared send buffer")
109
                            break
110
                        if model_turn := server_content.model_turn:
111
                            for part in model_turn.parts:
112
                                if part.inline_data and isinstance(
113
                                    part.inline_data.data, bytes
114
                                ):
115
                                    try:
116
                                        await self._call.send_audio(
117
                                            part.inline_data.data
118
                                        )
119
                                    except BufferFullError:
120
                                        logger.warning(
121
                                            "Send buffer full - drop chunk or "
122
                                            "raise CallAudioConfig.buffer_size"
123
                                        )
124
                                if part.text is not None:
125
                                    logger.debug("Text: %s", part.text)
126
        except (ConnectionClosed, genai_errors.APIError):
127
            logger.debug("Gemini session closed")
128
        except asyncio.CancelledError:
129
            raise
130
        except CallClosedError:
131
            logger.debug("Call closed; stopping stream from Gemini")
132
        except Exception:
133
            logger.exception("Error in stream from Gemini")
134
            raise
135
136
137
async def main():
138
    config = SessionManagerConfig.create(
139
        api_key=os.getenv("AGENTDUET_API_KEY"),
140
        connector_uuid=os.getenv("AGENTDUET_CONNECTOR_UUID"),
141
        call_audio=CallAudioConfig(
142
            sample_rate=24000,
143
            buffer_size=1024 * 1024,
144
        ),
145
    )
146
147
    async with SessionManager(config) as sm:
148
        logger.info("SessionManager started %s", sm.id)
149
150
        @sm.on_incoming_call
151
        async def on_call(noti: IncomingCallNotification):
152
            session = await sm.open_session(new_session_id(), noti.subscriber)
153
            call = await session.process_call(noti)
154
            logger.info("Incoming call=%s caller=%s", call.id, call.caller)
155
            try:
156
                async with genai_client.aio.live.connect(
157
                    model=MODEL, config=CONFIG
158
                ) as gemini_session:
159
                    result = await call.answer()
160
                    if not result:
161
                        logger.error(
162
                            "Answer failed %s: %s (%s)",
163
                            call.id,
164
                            result.error_message,
165
                            result.error_code,
166
                        )
167
                        return
168
                    await GeminiLiveIntegration(call, gemini_session).run()
169
            except Exception:
170
                logger.exception("Error in Gemini Live bridge")
171
                await call.close()
172
                raise
173
174
        await sm.run_forever()
175
176
177
if __name__ == "__main__":
178
    asyncio.run(main())