Einleitung und Anwendungsfall
Im vorherigen Artikel haben wir betrachtet, wie man Gemini nutzt, um strukturierte Ausgaben im JSON-Format für eine Klassifikationsaufgabe zu erzeugen. Falls du ihn noch nicht gelesen hast, hier eine kurze Zusammenfassung des Anwendungsfalls: In einem neu automatisierten Logistiklager wird neue Software eingeführt, um logistische Operationen abzuwickeln. Diese Software erfordert viele neue Attribute für jedes Produkt, die bisher unbekannt waren. Da das Unternehmen im E-Commerce über 13'000 Produkte verkauft, ist es zu zeitaufwändig, dies manuell zu erledigen, und einen Standardwert zuzuweisen ist nicht geeignet, da dies den Testbetrieb beeinträchtigen könnte.
In diesem Beispiel betrachten wir ein Attribut, das die Zerbrechlichkeit eines Produkts bestimmt – angegeben durch Ganzzahlwerte von 1 bis 4. Der Wert 1 steht für ein robustes Produkt, der Wert 4 für ein sehr zerbrechliches.
Da wir diese Klassifikation für über 13'000 Produkte durchführen müssen, ist es zeitaufwändig und kostspielig, Gemini für jedes Produkt einzeln aufzurufen. Deshalb hilft Batch Prediction dabei, Zeit und Kosten zu reduzieren, indem mehrere Datenpunkte gleichzeitig verarbeitet werden.
Was ist Batch Prediction?
Batch Prediction auf Vertex AI ermöglicht es, viele Anfragen in einem einzigen Job zu verarbeiten, anstatt einzelne API-Aufrufe zu machen. Vorteile gegenüber Online-Prediction:
- Kostenersparnis: Bis zu 50% günstiger als Einzelanfragen
- Höherer Durchsatz: Tausende von Anfragen parallel verarbeitet
- Keine Rate-Limits: Keine Begrenzung der Anfragen pro Minute
- Asynchron: Job läuft im Hintergrund, Ergebnisse werden in Cloud Storage gespeichert
Eingabedaten vorbereiten
Batch Prediction erwartet die Eingabedaten im JSONL-Format (JSON Lines), wobei jede Zeile eine einzelne Anfrage darstellt:
import json
products = [
{"id": "P001", "name": "Glasvase, mundgeblasen, 30cm"},
{"id": "P002", "name": "Edelstahl-Wasserkocher, 1.7L"},
{"id": "P003", "name": "Keramikteller-Set, 6-teilig"},
# ... 13'000+ Produkte
]
with open("batch_input.jsonl", "w") as f:
for product in products:
request = {
"request": {
"contents": [{
"role": "user",
"parts": [{
"text": f"""Classify the fragility of this product
on a scale of 1-4:
1 = very robust, 2 = somewhat robust,
3 = somewhat fragile, 4 = very fragile
Product: {product['name']}"""
}]
}],
"generation_config": {
"response_mime_type": "application/json",
"response_schema": {
"type": "object",
"properties": {
"fragility_score": {
"type": "integer"
},
"reasoning": {
"type": "string"
}
},
"required": ["fragility_score", "reasoning"]
}
}
}
}
f.write(json.dumps(request) + "\n")Daten nach Cloud Storage hochladen
Die JSONL-Datei muss in einem Google Cloud Storage Bucket liegen:
from google.cloud import storage
storage_client = storage.Client()
bucket = storage_client.bucket("dein-bucket-name")
blob = bucket.blob("batch_prediction/input/batch_input.jsonl")
blob.upload_from_filename("batch_input.jsonl")Batch Prediction Job starten
Nun starten wir den Batch Prediction Job über den GenAI Client:
from google import genai
from google.genai.types import CreateBatchJobConfig
client = genai.Client(
vertexai=True,
project="dein-projekt-id",
location="us-central1"
)
batch_job = client.batches.create(
config=CreateBatchJobConfig(
model="gemini-2.0-flash",
src="gs://dein-bucket/batch_prediction/input/batch_input.jsonl",
dest="gs://dein-bucket/batch_prediction/output/",
)
)
print(f"Job gestartet: {batch_job.name}")
print(f"Status: {batch_job.state}")Job-Status prüfen und Ergebnisse abrufen
Der Job läuft asynchron. Wir können den Status regelmässig prüfen:
import time
while batch_job.state.name not in ("JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED"):
time.sleep(30)
batch_job = client.batches.get(name=batch_job.name)
print(f"Status: {batch_job.state.name}")
print(f"Job abgeschlossen: {batch_job.state.name}")Die Ergebnisse werden als JSONL-Datei im angegebenen Cloud Storage Pfad gespeichert und können dann heruntergeladen und weiterverarbeitet werden:
# Ergebnisse herunterladen und verarbeiten
output_blob = bucket.blob("batch_prediction/output/predictions.jsonl")
output_blob.download_to_filename("predictions.jsonl")
results = []
with open("predictions.jsonl", "r") as f:
for line in f:
result = json.loads(line)
response_text = result["response"]["candidates"][0] \
["content"]["parts"][0]["text"]
parsed = json.loads(response_text)
results.append(parsed)
print(f"Total klassifiziert: {len(results)}")
print(f"Beispiel: {results[0]}")Fazit
Batch Prediction mit Gemini ist ein leistungsstarkes Werkzeug, wenn grosse Datenmengen klassifiziert oder gelabelt werden müssen. In unserem Fall konnten wir über 13'000 Produkte effizient und kosteneffektiv mit Zerbrechlichkeitswerten versehen – ein Prozess, der manuell Wochen gedauert hätte und mit Einzelanfragen deutlich teurer gewesen wäre.
Die Kombination von strukturierten Ausgaben (aus dem vorherigen Artikel) und Batch Prediction bietet eine skalierbare Lösung für die automatisierte Datenanreicherung in Unternehmen.