Batch-requests met de Gemini API

Wanneer gebruik je batch-verwerking

Batch-verwerking is de juiste keuze als je:

• Honderden tot miljoenen prompts wilt verwerken
• Geen real-time antwoord nodig hebt (resultaat mag tot 24 uur later komen)
• Kosten wilt minimaliseren (50 procent korting op input- en outputtokens)
• Rate limit-problemen wilt vermijden bij grote volumes

Typische use cases: een hele productcatalogus verrijken, samenvattingen genereren voor een archief, een kennisbank embedden, of classificaties draaien op grote datasets.

De juiste SDK en modellen

Gebruik de nieuwe, unified google-genai SDK (from google import genai). De oude google-generativeai SDK is gedeprec8 en krijgt geen nieuwe features meer.

pip install google-genai

Gangbare modellen in 2026 zijn gemini-2.5-flash (snel en goedkoop) en gemini-3.5-flash (nieuwer). Voor embeddings gebruik je gemini-embedding-001. Controleer de officiele modellenpagina van Google voor de actuele lijst, want modelnamen wisselen regelmatig.

Een batch-job aanmaken met inline requests

Voor kleinere batches (inputpayload onder 20 MB) geef je de requests direct mee:

from google import genai

client = genai.Client()

producten = [
    {"id": "P001", "naam": "Draadloze muis"},
    {"id": "P002", "naam": "Mechanisch toetsenbord"},
    {"id": "P003", "naam": "4K monitor 27 inch"},
]

inline_requests = [
    {
        "contents": [
            {"role": "user", "parts": [{"text": f"Schrijf een korte productomschrijving voor: {p['naam']}"}]}
        ]
    }
    for p in producten
]

batch_job = client.batches.create(
    model="gemini-2.5-flash",
    src=inline_requests,
    config={"display_name": "productomschrijvingen"},
)

print(f"Aangemaakt: {batch_job.name}")

Een batch-job aanmaken met een JSONL-bestand

Voor grote batches (tot 2 GB inputbestand) upload je een JSON Lines-bestand. Elke regel is een volledige request met een eigen key, zodat je resultaten later kunt terugkoppelen aan je brondata:

import json
from google import genai
from google.genai import types

client = genai.Client()

producten = [
    {"id": "P001", "naam": "Draadloze muis"},
    {"id": "P002", "naam": "Mechanisch toetsenbord"},
]

with open("requests.jsonl", "w") as f:
    for p in producten:
        regel = {
            "key": p["id"],
            "request": {
                "contents": [
                    {"role": "user", "parts": [{"text": f"Categoriseer dit product: {p['naam']}"}]}
                ]
            },
        }
        f.write(json.dumps(regel) + "
")

uploaded = client.files.upload(
    file="requests.jsonl",
    config=types.UploadFileConfig(display_name="batch-requests", mime_type="jsonl"),
)

batch_job = client.batches.create(
    model="gemini-2.5-flash",
    src=uploaded.name,
    config={"display_name": "product-classificatie"},
)

Voortgang monitoren

Een batch-job doorloopt verschillende statussen. Pollen doe je met client.batches.get():

import time

completed_states = {
    "JOB_STATE_SUCCEEDED",
    "JOB_STATE_FAILED",
    "JOB_STATE_CANCELLED",
    "JOB_STATE_EXPIRED",
}

job = client.batches.get(name=batch_job.name)
while job.state.name not in completed_states:
    print(f"Status: {job.state.name}")
    time.sleep(30)
    job = client.batches.get(name=batch_job.name)

print(f"Eindstatus: {job.state.name}")

De belangrijkste statussen:

Resultaten ophalen

De uitvoer komt terug als inline responses (bij inline requests) of als bestand (bij JSONL-invoer):

job = client.batches.get(name=batch_job.name)

if job.state.name == "JOB_STATE_SUCCEEDED":
    if job.dest and job.dest.file_name:
        inhoud = client.files.download(file=job.dest.file_name)
        for regel in inhoud.decode("utf-8").splitlines():
            resultaat = json.loads(regel)
            print(resultaat.get("key"), resultaat.get("response"))
    elif job.dest and job.dest.inlined_responses:
        for i, response in enumerate(job.dest.inlined_responses):
            if response.response:
                print(f"Antwoord {i + 1}: {response.response.text}")

Een job annuleren

Een lopende job stop je direct, al gegenereerde uitvoer blijft beschikbaar:

client.batches.cancel(name=batch_job.name)

Batch embeddings

Embeddings hebben hun eigen methode, create_embeddings, met hetzelfde job-model:

from google import genai

client = genai.Client()

teksten = [
    "Artificiele intelligentie verandert de wereld",
    "Machine learning is een onderdeel van AI",
    "Neurale netwerken zijn geinspireerd op het menselijk brein",
]

inline_requests = [{"content": t} for t in teksten]

job = client.batches.create_embeddings(
    model="gemini-embedding-001",
    src={"inlined_requests": inline_requests},
    config={"display_name": "kennisbank-embeddings"},
)
print(f"Embedding-job: {job.name}")

Alternatief: parallelle calls voor real-time werk

Heb je toch synchrone resultaten nodig (bijvoorbeeld in een interactieve flow), dan zijn begrensde parallelle calls de beste keuze. Let op: dit valt onder de normale, niet-verlaagde tarieven en je raakt sneller rate limits.

import asyncio
from google import genai

client = genai.Client()

async def generate_for_item(item: dict, semaphore: asyncio.Semaphore) -> dict:
    async with semaphore:
        try:
            response = await client.aio.models.generate_content(
                model="gemini-2.5-flash",
                contents=f"Schrijf een productomschrijving voor: {item['naam']}",
            )
            return {"id": item["id"], "tekst": response.text, "status": "ok"}
        except Exception as e:
            return {"id": item["id"], "tekst": None, "status": str(e)}

async def process_batch(items: list, concurrency: int = 5) -> list:
    semaphore = asyncio.Semaphore(concurrency)
    return await asyncio.gather(*(generate_for_item(i, semaphore) for i in items))

results = asyncio.run(process_batch(producten))
for r in results:
    print(f"{r['id']}: {r['status']}")

Vertex AI Batch Prediction

Werk je binnen Google Cloud met grote, herhaalbare pipelines, dan biedt Vertex AI Batch Prediction dezelfde 50 procent korting met diepere integratie: input en output via Cloud Storage of BigQuery, IAM-rechten en monitoring binnen je project. De Gemini Batch API is doorgaans eenvoudiger om mee te starten; Vertex AI loont als je al in dat ecosysteem zit.