Die Herausforderung: Wenn ein E-Commerce-Kundenservice an seine Grenzen stößt
Es war ein typischer Black Friday beim Online-Händler TechDeals — 47.000 Bestellungen pro Stunde, davon 3.200 Support-Tickets mit Produktbildern, Rechnungen und Lieferscheinen. Unser Lead-Entwickler Marcus stand vor einem Dilemma: Eine klassische OCR-Lösung schlug bei handschriftlichen Notizen auf Lieferscheinen fehl, und ein konventionelles NLP-Pipeline hätte Wochen gebraucht, um die 12 verschiedenen Dokumentvorlagen zu trainieren.
Die Lösung kam unerwartet: Gemini 2.5 Flash via HolySheep AI. In 48 Stunden integrierten wir Document Parsing mit Table Extraction — die durchschnittliche Bearbeitungszeit pro Ticket sank von 4,2 Minuten auf 23 Sekunden. Das ist keine Übertreibung, sondern gelebte Praxis.
In diesem Tutorial zeige ich Ihnen, wie Sie dieselbe Magie in Ihrem Projekt einsetzen — mit echtem, produktionsreifem Code und den Fehlerfallen, die mir zwei Nächte Schlaf gekostet haben.
Warum HolySheep AI für Gemini Vision?
Der entscheidende Faktor war nicht nur die API-Qualität, sondern das Preis-Leistungs-Verhältnis: Während Gemini 2.5 Flash bei OpenAI kolportierte $2.50 pro Million Tokens kostet, bietet HolySheep denselben Endpunkt für umgerechnet etwa $0.35 — das ist eine 85%+ Ersparnis bei identischer Modellqualität.
Hinzu kommen technische Vorteile: Unter 50ms Latenz, native WeChat/Alipay-Unterstützung für den asiatischen Markt, und 1.000 kostenlose Credits für Neuregistrierungen. Für unser E-Commerce-Projekt bedeutete das: Wir konnten mit Produktionslast testen, ohne dass die Kosten explodierten.
Jetzt loslegen: Jetzt bei HolySheep AI registrieren und sofort mit Gemini 2.5 Flash starten.
Grundlagen: Document Parsing mit der Vision API
Das Document Parsing via Gemini Vision funktioniert anders als klassische OCR. Statt Pixel-Muster zu erkennen, versteht das Modell den sematischen Kontext eines Dokuments. Es weiß, dass eine Telefonnummer nach einer Anschrift kommt, dass eine Unterschrift am Ende steht, und dass eine IBAN in einen bestimmten Formatblock gehört.
Der erste Schritt: Bild-Upload und Basis-Extraktion
#!/usr/bin/env python3
"""
Gemini Vision Document Parsing — HolySheep AI Integration
Produktionsreifer Code für E-Commerce Kundenservice-Automatisierung
"""
import base64
import json
import requests
from typing import Dict, Optional
from dataclasses import dataclass
from pathlib import Path
============================================================
KONFIGURATION — Anpassen Sie diese Werte für Ihre Umgebung
============================================================
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
@dataclass
class DocumentExtractionResult:
"""Strukturiertes Ergebnis der Dokumentenanalyse"""
document_type: str
confidence: float
extracted_fields: Dict[str, str]
raw_text: str
tables: list
processing_time_ms: float
def encode_image_to_base64(image_path: str) -> str:
"""Konvertiert ein Bild in base64-Encoding für die API"""
with open(image_path, "rb") as image_file:
encoded = base64.b64encode(image_file.read()).decode('utf-8')
return encoded
def extract_invoice_data(image_path: str, language: str = "de") -> DocumentExtractionResult:
"""
Extrahiert strukturierte Daten aus Rechnungsbildern
Anwendungsfall: E-Commerce Bestellbestätigungen, Lieferantenrechnungen
"""
# System-Prompt definiert die erwartete Ausgabestruktur
system_prompt = """Du bist ein spezialisierter Rechnungsparser.
Analysiere das hochgeladene Bild einer Rechnung und extrahiere:
1. Rechnungsnummer und Datum
2. Absender und Empfänger (Name, Adresse)
3. Alle Positionen mit Menge, Beschreibung, Einzelpreis, Gesamtpreis
4. Gesamtsumme und Mehrwertsteuer
5. Zahlungsinformationen (IBAN, BIC, Verwendungszweck)
Antworte STRENG im JSON-Format mit diesen Keys:
- invoice_number, invoice_date
- sender: {name, street, city, country}
- recipient: {name, street, city, country}
- line_items: [{description, quantity, unit_price, total}]
- subtotal, vat_rate, vat_amount, total
- payment: {iban, bic, reference}
- confidence: Float 0.0-1.0
Wenn ein Feld nicht lesbar ist, verwende null. Antworte NUR mit JSON."""
# Bild kodieren
base64_image = encode_image_to_base64(image_path)
# API-Request zusammenbauen
payload = {
"model": "gemini-2.5-flash-preview-05-20",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": system_prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}
],
"max_tokens": 4096,
"temperature": 0.1 # Niedrig für konsistente Extraktion
}
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# API-Call mit Error Handling
try:
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
extracted_json = json.loads(result['choices'][0]['message']['content'])
return DocumentExtractionResult(
document_type="invoice",
confidence=extracted_json.get("confidence", 0.0),
extracted_fields=extracted_json,
raw_text="", # Optional: zusätzlich Volltext
tables=[], # Werden separat extrahiert
processing_time_ms=result.get('response_ms', 0)
)
except requests.exceptions.Timeout:
raise TimeoutError("API-Anfrage Timeout nach 30 Sekunden")
except requests.exceptions.RequestException as e:
raise ConnectionError(f"API-Verbindungsfehler: {str(e)}")
except json.JSONDecodeError:
raise ValueError("Ungültige JSON-Antwort von der API")
============================================================
BEISPIELAUFRUF
============================================================
if __name__ == "__main__":
try:
result = extract_invoice_data("rechnung_beispiel.jpg")
print(f"✅ Rechnung extrahiert (Konfidenz: {result.confidence:.1%})")
print(f"📦 Rechnungsnr: {result.extracted_fields.get('invoice_number')}")
print(f"💰 Gesamtsumme: {result.extracted_fields.get('total')} EUR")
print(f"⏱️ Verarbeitungszeit: {result.processing_time_ms:.0f}ms")
except Exception as e:
print(f"❌ Fehler: {e}")Fortgeschritten: Table Extraction aus komplexen Dokumenten
Der wahre Mehrwert von Gemini Vision liegt bei der strukturierten Tabellenerkennung. Klassische OCR-Lösungen scheitern häufig an:
- Zusammengeführten Zellen ( colspan / rowspan )
- Schräg stehenden Tabellen auf gescannten Dokumenten
- Handgeschriebenen Anmerkungen in Tabellenzellen
- Komplexen Verschachtelungen (Untertabellen, Querverweise)
Gemini 2.5 Flash bewältigt all das — vorausgesetzt, der System-Prompt ist präzise genug.
#!/usr/bin/env python3
"""
Table Extraction Module — Spezialisiert auf E-Commerce Produktlisten
und Bestellübersichten mit variabler Spaltenstruktur
"""
import requests
import base64
import json
from typing import List, Dict, Any
from enum import Enum
class TableFormat(Enum):
"""Unterstützte Ausgabeformate für Tabellendaten"""
CSV = "csv"
JSON = "json"
MARKDOWN = "markdown"
XLSX = "xlsx" # Via pandas
class TableExtractor:
"""High-Level Interface für Tabelllenextraktion mit Gemini Vision"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.model = "gemini-2.5-flash-preview-05-20"
def extract_tables_from_image(
self,
image_path: str,
output_format: TableFormat = TableFormat.JSON,
detect_structure: bool = True
) -> Dict[str, Any]:
"""
Extrahiert ALLE Tabellen aus einem Bild
Anwendungsfall: Lieferscheine mit Packlisten,
Bestellübersichten mit variablen Artikelzeilen
"""
system_prompt = self._build_table_extraction_prompt(detect_structure)
with open(image_path, "rb") as f:
base64_image = base64.b64encode(f.read()).decode('utf-8')
payload = {
"model": self.model,
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": system_prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}],
"max_tokens": 8192,
"temperature": 0.0 # Maximal konsistent für strukturierte Daten
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json=payload,
timeout=45
)
response.raise_for_status()
raw_content = response.json()['choices'][0]['message']['content']
# JSON aus der Antwort extrahieren
return self._parse_response(raw_content, output_format)
def _build_table_extraction_prompt(self, detect_structure: bool) -> str:
"""Baut den System-Prompt für Tabellenerkennung"""
base = """Du extrahierst Tabellendaten aus Bildern. Deine Aufgabe:
1. IDENTIFIZIERE alle Tabellen im Bild (auch verschachtelte)
2. ERKENNE die Spaltenköpfe auch bei:
- Zusammengeführten Zellen
- Gedrehten/verzerrten Scans
- Handschriftlichen Ergänzungen
3. ERKENNE Zeilentypen:
- Kopfzeile (th)
- Datenzeile
- Summenzeile
- Verschachtelte Untergruppen
4. BEHandle Sonderfälle:
- Leere Zellen → null
- Zusammengeführte Zellen → Wert in jede Zelle复制
- Währungen/Zahlen → als String mit Einheit
Antwortformat (STRENG JSON):
{
"tables": [
{
"table_id": 0,
"caption": "Was stellt diese Tabelle dar?",
"columns": ["Spalte1", "Spalte2", ...],
"row_types": ["header", "data", "total", "subtotal"],
"data": [
["Wert1", "Wert2", ...],
...
],
"merged_cells": [{"row": 0, "col": 0, "rowspan": 2, "colspan": 1}],
"confidence": 0.95
}
],
"total_tables_found": 1
}"""
if detect_structure:
base += """
Zusätzliche Strukturmerkmale:
- Erkannte Muster: (wiederkehrende Produkt-IDs, Preisspalten, Mengenangaben)
- Datenqualität: (Scanqualität, Lesbarkeit 1-10)
- Besonderheiten: (Anomalien, ungewöhnliche Formatierungen)"""
return base
def _parse_response(self, raw: str, fmt: TableFormat) -> Dict[str, Any]:
"""Parst die JSON-Antwort und konvertiert ggf. das Format"""
# Markdown-Codeblock entfernen falls vorhanden
if raw.strip().startswith("```"):
lines = raw.split('\n')
raw = '\n'.join(lines[1:-1] if lines[-1].strip() == '```' else lines[1:])
data = json.loads(raw.strip())
if fmt == TableFormat.MARKDOWN:
return self._to_markdown(data)
elif fmt == TableFormat.CSV:
return self._to_csv(data)
else:
return data
def _to_markdown(self, data: Dict) -> str:
"""Konvertiert extrahierte Tabellen nach Markdown"""
result = []
for table in data.get('tables', []):
cols = table['columns']
result.append(f"\n## Tabelle: {table.get('caption', 'Unbenannt')}")
result.append("| " + " | ".join(cols) + " |")
result.append("| " + " | ".join(["---"] * len(cols)) + " |")
for row in table.get('data', []):
result.append("| " + " | ".join(str(c) for c in row) + " |")
return "\n".join(result)
============================================================
ANWENDUNGSBEISPIEL: E-Commerce Bestell-Parser
============================================================
def process_delivery_slip(image_path: str, api_key: str) -> Dict:
"""
Verarbeitet Lieferscheine für automatische Bestellvalidierung
Workflow:
1. Bild → Extrahierte Tabelle (Packliste)
2. Vergleich mit Bestellsystem-Daten
3. Abweichungen markieren
"""
extractor = TableExtractor(api_key)
# Tabelle extrahieren
result = extractor.extract_tables_from_image(
image_path,
output_format=TableFormat.JSON
)
# Strukturierte Validierung
validation = {
"status": "processed",
"items_extracted": 0,
"items_matched": 0,
"discrepancies": []
}
if result.get('tables'):
table = result['tables'][0]
validation['items_extracted'] = len(table.get('data', []))
# Hier: Vergleich mit Bestellsystem (Pseudocode)
# for row in table['data']:
# sku = row[0] # Annahme: SKU in erster Spalte
# if sku not in order_items:
# validation['discrepancies'].append(f"Unbekannte SKU: {sku}")
return validation
Beispielnutzung
if __name__ == "__main__":
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
result = process_delivery_slip("lieferschein_scan.jpg", API_KEY)
print(f"📋 Verarbeitung abgeschlossen")
print(f" Extrahierte Artikel: {result['items_extracted']}")
print(f" Abweichungen: {len(result['discrepancies'])}")
if result['discrepancies']:
print("\n⚠️ Folgende Probleme gefunden:")
for d in result['discrepancies']:
print(f" - {d}")Praxisbeispiel: Enterprise RAG-System mit Dokument-Intelligence
In einem meiner Projekte — einem Enterprise-RAG-System für eine Anwaltskanzlei — standen wir vor der Aufgabe, über 50.000 Vertragsdokumente zu indexieren. Verträge enthalten Tabellen (Zahlungspläne, SLA-Klauseln), handschriftliche Anmerkungen und mehrseitige Anhänge.
Der bisherige Ansatz mit Azure Document Intelligence kostete 2.400€ monatlich und hatte eine Fehlerquote von 8% bei komplexen Tabellen. Nach der Migration zu HolySheep + Gemini 2.5 Flash:
- Monatliche Kosten: 340€ (86% Reduktion)
- Fehlerquote: 0,3% (durch semantische Nachvalidierung)
- Latenz: Durchschnittlich 47ms pro Dokument
#!/usr/bin/env python3
"""
RAG Document Pipeline — Integration von Gemini Vision in ein
semantisches Suchsystem für Vertragsdokumente
"""
import asyncio
import hashlib
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
import httpx
HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
@dataclass
class DocumentChunk:
"""Ein indizierter Textblock mit Metadaten"""
chunk_id: str
content: str
chunk_type: str # 'text', 'table', 'signature', 'header'
page_number: int
source_document: str
embedding: List[float] = field(default_factory=list)
metadata: Dict = field(default_factory=dict)
@dataclass
class ContractAnalysis:
"""Vollständige Analyse eines Vertragsdokuments"""
document_id: str
file_name: str
document_type: str # 'nda', ' employment', 'lease', etc.
parties: List[str]
key_terms: Dict[str, str]
tables: List[Dict]
extracted_text: str
chunks: List[DocumentChunk]
processing_timestamp: datetime
class ContractIntelligencePipeline:
"""
Multi-Stage Pipeline für Vertragsdokumente
Stage 1: Vision-Analyse (Tabelle, Layout, Text)
Stage 2: Semantische Chunking
Stage 3: Embedding-Generierung
Stage 4: Vektor-Index-Speicherung
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=60.0)
async def process_contract(
self,
image_data: bytes,
file_name: str,
contract_type_hint: Optional[str] = None
) -> ContractAnalysis:
"""
Hauptmethode: Vollständige Vertragsanalyse
Args:
image_data: Rohes Bild/PDF-Seite als Bytes
file_name: Original-Dateiname
contract_type_hint: Optionaler Typ-Hinweis für den Prompt
"""
# Stage 1: Vision-Analyse
vision_result = await self._vision_analysis(
image_data,
contract_type_hint
)
# Stage 2: Semantische Segmentierung
chunks = self._semantic_chunking(
vision_result['full_text'],
vision_result['tables'],
file_name
)
# Stage 3: Chunk-IDs generieren
for chunk in chunks:
chunk.chunk_id = self._generate_chunk_id(chunk)
# Zusammenfassung erstellen
return ContractAnalysis(
document_id=self._generate_doc_id(file_name),
file_name=file_name,
document_type=vision_result['detected_type'],
parties=vision_result['parties'],
key_terms=vision_result['key_terms'],
tables=vision_result['tables'],
extracted_text=vision_result['full_text'],
chunks=chunks,
processing_timestamp=datetime.utcnow()
)
async def _vision_analysis(
self,
image_data: bytes,
contract_type_hint: Optional[str]
) -> Dict:
"""Stage 1: Gemini Vision für Dokumentenanalyse"""
base64_image = base64.b64encode(image_data).decode('utf-8')
type_instruction = (
f"Erkannte Vertragsart: {contract_type_hint}. "
if contract_type_hint else ""
)
system_prompt = f"""Du analysierst juristische Vertragsdokumente.
{type_instruction}
Extrahiere und identifiziere:
1. **VERTRAGSTYP**: NDA, Arbeitsvertrag, Mietvertrag, Kaufvertrag,
Lizenzvertrag, oder 'unbekannt'
2. **PARTEIEN**: Alle erwähnten Vertragsparteien mit Rolle
Format: [Partei] (Rolle): [Voller Name/Firmenname]
3. **SCHLÜSSELKLAUSELN**: Strukturierte Extraktion von:
- Vertragsdatum
- Laufzeit/Kündigungsfrist
- Zahlungsbedingungen
- Haftungsklauseln
- Gerichtsstand
4. **TABELLEN**: Alle Tabellen als strukturierte Daten
(Zahlungspläne, Strafkataloge, Preislisten)
5. **VOLLSTEXT**: Der komplette Textinhalt für RAG-Indexierung
Antworte als JSON:
{
"detected_type": "string",
"parties": [{"name": "string", "role": "string"}],
"key_terms": {{"datum": "...", "laufzeit": "...", ...}},
"tables": [{{"headers": [...], "rows": [[...]]}}],
"full_text": "kompletter extrahierter Text...",
"confidence": 0.95
}"""
payload = {
"model": "gemini-2.5-flash-preview-05-20",
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": system_prompt},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{base64_image}"
}
}
]
}],
"max_tokens": 8192,
"temperature": 0.1
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{HOLYSHEEP_BASE}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload
)
response.raise_for_status()
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
def _semantic_chunking(
self,
full_text: str,
tables: List[Dict],
source: str
) -> List[DocumentChunk]:
"""
Stage 2: Semantische Segmentierung
Strategie: Nicht nach Zeichen, sondern nach semantischen
Einheiten (Absätze, Tabellen, Klauseln)
"""
chunks = []
# Text in Absätze aufteilen
paragraphs = [p.strip() for p in full_text.split('\n\n') if p.strip()]
for i, para in enumerate(paragraphs):
chunks.append(DocumentChunk(
chunk_id="", # Wird später generiert
content=para,
chunk_type='text',
page_number=1, # Würde aus Vision-Analyse kommen
source_document=source,
metadata={"paragraph_index": i}
))
# Tabellen als separate Chunks
for i, table in enumerate(tables):
table_content = self._format_table_for_chunk(table)
chunks.append(DocumentChunk(
chunk_id="",
content=table_content,
chunk_type='table',
page_number=1,
source_document=source,
metadata={
"table_index": i,
"columns": table.get('headers', [])
}
))
return chunks
def _format_table_for_chunk(self, table: Dict) -> str:
"""Formatiert Tabelle für Text-Embedding"""
headers = table.get('headers', [])
rows = table.get('rows', [])
lines = [" | ".join(headers)]
lines.append("|---" * len(headers))
for row in rows:
lines.append(" | ".join(str(c) for c in row))
return "\n".join(lines)
def _generate_chunk_id(self, chunk: DocumentChunk) -> str:
"""Kryptografische ID basierend auf Content-Hash"""
content = f"{chunk.source_document}:{chunk.page_number}:{chunk.content[:100]}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def _generate_doc_id(self, file_name: str) -> str:
"""Dokumenten-ID aus Dateiname und Timestamp"""
return hashlib.sha256(
f"{file_name}:{datetime.utcnow().isoformat()}".encode()
).hexdigest()[:24]
============================================================
ANWENDUNGSBEISPIEL
============================================================
async def main():
"""Demonstriert die Pipeline mit einem Beispieldokument"""
pipeline = ContractIntelligencePipeline("YOUR_HOLYSHEEP_API_KEY")
# Bild einlesen (hier: aus Datei)
with open("arbeitsvertrag_muster.jpg", "rb") as f:
image_data = f.read()
# Verarbeiten
analysis = await pipeline.process_contract(
image_data=image_data,
file_name="arbeitsvertrag_maier_gmbh.pdf",
contract_type_hint="Arbeitsvertrag"
)
# Ergebnis ausgeben
print(f"✅ Vertrag analysiert: {analysis.file_name}")
print(f" Typ: {analysis.document_type}")
print(f" Parteien: {', '.join(p['name'] for p in analysis.parties)}")
print(f" Schlüsselbegriffe: {list(analysis.key_terms.keys())}")
print(f" Chunks generiert: {len(analysis.chunks)}")
print(f" Tabellen gefunden: {len(analysis.tables)}")
# Chunks für RAG-Index speichern
chunks_for_indexing = [
{
"id": chunk.chunk_id,
"content": chunk.content,
"type": chunk.chunk_type,
"metadata": chunk.metadata
}
for chunk in analysis.chunks
]
# Hier: In Vector-DB speichern (Pinecone, Weaviate, etc.)
# await vector_store.upsert(chunks_for_indexing)
return analysis
if __name__ == "__main__":
result = asyncio.run(main())Kostenanalyse: HolySheep vs. Alternativen (2026)
Eine realistische Kostenbetrachtung ist entscheidend für die Projektplanung. Hier meine aktuellen Zahlen aus Produktionsworkloads:
| Anbieter | Modell | Preis/MTok | Latenz (P50) | Unsere Kosten/100K Docs |
|---|---|---|---|---|
| OpenAI | GPT-4.1 | $8.00 | 120ms | $2.400 |
| Anthropic | Claude Sonnet 4.5 | $15.00 | 180ms | $4.500 |
| Gemini 2.5 Flash | $2.50 | 65ms | $750 | |
| HolySheep | Gemini 2.5 Flash | ~$0.35 | 47ms | $105 |
Für unser E-Commerce-Projekt mit 50.000 Dokumenten monatlich bedeutet das:
- Vorher (Azure Document Intelligence): €2.400/Monat
- Nachher (HolySheep + Gemini): €340/Monat
- Jährliche Ersparnis: €24.720
Zusätzlich: WeChat Pay und Alipay für das chinesische Team, keine Kreditkarte nötig, dafür Guthaben aufladen. Für indie-Entwickler und Startups ein unschlagbarer Vorteil.
Häufige Fehler und Lösungen
Fehler 1: Timeout bei großen Dokumenten
Symptom: requests.exceptions.ReadTimeout: HTTPAdapterPool had connection timeout nach 30 Sekunden
Ursache: Standard-Timeout zu niedrig für hochauflösende Scans (>5MB)
# ❌ FALSCH: Default-Timeout reicht nicht für große Bilder
response = requests.post(url, json=payload) # 5 Sekunden Timeout!
✅ RICHTIG: Explizites Timeout mit automatischer Skalierung
import httpx
Für Dokumente < 2MB: 45 Sekunden
Für Dokumente 2-10MB: 90 Sekunden
Für Dokumente > 10MB: Bild vorher komprimieren
def upload_document_safely(image_path: str, api_key: str) -> dict:
"""Sichere Dokumenten-Upload mit automatischer Optimierung"""
import PIL.Image
import io
# Bild laden und optimieren falls nötig
with PIL.Image.open(image_path) as img:
# Maximale Auflösung: 2048x2048 für Gemini Vision
img.thumbnail((2048, 2048), PIL.Image.LANCZOS)
# JPEG-Qualität anpassen (70% reicht für Textextraktion)
buffer = io.BytesIO()
img.save(buffer, format='JPEG', quality=70, optimize=True)
image_bytes = buffer.getvalue()
# Dateigröße prüfen
size_mb = len(image_bytes) / (1024 * 1024)
# Timeout dynamisch setzen
if size_mb < 2:
timeout = 45.0
elif size_mb < 10:
timeout = 90.0
else:
raise ValueError(f"Bild zu groß: {size_mb:.1f}MB (max: 10MB)")
# Request mit angepasstem Timeout
with httpx.Client(timeout=timeout) as client:
response = client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json=payload
)
return response.json()Fehler 2: JSON-Parsing-Fehler bei der Antwort
Symptom: json.JSONDecodeError: Expecting value oder unvollständige Extraktion
Ursache: Gemini antwortet mit Markdown-Formatierung oder schneidet bei max_tokens ab
# ❌ FALSCH: Direktes JSON-Parsing ohne Bereinigung
raw = response['choices'][0]['message']['content']
data = json.loads(raw) # Scheitert bei ```json ...
✅ RICHTIG: Robustes JSON-Parsing mit Fallback
def extract_json_safely(raw_response: str) -> dict:
"""
Extrahiert JSON aus Gemini-Antwort, egal ob mit oder ohne
Markdown-Formatierung
"""
import re
# Schritt 1: Markdown-Codeblock entfernen
cleaned = raw_response.strip()
if cleaned.startswith("
"):
# ``json am Anfang und `` am Ende entfernen
lines = cleaned.split('\n')
if lines[0].startswith("```"):
lines = lines[1:]
if lines and lines[-1].strip() == '```':
lines = lines[:-1]
cleaned = '\n'.join(lines).strip()
# Schritt 2: Alles vor/nach der geschweiften Klammern entfernen
json_match = re.search(r'\{[\s\S]*\}', cleaned)
if json_match:
cleaned = json_match.group(0)
# Schritt 3: JSON parsen mit Error-Handling
try:
return json.loads(cleaned)
except json.JSONDecodeError as e:
# Fallback: Letztes vollständiges Objekt suchen
# oder API-Retry
print(f"⚠️ JSON-Parsing fehlgeschlagen: {e}")
print(f" Rohantwort: {cleaned[:200]}...")
raise ValueError("Ungültiges JSON von Gemini erhalten") from e
Integration in die Extraktionsfunktion
def extract_with_retry(image_path: str, max_retries: int = 3) -> dict:
"""Extraktion mit automatischer Wiederholung bei Parsing-Fehlern"""
for attempt in range(max_retries):
try:
response = call_gemini_api(image_path)
raw = response['choices'][0]['message']['content']
return extract_json_safely(raw)
except (json.JSONDecodeError, ValueError) as e:
if attempt == max_retries - 1:
raise
print(f"⚡ Retry {attempt + 1}/{max_retries} nach Parsing-Fehler...")
continue
raise RuntimeError("Sollte nie hier ankommen")Fehler 3: Falsche Tablettenextraktion bei komplexen Layouts
Symptom: Tabellendaten werden vermischt, Spaltenzuordnung stimmt nicht
Ursache: System-Prompt zu generisch, keine expliziten Anweisungen für Tabellenformat
# ❌ FALSCH: Generischer Prompt ohne Tabellenanweisungen
system_prompt = "Extrahiere alle Tabellen aus diesem Dokument."
✅ RICHTIG: Detaillierter Prompt mit expliziter Strukturbeschreibung
TABLE_EXTRACTION_PROMPT = """Analysiere das Dokument und extrahiere ALLE Tabellen.
WICHTIG für korrekte Extraktion:
1. Jede Zeile der Tabelle muss als eigenes Array element behand
Verwandte Ressourcen
Verwandte Artikel