Streaming responses met de Gemini API

Waarom streaming

Bij een standaard API-call wacht je tot het model het volledige antwoord heeft gegenereerd. Voor lange antwoorden (samenvatten, schrijven, analyseren) kan dat enkele seconden tot meer dan een halve minuut duren. Met streaming ontvang je de tokens zodra ze beschikbaar zijn, vergelijkbaar met hoe een chatinterface het antwoord live laat verschijnen.

Streaming is vooral waardevol voor conversatie-interfaces, live dashboards en elke toepassing waar wachttijd de gebruikerservaring schaadt. De totale verwerkingstijd verandert niet, maar de waargenomen snelheid wel.

Streaming in Python

Installeer eerst de SDK met pip install google-genai. De client leest je sleutel automatisch uit de omgevingsvariabele GEMINI_API_KEY (of GOOGLE_API_KEY).

from google import genai

client = genai.Client()

response_stream = client.models.generate_content_stream(
    model="gemini-2.5-flash",
    contents="Schrijf een gedetailleerde uitleg over hoe zonnepanelen werken.",
)

for chunk in response_stream:
    if chunk.text:
        print(chunk.text, end="", flush=True)

print()

De flush=True is belangrijk: zonder deze parameter buffert Python de output en zie je toch batches in plaats van een continue stroom.

Streaming in JavaScript (Node.js)

Installeer de SDK met npm install @google/genai.

import { GoogleGenAI } from "@google/genai";

const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY });

const response = await ai.models.generateContentStream({
    model: "gemini-2.5-flash",
    contents: "Schrijf een recept voor erwtensoep met stap-voor-stap instructies.",
});

for await (const chunk of response) {
    if (chunk.text) {
        process.stdout.write(chunk.text);
    }
}

console.log();

Let op dat chunk.text in de nieuwe JavaScript-SDK een property is en geen functie-aanroep meer (chunk.text()), zoals in de oude SDK.

Streaming in een webapplicatie (Server-Sent Events)

Voor een browser-frontend stuur je de chunks door via Server-Sent Events (SSE). Onderstaand voorbeeld gebruikt Flask.

import os
from flask import Flask, Response, request, stream_with_context
from google import genai

app = Flask(__name__)
client = genai.Client()

@app.route("/stream")
def stream_response():
    prompt = request.args.get("prompt", "Vertel iets interessants.")

    def generate():
        response_stream = client.models.generate_content_stream(
            model="gemini-2.5-flash",
            contents=prompt,
        )
        for chunk in response_stream:
            if chunk.text:
                yield f"data: {chunk.text}

"
        yield "data: [DONE]

"

    return Response(
        stream_with_context(generate()),
        mimetype="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

Frontend JavaScript:

const eventSource = new EventSource(`/stream?prompt=${encodeURIComponent(prompt)}`);
const output = document.getElementById("output");

eventSource.onmessage = (event) => {
    if (event.data === "[DONE]") {
        eventSource.close();
        return;
    }
    output.textContent += event.data;
};

Streaming met chatsessies

Voor een gesprek met geschiedenis gebruik je client.chats.create(...). De sessie houdt de history automatisch bij.

chat = client.chats.create(model="gemini-2.5-flash")

full_response = ""
for chunk in chat.send_message_stream("Leg uit hoe quantumcomputers werken."):
    if chunk.text:
        print(chunk.text, end="", flush=True)
        full_response += chunk.text

print()

Volledige respons en tokengebruik na streaming

Elke chunk draagt zijn eigen usage_metadata; de laatste chunk bevat de geaggregeerde totalen. Verzamel de chunks om er na afloop bij te kunnen.

response_stream = client.models.generate_content_stream(
    model="gemini-2.5-flash",
    contents="Schrijf een lang essay over de geschiedenis van de boekdrukkunst.",
)

chunks = []
for chunk in response_stream:
    chunks.append(chunk)
    if chunk.text:
        print(chunk.text, end="", flush=True)

usage = chunks[-1].usage_metadata
print(f"
Tokens gebruikt: {usage.total_token_count}")

Streaming met multimodale invoer

Streaming werkt ook met afbeeldingen, audio of andere bestanden in dezelfde call. Bouw de invoer op als een lijst met tekst en een Part.

from google import genai
from google.genai import types

client = genai.Client()

with open("diagram.png", "rb") as f:
    image_bytes = f.read()

response_stream = client.models.generate_content_stream(
    model="gemini-2.5-flash",
    contents=[
        "Geef een gedetailleerde technische analyse van dit architectuurdiagram.",
        types.Part.from_bytes(data=image_bytes, mime_type="image/png"),
    ],
)

for chunk in response_stream:
    if chunk.text:
        print(chunk.text, end="", flush=True)

Foutafhandeling bij streaming

Fouten kunnen optreden bij het opzetten van de stream of tijdens het itereren. Vang ze af met errors.APIError, de basisklasse voor API-fouten in de nieuwe SDK.

from google.genai import errors

def stream_with_error_handling(prompt: str):
    try:
        response_stream = client.models.generate_content_stream(
            model="gemini-2.5-flash",
            contents=prompt,
        )
        for chunk in response_stream:
            if chunk.text:
                yield chunk.text
    except errors.APIError as e:
        yield f"[API-fout {e.code}: {e.message}]"
    except Exception as e:
        yield f"[Onverwachte fout: {e}]"

Wil je weten waarom een antwoord stopte (bijvoorbeeld door een veiligheidsfilter), controleer dan chunk.candidates[0].finish_reason en chunk.prompt_feedback op de chunks die binnenkomen.