Gemini Client

The Gemini client provides access to Google's Gemini models.

Configuration

from padwan_ai.gemini import GeminiClient

client = GeminiClient(
    api_key="...",  # or set GEMINI_API_KEY env var
    model="gemini-3.8-flash",  # default model
)

padwan_ai.gemini.supports_audio(model, fmt=None) / supports_vision(model) check model capabilities; AUDIO_FORMATS lists supported formats.

Usage

Basic Chat

from padwan_ai.conversation import Message

async with GeminiClient() as client:
    response, usage = await client.complete_chat(
        [Message(role="user", content="Hello!")]
    )
    print(response["content"])

Streaming

from padwan_ai.conversation import Message

async with GeminiClient() as client:
    stream = client.stream_chat([Message(role="user", content="Tell me a story")])
    async for chunk in stream:
        print(chunk, end="")

Unlike the OpenAI-family clients and Anthropic's stream_chat, Gemini's stream_chat silently ignores extra_params, and complete_chat doesn't accept it at all.

Thinking models

Gemini's reasoning models (e.g. gemini-2.5-flash, gemini-2.5-pro) can emit internal thought tokens alongside the final answer. Configure them via two fields:

thinking_config: ThinkingConfig | None (Gemini-specific) — merged into every generationConfig. Set thinkingBudget to allocate tokens for reasoning and includeThoughts=True to have the thought parts streamed back.
on_thought: Callable[[str], None] | None (defined on LLMClientBase, so any provider can adopt it) — called with each thought text chunk as it arrives. Thought chunks are not yielded as part of the normal text stream; this callback is the only way to see them.

from padwan_ai import GeminiClient

thoughts: list[str] = []

async with GeminiClient(
    model="gemini-3.8-flash",
    on_thought=thoughts.append,
    thinking_config={"thinkingBudget": 2048, "includeThoughts": True},
) as client:
    stream = client.stream_chat(
        [{"role": "user", "content": "What is 7 * 8? Think step by step."}]
    )
    async for chunk in stream:
        print(chunk, end="")

print("\n\nReasoning:", "".join(thoughts))

Without includeThoughts=True, the model may still think internally (consuming the budget) but won't emit the thoughts — the callback will never fire. Without a non-zero thinkingBudget, the model won't think at all.

Embeddings

from padwan_ai import vectors

async with GeminiClient(model="gemini-embedding-001") as client:
    resp = await client.fetch_embeddings(
        ["text 1", "text 2"],
        dimensions=768,
        extra_params={"taskType": "RETRIEVAL_DOCUMENT"},
    )
    vecs = vectors(resp, "gemini")  # one list[float] per input, in input order

Uses batchEmbedContents; the response holds one values vector per input, in order.

Text-to-speech

async with GeminiClient(model="gemini-3.8-flash-tts") as client:
    speech, usage = await client.generate_speech(
        "Say cheerfully: have a nice day!", "Kore"
    )
    # speech.audio: bytes, speech.mime_type: "audio/wav"

    # Two speakers: tag each line; speaker names must match the voice mapping.
    speech, _ = await client.generate_speech(
        [
            {
                "text": "Hi Bob!",
                "speechMetadata": {"speaker": "Alice", "style": "excited"},
            },
            {"text": "<laughs> Hey Alice.", "speechMetadata": {"speaker": "Bob"}},
        ],
        {"Alice": "Leda", "Bob": "Puck"},
    )

3.8 models return WAV and need tagged lines for multi-speaker. 2.5 models return raw 24 kHz PCM (audio/L16) and reject speechMetadata, so pass them a plain Alice: ... script instead.

generate_speech also accepts model= (overriding the client default) and language_code=. For raw request bodies, complete(body, model)/stream bypass the ChatMessage shaping.

Batch Processing

Gemini supports batch processing for large-scale requests via methods on GeminiClient.

Creating a batch

from padwan_ai.gemini import GeminiClient, BatchRequest

async with GeminiClient() as client:
    requests = [
        BatchRequest(
            contents=[{"role": "user", "parts": [{"text": "Question 1"}]}],
            key="q1",
        ),
        BatchRequest(
            contents=[{"role": "user", "parts": [{"text": "Question 2"}]}],
            key="q2",
        ),
    ]
    job = await client.create_batch(requests, display_name="my-batch")
    print(job.name)  # e.g. "batches/123456"

BatchRequest accepts optional generation_config and system_instruction fields. If key is omitted, requests are auto-keyed as request-0, request-1, etc.

Polling for results

from padwan_ai.gemini import BatchResult

job = await client.get_batch(job.name)
if job.succeeded:
    for resp in job.inlined_responses or []:
        result = BatchResult.from_inlined_response(resp)
        print(result.key, result.content)

Listing and cancelling

jobs, next_token = await client.list_batches(page_size=10)
# returns the raw BatchJobResponse dict, not a BatchJob
await client.cancel_batch("batches/123456")

Batch types reference

Type	Description
`BatchRequest`	Single request: `contents`, `generation_config`, `system_instruction`, `key`
`BatchJob`	Job state: `name`, `state`, `display_name`, `model`, `error`, `dest`, `stats`, `inlined_responses`, `is_terminal`, `succeeded`
`BatchResult`	Parsed result: `key`, `content`, `input_tokens`, `output_tokens`, `total_tokens`

Realtime (Live API)

RealtimeClient with a Gemini model opens a speech-to-speech Live API session over a WebSocket. Audio is mono little-endian PCM16 — 16 kHz in (LIVE_INPUT_SAMPLE_RATE), 24 kHz out (LIVE_OUTPUT_SAMPLE_RATE). Requires the realtime extra.

from padwan_ai import RealtimeClient

async with RealtimeClient(
    "gemini-3.1-flash-live-preview",
    instructions="Answer briefly.",
    voice="Puck",
) as conn:
    await conn.append_audio(pcm16_chunk)  # mono PCM16 @ 16 kHz
    async for message in conn:
        if audio := conn.audio_delta_bytes(message):
            playback.write(audio)  # PCM16 @ 24 kHz
        if conn.is_turn_complete(message):
            break

Automatic activity detection (server VAD) is the default. Pass turn_detection=NO_TURN_DETECTION to disable it and mark turns yourself with conn.activity_start() / conn.activity_end(); a mapping tunes the automaticActivityDetection fields (e.g. {"silenceDurationMs": 500}). Input/output transcription events arrive via conn.input_transcript(message) / conn.output_transcript(message); disable them by passing transcription_model=None to RealtimeClient (or GeminiRealtimeClient(transcription=False) directly). conn.send_text(...) sends a text turn instead of audio, and conn.audio_stream_end() signals the end of an audio turn; GeminiRealtimeClient.setup_payload holds the session config sent on connect.