Spaces:
Running
Running
| """Minimal Gemini Live smoke test — isolates SDK behavior from app. | |
| Reads GEMINI_API_KEY from environment, connects to the Live API with | |
| the configured model, sends a single text turn, prints every event | |
| received until session closes or 30s timeout. | |
| Use: | |
| GEMINI_API_KEY=... /venvs/apps_venv/bin/python scripts/gemini_live_smoke.py | |
| GEMINI_API_KEY=... GEMINI_LIVE_MODEL=gemini-2.0-flash-live-001 /venvs/apps_venv/bin/python scripts/gemini_live_smoke.py | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import os | |
| import sys | |
| async def main() -> int: | |
| key = os.environ.get("GEMINI_API_KEY", "").strip() | |
| if not key: | |
| print("ERROR: GEMINI_API_KEY not set", file=sys.stderr) | |
| return 1 | |
| model = os.environ.get("GEMINI_LIVE_MODEL", "gemini-2.5-flash-native-audio-latest") | |
| print(f"[smoke] model={model}") | |
| try: | |
| from google import genai | |
| except ImportError as e: | |
| print(f"ERROR: google-genai not installed: {e}", file=sys.stderr) | |
| return 1 | |
| print(f"[smoke] google-genai version={getattr(genai, '__version__', '?')}") | |
| client = genai.Client(api_key=key, http_options={"api_version": "v1beta"}) | |
| # Native-audio models REQUIRE AUDIO modality. The 1007 error | |
| # "Cannot extract voices from a non-audio request" confirms this. | |
| config = { | |
| "response_modalities": ["AUDIO"], | |
| } | |
| try: | |
| async with client.aio.live.connect(model=model, config=config) as session: | |
| print("[smoke] connected; sending one text turn (turn_complete=True)...") | |
| await session.send_client_content( | |
| turns=[{"role": "user", "parts": [{"text": "Say hello in one short friendly sentence."}]}], | |
| turn_complete=True, | |
| ) | |
| event_count = 0 | |
| audio_bytes_total = 0 | |
| try: | |
| async with asyncio.timeout(30): | |
| async for resp in session.receive(): | |
| event_count += 1 | |
| text = getattr(resp, "text", None) | |
| data = getattr(resp, "data", None) | |
| if data: | |
| audio_bytes_total += len(data) | |
| sc = getattr(resp, "server_content", None) | |
| tc = getattr(sc, "turn_complete", None) if sc else None | |
| model_turn = getattr(sc, "model_turn", None) if sc else None | |
| mt_parts_summary = "" | |
| if model_turn is not None: | |
| parts = getattr(model_turn, "parts", None) or [] | |
| mt_parts_summary = f" model_turn.parts={len(parts)}" | |
| for i, p in enumerate(parts[:3]): | |
| ip = getattr(p, "inline_data", None) | |
| tp = getattr(p, "text", None) | |
| th = getattr(p, "thought", None) | |
| print( | |
| f"[smoke] part {i}: text={tp!r}, " | |
| f"inline_data={'<%d bytes>' % len(getattr(ip, 'data', b'')) if ip else None}, " | |
| f"thought={th}" | |
| ) | |
| print( | |
| f"[smoke] event #{event_count}: text={text!r}, " | |
| f"data={'<%d bytes>' % len(data) if data else None}, " | |
| f"turn_complete={tc}{mt_parts_summary}" | |
| ) | |
| if tc: | |
| print("[smoke] turn_complete=True — exiting receive() loop") | |
| break | |
| except asyncio.TimeoutError: | |
| print(f"[smoke] timed out after 30s, events={event_count}, audio_total={audio_bytes_total} bytes") | |
| print(f"[smoke] done. total events={event_count}, total_audio_bytes={audio_bytes_total}") | |
| except Exception as e: | |
| import traceback | |
| print(f"[smoke] CONNECTION ERROR: {e}") | |
| traceback.print_exc() | |
| return 1 | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(asyncio.run(main())) | |