Skip to content

gemini_live

gemini_live

Gemini Live (BidiGenerateContent) realtime voice session.

Classes

GeminiLiveSession

GeminiLiveSession(
    *,
    api_key: Optional[str] = None,
    model: str = _DEFAULT_MODEL,
    voice: str = "Puck",
    instructions: str = "",
    language: str = "",
    enable_tools: bool = True,
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
)

Bases: RealtimeVoiceSession

Duplex audio session against Gemini Live over WebSockets.

Source code in src/diapason/speech/realtime/gemini_live.py
def __init__(
    self,
    *,
    api_key: Optional[str] = None,
    model: str = _DEFAULT_MODEL,
    voice: str = "Puck",
    instructions: str = "",
    language: str = "",
    enable_tools: bool = True,
    max_tool_steps: int = 12,
    allowed_tools: Optional[Sequence[str]] = None,
) -> None:
    from diapason.core.cloud_keys import get_cloud_key

    # Environment first, then the desktop app's Keychain store — the
    # LaunchAgent server never receives the app's env injection, so the
    # key pasted into Settings must be reachable from here too.
    self._api_key = api_key or get_cloud_key("GEMINI_API_KEY", "GOOGLE_API_KEY")
    self._model = model or _DEFAULT_MODEL
    self._voice = voice or "Zephyr"
    self._instructions = instructions
    self._language = language
    self._enable_tools = enable_tools
    self._allowed_tools = list(allowed_tools) if allowed_tools else None
    self._budget = VoiceToolBudget(max_tool_steps)
    self._ws: Any = None
    self._queue: asyncio.Queue[Optional[SessionEvent]] = asyncio.Queue()
    self._reader_task: Optional[asyncio.Task[None]] = None
    self._closed = False

Functions: