Gemini 3.8 Flash & Project Astra: thinking_level Architecture & WebSocket Live Audio/Video Agents

from google import genai
from google.genai import types

client = genai.Client()

response = client.models.generate_content(
    model="gemini-3.8-flash",
    contents=(
        "Analyze a payment retry pipeline for race conditions. "
        "Return: invariant violations, minimal repro timeline, and a safe locking/idempotency redesign."
    ),
    config=types.GenerateContentConfig(
        thinking_config=types.ThinkingConfig(thinking_level="medium")
    ),
)

print(response.text)

import asyncio
from pathlib import Path
from google import genai
from google.genai import types

MODEL = "gemini-3.1-flash-live-preview"
client = genai.Client()

async def main() -> None:
    config = {
        "response_modalities": ["AUDIO"],
        "input_audio_transcription": {},
    }
    async with client.aio.live.connect(model=MODEL, config=config) as session:
        pcm = Path("input.pcm").read_bytes()
        chunk_bytes = 3200  # ~100 ms @ 16kHz PCM16 mono
        for i in range(0, len(pcm), chunk_bytes):
            await session.send_realtime_input(
                audio=types.Blob(
                    data=pcm[i:i + chunk_bytes],
                    mime_type="audio/pcm;rate=16000",
                )
            )
        await session.send_realtime_input(audio_stream_end=True)
        async for msg in session.receive():
            content = msg.server_content
            if content and content.model_turn:
                for part in content.model_turn.parts:
                    if part.inline_data:
                        Path("output.pcm").open("ab").write(part.inline_data.data)

if __name__ == "__main__":
    asyncio.run(main())