In meiner mehrjährigen Arbeit als Solutions Architect habe ich hunderte Dokumentenverarbeitungs-Pipelines entwickelt und deployed. Die größte Herausforderung war immer dieselbe: Wie extrahiert man zuverlässig strukturierte Daten aus heterogenen, komplexen Dokumenten — Rechnungen, Verträge, technische Spezifikationen, gescannte Archive?.
Die Antwort liegt in der Kombination von Optical Character Recognition (OCR) mit Large Language Models (LLM). In diesem Guide zeige ich Ihnen eine produktionsreife Architektur, die wir bei HolySheep AI in über 50 Enterprise-Projekten validiert haben.
Die Architektur: Warum OCR + LLM?
Traditionelle OCR-Lösungen scheitern an:
- Handgeschriebenen Anmerkungen und Stempeln
- Tabellen mit zusammengeführten Zellen
- Mehrsprachigen Dokumenten mit unterschiedlichen Schriften
- Unstrukturierten Layouts ohne klare Raster
Die Kombination mit LLM löst diese Probleme, indem das Modell:
- Kontext versteht und Ambiguitäten auflöst
- Struktur aus scheinbar unstrukturierten Daten extrahiert
- Domänenwissen für Validierung nutzt
- Fehlerhafte OCR-Ergebnisse korrigiert
Systemarchitektur im Detail
Komponentenübersicht
"""
Produktionsreife OCR + LLM Pipeline
Architektur: Async Processing mit Retry-Logic und Circuit Breaker
"""
import asyncio
import base64
import hashlib
import json
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from concurrent.futures import ThreadPoolExecutor
HolySheep AI SDK - Offizielle API
import openai
Konfiguration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1" # WICHTIG: Niemals api.openai.com!
@dataclass
class DocumentMetadata:
"""Metadaten für jedes Dokument"""
document_id: str
filename: str
page_count: int
file_size_bytes: int
mime_type: str
extracted_at: str = ""
processing_time_ms: float = 0.0
confidence_score: float = 0.0
class ProcessingStatus(Enum):
"""Verarbeitungsstatus"""
PENDING = "pending"
OCR_IN_PROGRESS = "ocr_in_progress"
LLM_PARSING = "llm_parsing"
VALIDATING = "validating"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class ExtractionResult:
"""Ergebnis der Dokumentenextraktion"""
status: ProcessingStatus
raw_text: str = ""
structured_data: dict = field(default_factory=dict)
errors: list = field(default_factory=list)
metadata: Optional[DocumentMetadata] = None
OCR-Modul mit Qualitätssicherung
class OCREngine:
"""
OCR-Engine mit Multi-Backend-Support und Qualitätsvalidierung
Unterstützt: Tesseract, Google Vision API, AWS Textract
"""
def __init__(self, config: dict):
self.config = config
self.quality_threshold = config.get("quality_threshold", 0.85)
self._setup_engines()
def _setup_engines(self):
"""Initialisiere verfügbare OCR-Engines"""
self.engines = {}
# Versuche Tesseract zu initialisieren
try:
import pytesseract
self.engines["tesseract"] = pytesseract
print("[✓] Tesseract OCR geladen")
except ImportError:
print("[!] Tesseract nicht verfügbar")
# Cloud-basierte Engines (optional)
self.cloud_engines = {
"google_vision": self._google_vision_fallback,
"aws_textract": self._aws_textract_fallback,
}
async def extract_text(
self,
image_path: str,
language: str = "deu+eng"
) -> tuple[str, float]:
"""
Extrahiere Text aus Bild mit automatischer Engine-Auswahl
Returns:
Tuple von (extrahierter_text, konfidenzwert)
"""
start_time = time.time()
# Versuche primäre Engine (Tesseract)
if "tesseract" in self.engines:
text, confidence = await self._tesseract_extract(image_path, language)
# Prüfe Qualitätsschwelle
if confidence >= self.quality_threshold:
elapsed = (time.time() - start_time) * 1000
print(f"[✓] OCR abgeschlossen in {elapsed:.0f}ms (Konfidenz: {confidence:.2%})")
return text, confidence
# Fallback auf Cloud-Engines bei niedriger Qualität
for engine_name, engine_func in self.cloud_engines.items():
try:
text, confidence = await engine_func(image_path)
if confidence >= self.quality_threshold:
return text, confidence
except Exception as e:
print(f"[!] {engine_name} fehlgeschlagen: {e}")
continue
# Letzter Fallback: Niedrigqualitative Extraktion
return text, confidence
async def _tesseract_extract(
self,
image_path: str,
language: str
) -> tuple[str, float]:
"""Tesseract-basierte Extraktion mit Post-Processing"""
def _sync_extract():
from PIL import Image
import pytesseract
img = Image.open(image_path)
# Erweiterte Tesseract-Konfiguration
custom_config = r"""
--oem 3
--psm 1
-l {language}
""".format(language=language)
# Extrahiere Text
raw_text = pytesseract.image_to_string(
img,
config=custom_config
)
# Berechne Konfidenz
data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)
confidences = [
float(c) for c in data['conf']
if c != '-1' and c != ''
]
avg_confidence = sum(confidences) / len(confidences) / 100 if confidences else 0
return raw_text, avg_confidence
# Async wrapper für Blocking OCR
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, _sync_extract)
async def _google_vision_fallback(self, image_path: str) -> tuple[str, float]:
"""Google Vision API Fallback"""
# Implementierung für Google Vision
# Hier API-Key und Credentials konfigurieren
pass
async def _aws_textract_fallback