Vertragsintelligenz mit OCR, hybridischem RAG und LangGraph
Ein Pipeline zur Vertragsextraktion: Bildreinigung, PaddleOCR mit fallback auf GPT-4o, FAISS in Kombination mit BM25 zur Suche, mehrfaches RAG-Verfahren sowie ein bewerteter Excel-Bericht.
Das Problem mit Verträgen
Verträge bilden die Grundlage jeder Organisation – von Lieferverträgen und Serviceaufträgen bis hin zu staatlichen Ausschreibungen. In einem mittelgroßen Unternehmen können gleichzeitig Hunderte solcher Verträge aktiv sein, in einem großen Konzern sogar Zehntausende. Dennoch bleiben die darin enthaltenen geschäftlichen Informationen in unstrukturierten Dateien gespeichert: gescannte PDFs, fotografierte Seiten, XML-Exporte aus Einkaufstools. Jeder, der die Zahlungsbedingungen eines Lieferanten oder das Ablaufdatum eines Vertrags erfahren möchte, muss das Dokument öffnen, Seiten voller rechtlicher Standardformulierungen durchlesen und die Antwort anschließend in eine Tabellenkalkulation eingeben.
Diese Routine führt zu drei Problemen:
- Fristen und Verlängerungen werden übersehen, weil die wichtigen Klauseln tief im Text versteckt sind.
- Inkonsistente Preise der Lieferanten bleiben in großem Umfang unentdeckt und verursachen finanzielle Verluste.
- Manuell erfasste Zahlungsbedingungen schaffen Compliance-Risiken.
Das grundlegende Problem besteht darin, dass Vertragsdaten in Dokumenten und nicht in Datenbanken gespeichert sind. Eine Frage, die mit einer einzigen SQL-Abfrage beantwortet werden könnte, erfordert von einem Mitarbeiter etwa 20 Minuten.
Das System automatisiert diese Überprüfung von Anfang bis Ende. Ein Vertrag wird als PDF, Scan oder XML eingegeben, und bereits Sekunden später listet ein Excel-Bericht die wichtigsten kommerziellen Bedingungen auf – jeweils mit einer Zuverlässigkeitsbewertung und einer Seitenreferenz. PaddleOCR, FAISS, BM25, Reciprocal Rank Fusion und GPT-4o laufen alle innerhalb einer einzigen LangGraph-Zustandsmaschine.
Verschiedene Teams profitieren auf unterschiedliche Weise. Die Einkaufsabteilung prüft Verträge schneller und kann Lieferanten früher aufnehmen. Die Finanzabteilung erkennt Preisdiskrepanzen zwischen Verträgen mit verschiedenen Lieferanten. Die Rechts- und Compliance-Abteilung bewahrt prüfbare Aufzeichnungen mit Quellenreferenzen auf. Die Operations-Abteilung verfolgt Verlängerungen und Fristen ohne manuelle Aufwand.
Die Implementierung umfasst folgende Techniken:
- Eine mehrstufige LangGraph-DAG für die Dokumentverarbeitung.
- Bildverbesserung (Rauschentfernung, CLAHE, Ausrichtungskorrektur) vor der OCR-Verarbeitung zur Steigerung der Genauigkeit.
- Eine OCR-Lösung mit doppelter Strategie: PaddleOCR als Hauptmethode mit GPT-4V als Backup.
- Hybrid-RAG, bei dem FAISS, BM25 und Reciprocal Rank Fusion gemeinsam besser abschneiden als jede einzelne Methode für die Informationsabrufung.
- Mehrfachabfragen im RAG-Verfahren, die die Trefferquote bei unterschiedlichen Vertragsformulierungen maximieren.
- Auswertung mehrseitiger Tariflisten mithilfe einer von LLMs gesteuerten Schema-Kanonisierung.
- Ausgaben mit Zuverlässigkeitsbewertung und Nachvollziehbarkeit, geeignet für den Unternehmensgebrauch.
Die fertige Anwendung
Die fertige Anwendung verarbeitet PDF-, Bild- (JPG, PNG, TIFF) sowie XML-Verträge. Die Seitenbilder werden durch Rauschentfernung, Kontrastanpassung und Ausrichtungskorrektur bereinigt, anschließend liest PaddleOCR sie ein, wobei für spärlich beschriftete Seiten ein GPT-4o Vision-Fallback genutzt wird. Die Suche erfolgt mithilfe eines hybriden FAISS- und BM25-Index, der durch Reciprocal Rank Fusion zusammengeführt wird, und multi-query RAG mit Zuverlässigkeitsbewertung extrahiert sieben Vertragsfelder. Die Ergebnisse werden in einem farbgekodierten Excel-Bericht mit den extrahierten Feldern, einer Gebührenübersicht sowie dem rohen OCR-Text dargestellt, während eine Streamlit-Oberfläche diese zusammen mit einem detaillierten Protokoll der Eingaben zur vollständigen Transparenz anzeigt.
Der gesamte Prozess in einem Überblick:
Upload contract (PDF / Image / XML)
│
▼
┌─────────────────────┐
│ 1. Preprocessing │ ← Rasterize, enhance, deduplicate pages
└─────────────────────┘
│
▼
┌─────────────────────┐
│ 2. OCR Extraction │ ← PaddleOCR + GPT-4V fallback
└─────────────────────┘
│
▼
┌─────────────────────┐
│ 3. Indexing │ ← Chunk → FAISS (dense) + BM25 (sparse)
└─────────────────────┘
│
▼
┌─────────────────────┐
│ 4. Extraction │ ← Multi-query RAG → GPT-4o → JSON fields
└─────────────────────┘
│
▼
┌─────────────────────┐
│ 5. Excel Output │ ← Color-coded 3-sheet workbook
└─────────────────────┘
Projektstruktur
Das Repository ist wie folgt organisiert.
agentic-ai-usecases/medium/contract-intelligence/
├── app.py # Streamlit entry point
├── graph.py # LangGraph pipeline definition
├── config/
│ └── settings.py # Centralized config (models, thresholds, paths)
├── models/
│ └── state.py # ContractState TypedDict
├── nodes/
│ ├── preprocessing.py # Image enhancement + deduplication
│ ├── ocr_extraction.py # PaddleOCR + GPT-4V fallback
│ ├── indexing.py # Chunk splitting + FAISS/BM25 indexing
│ ├── extraction_agent.py # Multi-query RAG + GPT-4o extraction
│ └── excel_generation.py # Multi-sheet Excel export
├── services/
│ ├── embeddings.py # Singleton sentence-transformers
│ ├── vector_store.py # HybridVectorStore (FAISS + BM25 + RRF)
│ └── llm.py # LLMService with retry + token tracking
└── utils/
├── ocr_cache.py # Smart OCR result caching
├── file_utils.py # PDF → images, native text extraction
└── progress.py # Thread-safe progress queue for UI
Schritt 0: Einrichtung der Umgebung
Beginnen Sie mit einer virtuellen Umgebung, um Abhängigkeiten zu isolieren. Unter macOS oder Linux:
python -m venv venv
source venv/bin/activate
Auf Windows:
python -m venv venv
venv\Scripts\activate
Dann installieren Sie die Bibliotheken für OCR, Suchfunktionen, Embeddings, Orchestrierung von LLMs sowie die Streamlit-Schnittstelle.
pip install -r requirements.txt
Die wichtigsten Pakete in requirements.txt sehen so aus.
langgraph>=1.0.7 # Workflow orchestration
langchain>=0.2.0 # LLM application framework
langchain-community # Community integrations
langchain-text-splitters # Document chunking utilities
openai>=1.0.0 # OpenAI API client
paddlepaddle # Paddle deep learning backend
paddleocr # OCR engine for PDFs/scanned docs
sentence-transformers # Embedding generation
faiss-cpu # Vector similarity search
rank-bm25 # Keyword-based retrieval
PyMuPDF # PDF parsing/rendering
opencv-python-headless # Image preprocessing
Pillow # Image manipulation
imagehash # Duplicate image detection
openpyxl # Excel export generation
streamlit>=1.28.1 # Web application UI
python-dotenv # Environment variable management
numpy # Numerical operations
pandas # Structured data processing
Eine .env-Datei enthält die OpenAI-Anmeldeinformationen sowie die Auswahl des Modells.
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini # or gpt-4o for higher accuracy
Schritt 1: Zustand der Pipeline und zentrale Konfiguration
LangGraph leitet alles über ein gemeinsames Zustandsobjekt weiter, weshalb es sinnvoll ist, dieses vor dem Erstellen beliebiger Knoten zu definieren. Es dient als einzige Quelle der Wahrheit für die Pipeline.
# models/state.py
from typing import TypedDict, Dict, List, Any
class ContractState(TypedDict, total=False):
# ── Input ──────────────────────────────────────────────────────────────
file_path: str
file_type: str # "pdf" | "image" | "xml"
original_filename: str
# ── Pre-processing ─────────────────────────────────────────────────────
page_image_paths: List[str] # enhanced page image paths
deduplicated_page_indices: List[int] # surviving page indices after dedup
# ── OCR / Text ─────────────────────────────────────────────────────────
raw_text_by_page: Dict[int, str] # page_num → extracted text
full_text: str # concatenated text (all pages)
# ── Indexing ───────────────────────────────────────────────────────────
chunks: List[str]
chunk_metadata: List[Dict[str, Any]]
session_faiss_ready: bool
# ── Extraction ─────────────────────────────────────────────────────────
extracted_fields: Dict[str, Dict[str, Any]]
# ── Output ─────────────────────────────────────────────────────────────
excel_output_path: str
errors: List[str]
# ── Meta ───────────────────────────────────────────────────────────────
current_step: str
processing_log: List[str]
prompt_log: List[Dict[str, Any]]
TypedDict bietet Typhinweise und IDE-Autocomplete, während der Zustand serialisierbar bleibt. Bei total=False ist jedes Feld optional, da jeder Knoten nur das schreibt, was er erzeugt.
Sämtliche einstellbaren Parameter befinden sich in einer einzigen Konfigurationsdatei.
# config/settings.py
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
EMBEDDING_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
EMBEDDING_DIM = 384
CHUNK_SIZE = 600 # Small child chunks for retrieval precision
CHUNK_OVERLAP = 100 # Context preservation across boundaries
TOP_K_RETRIEVAL = 8 # Child chunks to retrieve per field
RRF_K = 60 # Reciprocal rank fusion constant
HIGH_CONFIDENCE = 0.80 # Green threshold
LOW_CONFIDENCE = 0.50 # Red threshold
PDF_DPI = 200 # PDF rasterization resolution
Schritt 2: LLM- und Embedding-Dienste
Das LLM-Modul umschließt die OpenAI-API für Text-, Bild- und strukturierte JSON-Generierung. Es kümmert sich um die Erstellung von Prompts, die Kodierung von Bildern, Wiederholungsversuche, die Überwachung der Token-Nutzung sowie ein fehlerverzeihendes JSON-Parsern, wodurch die Extraktionsworkflows zuverlässig bleiben.
# services/llm.py
import json
import re
import base64
import time
from typing import Optional, List, Tuple
from openai import OpenAI, RateLimitError, APIConnectionError, APIStatusError
from config.settings import OPENAI_API_KEY, OPENAI_MODEL
_RETRYABLE = (RateLimitError, APIConnectionError)
_MAX_RETRIES = 3
_RETRY_DELAY = 5 # seconds
class LLMService:
def __init__(self, model: str = OPENAI_MODEL, api_key: str = OPENAI_API_KEY):
self.model = model
self.client = OpenAI(api_key=api_key)
def _build_messages(self, prompt: str, images: Optional[List[str]] = None) -> list:
if images:
content: list = [{"type": "text", "text": prompt}]
for img_path in images:
with open(img_path, "rb") as f:
b64 = base64.b64encode(f.read()).decode("utf-8")
content.append({
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
})
return [{"role": "user", "content": content}]
return [{"role": "user", "content": prompt}]
def _make_api_call(self, messages: list) -> Tuple[str, int, int]:
"""Returns (content, input_tokens, output_tokens)."""
call_type = "vision" if isinstance(messages[0]["content"], list) else "text"
for attempt in range(1, _MAX_RETRIES + 1):
try:
print(f"[OpenAI] Calling {self.model} ({call_type})...", flush=True)
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
)
print(f"[OpenAI] Done ({self.model})", flush=True)
content = response.choices[0].message.content or ""
usage = response.usage
in_tok = usage.prompt_tokens if usage else 0
out_tok = usage.completion_tokens if usage else 0
return content, in_tok, out_tok
except _RETRYABLE as e:
if attempt < _MAX_RETRIES:
wait = _RETRY_DELAY * attempt
print(f"[OpenAI] {type(e).__name__} — retrying in {wait}s (attempt {attempt}/{_MAX_RETRIES})", flush=True)
time.sleep(wait)
else:
print(f"[OpenAI] Failed after {_MAX_RETRIES} attempts: {e}", flush=True)
raise
except APIStatusError as e:
print(f"[OpenAI] API error {e.status_code}: {e.message}", flush=True)
raise
except Exception as e:
print(f"[OpenAI] Unexpected error: {e}", flush=True)
raise
return "", 0, 0
def generate(self, prompt: str, images: Optional[List[str]] = None) -> str:
messages = self._build_messages(prompt, images)
content, _, _ = self._make_api_call(messages)
return content
def generate_tracked(self, prompt: str, images: Optional[List[str]] = None) -> Tuple[str, int, int]:
"""Returns (text, input_tokens, output_tokens)."""
messages = self._build_messages(prompt, images)
return self._make_api_call(messages)
def generate_json(self, prompt: str, images: Optional[List[str]] = None) -> dict:
full_prompt = (
prompt
+ "\n\nIMPORTANT: Respond ONLY with valid JSON. "
"No markdown code blocks, no explanation, just the JSON object."
)
messages = self._build_messages(full_prompt, images)
content, _, _ = self._make_api_call(messages)
return self._parse_json(content)
def generate_json_tracked(
self, prompt: str, images: Optional[List[str]] = None
) -> Tuple[dict, str, int, int]:
"""Returns (parsed_dict, raw_text, input_tokens, output_tokens)."""
full_prompt = (
prompt
+ "\n\nIMPORTANT: Respond ONLY with valid JSON. "
"No markdown code blocks, no explanation, just the JSON object."
)
messages = self._build_messages(full_prompt, images)
content, in_tok, out_tok = self._make_api_call(messages)
return self._parse_json(content), content, in_tok, out_tok
@staticmethod
def _parse_json(text: str) -> dict:
text = text.strip()
for fence in ("```json", "```"):
if text.startswith(fence):
text = text[len(fence):]
if text.endswith("```"):
text = text[:-3]
text = text.strip()
try:
return json.loads(text)
except json.JSONDecodeError:
match = re.search(r"\{.*\}", text, re.DOTALL)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
return {}
Dienste zur Vektorerzeugung nutzen Sentence Transformers, um aus Vertragstexten dichte Vektoren zu erzeugen. Die Vektoren werden normalisiert, damit die Suche nach semantischer Ähnlichkeit effizient bleibt, und sie dienen sowohl der Indizierung als auch der Abfrage.
# services/embeddings.py
import numpy as np
from sentence_transformers import SentenceTransformer
from config.settings import EMBEDDING_MODEL
class EmbeddingService:
_instance: "EmbeddingService | None" = None
def __init__(self):
self._model = SentenceTransformer(EMBEDDING_MODEL)
@classmethod
def get_instance(cls) -> "EmbeddingService":
if cls._instance is None:
cls._instance = cls()
return cls._instance
def encode(self, texts: list[str], batch_size: int = 32) -> np.ndarray:
vecs = self._model.encode(
texts,
batch_size=batch_size,
normalize_embeddings=True,
show_progress_bar=False,
)
return vecs.astype(np.float32)
def encode_single(self, text: str) -> np.ndarray:
return self.encode([text])[0]
Schritt 3: Vorausverarbeitung
Rohscans von Verträgen weisen in der Regel viel Rauschen auf, haben geringen Kontrast und sind leicht verzerrt. Vor der OCR-Verarbeitung durchläuft jedes Seitenbild drei Verbesserungsstufen.
Bildverbesserung
# nodes/preprocessing.py
import cv2
import numpy as np
# ── Image enhancement ────────────────────────────────────────────────────────
def _deskew(gray: np.ndarray) -> np.ndarray:
"""Estimate rotation from text block contours and correct it."""
_, binary = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU)
contours, _ = cv2.findContours(binary, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
large = [c for c in contours if cv2.contourArea(c) > 200]
if len(large) < 5:
return gray
all_pts = np.vstack([c.reshape(-1, 2) for c in large])
rect = cv2.minAreaRect(all_pts)
angle = rect[-1]
if angle < -45:
angle += 90
if abs(angle) < 0.5:
return gray # already straight
h, w = gray.shape
M = cv2.getRotationMatrix2D((w // 2, h // 2), angle, 1.0)
return cv2.warpAffine(gray, M, (w, h),
flags=cv2.INTER_CUBIC,
borderMode=cv2.BORDER_REPLICATE)
def enhance_image(path: str) -> None:
"""In-place: denoise → CLAHE contrast → deskew."""
img = cv2.imread(path)
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
# Stage 1: Remove salt-and-pepper noise
gray = cv2.fastNlMeansDenoising(gray, h=10)
# Stage 2: Adaptive contrast enhancement (CLAHE)
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
gray = clahe.apply(gray)
# Stage 3: Correct rotation using text block contours
gray = _deskew(gray)
out = cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)
cv2.imwrite(path, out)
Die Phasen erfolgen nacheinander.
Raw scan (noisy, low contrast, rotated)
│
▼
fastNlMeansDenoising (h=10)
→ removes scanner noise, reduces speckles
│
▼
CLAHE (clipLimit=2.0, tileGridSize=(8,8))
→ locally enhances contrast in text areas
│
▼
deskew (via contour-based angle detection)
→ corrects scan rotation < 0.5° threshold ignored
│
▼
Clean, high-contrast, straight image for OCR
Warum CLAHE und nicht einfache Histogrammgleichung? Die Standardgleichung erhöht den Kontrast global und kann Bereiche, die bereits hell sind, übermäßig aufhellen. CLAHE, abgekürzt für Contrast Limited Adaptive Histogram Equalization, arbeitet mit kleinen Bereichen. Text auf verblassten Stempeln oder blassen Überschriften wird verbessert, ohne dass die sauberen Bereiche verfälscht werden.
Beseitigung doppelter Seiten
Lange Verträge enthalten häufig wiederholte Seiten, wie doppelte Titelseiten, leere Trennseiten und Kopien. Die OCR-Verarbeitung all dieser Seiten kostet Zeit und füllt den Index mit doppeltem Text.
# nodes/preprocessing.py
import imagehash
from PIL import Image
# ── Page deduplication (perceptual hash) ────────────────────────────────────
def _dedup_pages(paths: list[str], threshold: int = 6) -> list[int]:
"""Return indices of unique pages using perceptual hashing."""
hashes = []
kept = []
for i, p in enumerate(paths):
h = imagehash.phash(Image.open(p))
# Keep page if its hash differs by more than `threshold` bits from all kept pages
if all((h - prev) > threshold for prev in hashes):
kept.append(i)
hashes.append(h)
return kept
Perzeptionelles Hashing (phash) reduziert jedes Seitenbild auf einen 64-Bit-Fingerabdruck, der aus Frequenzkomponenten erstellt wird. Seiten, deren Hamming-Distanz zu einer früheren Seite 6 oder weniger beträgt, gelten als visuell identisch, sodass sowohl exakte Kopien als auch nahezu identische Kopien, wie beispielsweise eine Seite, die zweimal in unterschiedlicher Qualität gescannt wurde, erkannt werden. Ein Beispiellauf:
Page 1: hash=a3f2... ──► unique, keep ✓
Page 2: hash=a3f2... ──► distance=0 ≤ 6, skip ✗
Page 3: hash=8b14... ──► distance=18 > 6, keep ✓
Page 4: hash=8b14... ──► distance=0 ≤ 6, skip ✗
Page 5: hash=c701... ──► distance=24 > 6, keep ✓
Ein generischer XML-Parser wird ebenfalls hinzugefügt, zusammen mit der Endfunktion, die alle Vorkonvertierungs Schritte nacheinander aufruft.
# nodes/preprocessing.py
# ── XML parser (generic) ─────────────────────────────────────────────────────
def _xml_to_text(xml_path: str) -> str:
tree = ET.parse(xml_path)
root = tree.getroot()
def _recurse(elem: ET.Element, indent: int = 0) -> List[str]:
tag = elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag
text = (elem.text or "").strip()
lines: List[str] = []
prefix = " " * indent
if text:
lines.append(f"{prefix}{tag}: {text}")
else:
lines.append(f"{prefix}{tag}:")
for attr_k, attr_v in elem.attrib.items():
lines.append(f"{prefix} @{attr_k}: {attr_v}")
for child in elem:
lines.extend(_recurse(child, indent + 1))
tail = (elem.tail or "").strip()
if tail:
lines.append(f"{prefix}{tail}")
return lines
return "\n".join(_recurse(root))
# ── Node ─────────────────────────────────────────────────────────────────────
def preprocess_node(state: ContractState) -> dict:
log = list(state.get("processing_log", []))
file_type = state["file_type"]
# XML: no image pipeline needed
if file_type == "xml":
text = _xml_to_text(state["file_path"])
log.append(f"XML parsed → {len(text)} chars")
return {
**state,
"page_image_paths": [],
"deduplicated_page_indices": [],
"raw_text_by_page": {0: text},
"full_text": text,
"processing_log": log,
"current_step": "ocr_extraction",
}
# PDF or image → rasterise
tmp = make_temp_dir()
if file_type == "pdf":
raw_paths = pdf_to_images(state["file_path"], tmp)
else:
dest = os.path.join(tmp, "page_0000.png")
shutil.copy(state["file_path"], dest)
raw_paths = [dest]
# Enhance
for p in raw_paths:
enhance_image(p)
# Deduplicate
kept_indices = _dedup_pages(raw_paths)
kept_paths = [raw_paths[i] for i in kept_indices]
log.append(
f"Pre-processing: {len(raw_paths)} pages, "
f"{len(kept_paths)} kept after dedup"
)
return {
**state,
"page_image_paths": kept_paths,
"deduplicated_page_indices": kept_indices,
"processing_log": log,
"current_step": "ocr_extraction",
}
Schritt 4: Dual-Strategie OCR
OCR ist der riskanteste Schritt, denn ein übersehener Tabellen- oder Formularfeld kann später die Extraktion stören. Der Knoten verwendet zwei Modelle, um den Abdeckungsbereich zu erweitern.
# nodes/ocr_extraction.py
from typing import Dict
from paddleocr import PaddleOCR
from models.state import ContractState
from services.llm import LLMService
from utils.file_utils import extract_native_pdf_text
from utils import progress
# Lazy-initialised singleton so the model loads once per process
_ocr_engine: PaddleOCR | None = None
def _get_ocr() -> PaddleOCR:
global _ocr_engine
if _ocr_engine is None:
# PP-OCRv4 mobile models are ~4x faster than v5 server on CPU.
# All rotation/orientation detection disabled — no autorotation.
_ocr_engine = PaddleOCR(
use_doc_orientation_classify=False,
use_doc_unwarping=False,
use_textline_orientation=False,
text_detection_model_name="PP-OCRv4_mobile_det",
text_recognition_model_name="PP-OCRv4_mobile_rec",
)
return _ocr_engine
def _ocr_image(path: str) -> str:
"""PaddleOCR 3.x uses predict() which returns a generator of result dicts."""
ocr = _get_ocr()
try:
# 3.x API: predict() yields result dicts per image
results = list(ocr.predict(path))
if not results:
return ""
res = results[0]
# Result is a dict-like object; rec_texts holds the text lines
if isinstance(res, dict):
texts = res.get("rec_texts") or []
return "\n".join(str(t) for t in texts if t)
# Fallback: some versions return objects with attributes
if hasattr(res, "rec_texts"):
return "\n".join(str(t) for t in res.rec_texts if t)
return ""
except AttributeError:
# Graceful fallback to 2.x API if somehow an older engine is used
result = ocr.ocr(path, cls=True) # type: ignore[attr-defined]
if not result or result[0] is None:
return ""
return "\n".join(line[1][0] for line in result[0] if line and len(line) >= 2)
Jede Seite folgt diesem Entscheidungsbaum.
For each page image:
│
▼
PaddleOCR (PP-OCRv4 mobile)
│
├── len(text) ≥ 80 chars ──► Use PaddleOCR result ✓
│
└── len(text) < 80 chars (sparse page)
│
▼
GPT-4o Vision model
│
├── len(vision) > len(paddle) ──► Use vision result ✓
│
└── else ──► Keep PaddleOCR result
+ PDF native text layer merge
(use whichever layer has more chars per page)
Warum 80 Zeichen als Grenze? Spärlich beschriftete Seiten enthalten Tabellen mit wenigen Worten pro Zelle, meist leere Formulare sowie Stempel- und Unterschriftenfelder. PaddleOCR kommt mit dichtem Text gut zurecht, versagt jedoch bei stark strukturierten Layouts, während GPT-4o Vision das gesamte Bild erfassen und eine Tabelle mithilfe von Trennzeichen neu erstellen kann.
Der Prompt für die Bildverarbeitung ist absichtlich spezifisch formuliert.
#nodes/ocr_extraction.py
def _vision_extract(llm: LLMService, image_path: str) -> str:
prompt = (
"You are a contract document reader. "
"Extract ALL visible text from this contract page exactly as it appears. "
"Preserve table structure using pipe | separators. "
"Include headers, dates, prices, party names, and all clause text. "
"Do not summarise or omit anything."
)
return llm.generate(prompt, images=[image_path])
Dann werden die Komponenten zu einem vollständigen OCR-Pipeline zusammengefügt.
#nodes/ocr_extraction.py
def _report(page_num: int, total: int, status: str) -> None:
msg = f"[OCR] Page {page_num + 1}/{total}: {status}"
print(msg, flush=True)
progress.post(msg)
def ocr_extraction_node(state: ContractState) -> dict:
log = list(state.get("processing_log", []))
# XML / pre-filled text: skip OCR
if state.get("full_text") or state["file_type"] == "xml":
log.append("OCR skipped – text already available")
return {**state, "processing_log": log, "current_step": "indexing"}
llm = LLMService()
raw_text_by_page: Dict[int, str] = {}
total_pages = len(state.get("page_image_paths", []))
for page_num, img_path in enumerate(state.get("page_image_paths", [])):
_report(page_num, total_pages, "OCR…")
paddle_text = _ocr_image(img_path)
# Use vision model for sparse pages (tables, forms, stamps)
if len(paddle_text.strip()) < 80:
_report(page_num, total_pages, "sparse — running vision model…")
try:
vision_text = _vision_extract(llm, img_path)
page_text = vision_text if len(vision_text) > len(paddle_text) else paddle_text
except Exception:
page_text = paddle_text
else:
page_text = paddle_text
_report(page_num, total_pages, f"done ({len(page_text)} chars)")
raw_text_by_page[page_num] = page_text
# Merge with native PDF text layer when richer
if state["file_type"] == "pdf":
try:
native = extract_native_pdf_text(state["file_path"])
for pg, nat_text in native.items():
existing = raw_text_by_page.get(pg, "")
if len(nat_text.strip()) > len(existing.strip()):
raw_text_by_page[pg] = nat_text
except Exception:
pass
full_text = "\n\n".join(
f"--- Page {pg + 1} ---\n{txt}"
for pg, txt in sorted(raw_text_by_page.items())
)
log.append(f"OCR complete: {len(raw_text_by_page)} pages → {len(full_text)} chars")
return {
**state,
"raw_text_by_page": raw_text_by_page,
"full_text": full_text,
"processing_log": log,
"current_step": "indexing",
}
Schritt 5: Erstellung des hybriden Indexes
Sobald pro Seite sauberer Text vorliegt, folgt eine suchfähige Wissensdatenbank, die eine Eltern-Kind-Strategie zur Aufteilung verwendet.
Aufteilung in Blöcke
# nodes/indexing.py
from typing import List, Dict, Any
from langchain_text_splitters import RecursiveCharacterTextSplitter
from models.state import ContractState
from services.vector_store import HybridVectorStore
from config.settings import CHUNK_SIZE, CHUNK_OVERLAP
# Reset per document — no cross-document persistence.
_session_store: HybridVectorStore | None = None
def get_session_store() -> HybridVectorStore:
return _session_store # type: ignore[return-value]
def _find_page(chunk: str, raw_text_by_page: Dict[int, str]) -> int:
"""Best-effort page number for a chunk (0-indexed)."""
needle = chunk[:60].strip()
for pg, pg_text in raw_text_by_page.items():
if needle in pg_text:
return pg
return 0
def indexing_node(state: ContractState) -> dict:
global _session_store
log = list(state.get("processing_log", []))
text = state.get("full_text", "")
raw_text_by_page = state.get("raw_text_by_page", {})
source_name = state.get("original_filename", "contract")
splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["\n\n", "\n", ". ", "; ", ", ", " "],
)
chunks: List[str] = splitter.split_text(text)
metadata: List[Dict[str, Any]] = []
for i, chunk in enumerate(chunks):
pg = _find_page(chunk, raw_text_by_page)
metadata.append({
"chunk_id": i,
"page": pg,
"source": source_name,
# Full page text stored here — used as parent context during extraction
"page_text": raw_text_by_page.get(pg, chunk),
})
_session_store = HybridVectorStore()
_session_store.add_documents(chunks, metadata)
log.append(f"Indexed {len(chunks)} chunks into session FAISS+BM25")
return {
**state,
"chunks": chunks,
"chunk_metadata": metadata,
"session_faiss_ready": True,
"processing_log": log,
"current_step": "extraction",
}
Die zentrale Idee ist die Suche zwischen Elternteil und Kind.
Full document text
│
▼
Split into small child chunks (600 chars, 100 overlap)
→ precise retrieval targets
│
Each chunk stores its full parent page in metadata
→ rich context window for the LLM
│
▼
At extraction time:
Retrieve small chunks (accurate matching)
│
▼
Expand to full parent pages (complete context)
Kleine Textabschnitte passen präzise zu Suchanfragen, doch wenn dem großen Sprachmodell ein 600-Zeichen-Teiltext übergeben wird, geht oft der umliegende Kontext verloren, der zur Verständnisbildung notwendig ist. Das Speichern der Seiten und das spätere Abrufen derselben gewährleistet sowohl Präzision als auch Kontext.
Der hybride Speicher
HybridVectorStore ist das Herzstück der Informationsabrufung. Er kombiniert zwei grundlegend unterschiedliche Suchmethoden.
# services/vector_store.py
from typing import List, Dict, Any
import faiss
import numpy as np
from rank_bm25 import BM25Okapi
from config.settings import EMBEDDING_DIM, TOP_K_RETRIEVAL, RRF_K
from services.embeddings import EmbeddingService
class HybridVectorStore:
"""FAISS (dense) + BM25 (sparse) with Reciprocal Rank Fusion."""
def __init__(self):
self._emb = EmbeddingService.get_instance()
self._index = faiss.IndexFlatIP(EMBEDDING_DIM)
self.chunks: List[str] = []
self.metadata: List[Dict[str, Any]] = []
self._bm25: BM25Okapi | None = None
# ── Public API ──────────────────────────────────────────────────────────
def add_documents(self, chunks: List[str], metadata: List[Dict[str, Any]]) -> None:
if not chunks:
return
vecs = self._emb.encode(chunks)
self._index.add(vecs)
self.chunks.extend(chunks)
self.metadata.extend(metadata)
self._rebuild_bm25()
def hybrid_search(self, query: str, k: int = TOP_K_RETRIEVAL) -> List[Dict[str, Any]]:
if not self.chunks:
return []
k = min(k, len(self.chunks))
# --- Dense search ---
q_vec = self._emb.encode_single(query).reshape(1, -1)
_, faiss_idxs = self._index.search(q_vec, k)
faiss_ranks: Dict[int, int] = {
int(idx): rank
for rank, idx in enumerate(faiss_idxs[0])
if idx >= 0
}
# --- Sparse search (BM25) ---
bm25_scores = self._bm25.get_scores(query.lower().split())
bm25_top = np.argsort(bm25_scores)[::-1][:k]
bm25_ranks: Dict[int, int] = {int(idx): rank for rank, idx in enumerate(bm25_top)}
# --- RRF fusion ---
all_idxs = set(faiss_ranks) | set(bm25_ranks)
rrf: Dict[int, float] = {}
for idx in all_idxs:
score = 0.0
if idx in faiss_ranks:
score += 1.0 / (RRF_K + faiss_ranks[idx])
if idx in bm25_ranks:
score += 1.0 / (RRF_K + bm25_ranks[idx])
rrf[idx] = score
sorted_idxs = sorted(rrf, key=lambda x: rrf[x], reverse=True)[:k]
return [
{
"chunk": self.chunks[i],
"metadata": self.metadata[i],
"score": rrf[i],
"faiss_rank": faiss_ranks.get(i),
"bm25_rank": bm25_ranks.get(i),
}
for i in sorted_idxs
]
def reset(self) -> None:
self._index = faiss.IndexFlatIP(EMBEDDING_DIM)
self.chunks = []
self.metadata = []
self._bm25 = None
def _rebuild_bm25(self) -> None:
tokenized = [c.lower().split() for c in self.chunks]
self._bm25 = BM25Okapi(tokenized)
Die Kombination beider Methoden lohnt sich, da jede die Schwächen der anderen ausgleicht.
So funktioniert die Reciprocal Rank Fusion (RRF). Die Formel 1/(k + rank) wandelt beliebige Ränge in vergleichbare Werte um. Bei k=60 addiert ein Erstplatzer aus jedem System 1/61, das etwa 0,016 entspricht. Ein Element, das bei FAISS erst und bei BM25 drittplatziert ist, ergibt insgesamt 1/61 + 1/63, also etwa 0,032 – das übertrifft einen Eintrag, den nur ein System gefunden hat. Dies ist nützlich bei Verträgen mit ungewöhnlicher Terminologie, wo BM25 Schwierigkeiten hat, sowie bei kopierten Standardtexten, bei denen FAISS zu viele Treffer liefert.
Schritt 6: Der Extraktionsagent
Hier kommt die Intelligenz ins Spiel. Für jedes der sieben Vertragsfelder sendet der Extraktionsagent mehrere gezielte Abfragen, holt die relevantesten Seiten ab und bittet GPT-4o, den gewünschten Wert herauszufinden.
Die sieben konfigurierbaren Felder
# config/settings.py
EXTRACT_FIELDS = [
"party_a_legal_name", # Commissioning/buying entity
"party_b_legal_name", # Fulfilling/supplying entity
"start_date", # Contract commencement date
"end_date", # Contract termination date
"payment_timeline", # Net payment period (e.g., "Net 30")
"payment_conditions", # Full narrative of payment triggers/rules
"price_details", # Dynamic rate card (multi-page, multi-column)
]
Mehrfachabfragen zur Suche
Jedes Feld erhält vier Abfragevarianten, um die Trefferquote zu maximieren. Eine einzige Abfrage für „Startdatum“ kann einen Vertrag übersehen, in dem „Beginn:“ oder „Gültig ab:“ steht, während mehrere Ansätze alle Varianten erfassen.
# nodes/extraction_agent.py
FIELD_QUERIES: Dict[str, List[str]] = {
"party_a_legal_name": [
"party initiating issuing commissioning contract name",
"contracting authority entity procuring services",
"issued by letterhead header organization name",
"client buyer purchaser government agency department",
],
"party_b_legal_name": [
"party providing fulfilling goods services name",
"vendor contractor manufacturer service provider legal name",
"signed by fulfilling party company registration name",
"seller supplier respondent awarded contractor",
],
"start_date": [
"contract start date commencement effective date",
"agreement begins from date contract period start",
"effective from date of contract",
],
"end_date": [
"contract end date expiry termination date",
"agreement expires valid until contract period end",
"contract completion date",
],
"price_details": [
"rate card price list pricing table fees charges",
"unit price amount cost per item service rate",
"pricing schedule tariff commercial rates",
"fee structure price per unit volume pricing",
],
"payment_timeline": [
"payment terms net days due net 30 net 60 payment schedule",
"invoice settlement period days due payment deadline",
"payment due within days of invoice",
],
"payment_conditions": [
"payment conditions triggers acceptance criteria invoice rules",
"payment upon delivery acceptance milestone conditions",
"invoice approval payment release conditions terms",
],
}
Der untenstehende Abrufablauf wird für jedes Feld wiederholt.
For field "party_a_legal_name":
│
▼
Run 4 queries × hybrid_search(k=6 per query)
├── "party initiating issuing commissioning contract name"
├── "contracting authority entity procuring services"
├── "issued by letterhead header organization name"
└── "client buyer purchaser government agency department"
│
▼
Deduplicate chunks (by first 80 chars)
Sort by RRF score → top-k unique chunks
│
▼
[SPECIAL] Page 0 anchoring for party fields:
Cover/letterhead page always included regardless of score
(party names often only appear there)
│
▼
Expand child chunks → full parent pages
(only pages scoring ≥ 50% of top page score kept)
│
▼
Send page context to GPT-4o for extraction
Ankerung auf Seite null für Parteienamen
In Regierungsaufträgen erscheinen Parteienamen oft nur auf dem Briefkopf oder der Titelseite, nicht in den Hauptklauseln. Wenn ein Bewertungsschwellenwert die Titelseite außer Acht lässt, weil innere Klauseln eine höhere Priorität haben, fehlt der Parteiname. Ein Sonderfall sorgt dafür, dass Seite null für diese Felder weiterhin berücksichtigt wird.
# nodes/extraction_agent.py
PAGE_0_ANCHOR_FIELDS = {"party_a_legal_name", "party_b_legal_name"}
def _retrieve_top_chunks(field: str, store: HybridVectorStore) -> List[Dict]:
# ... standard multi-query retrieval ...
# Always anchor page 0 for party fields
if field in PAGE_0_ANCHOR_FIELDS:
page_0_chunks = store.hybrid_search(
"contract issued by party name letterhead", k=4
)
for r in page_0_chunks:
if r["metadata"].get("page", 999) == 0: # cover page
key = r["chunk"][:80]
if key not in seen:
seen.add(key)
results.append(r)
return results[:k_cap]
Erweiterung auf den übergeordneten Kontext mit Filtern
Bevor der Kontext beim LLM ankommt, werden die Unterabschnitte zu vollständigen übergeordneten Seiten erweitert. Seiten mit niedrigem Score werden entfernt, damit die Anfrage konzentriert bleibt.
# nodes/extraction_agent.py
def _build_parent_context(chunks: List[Dict]) -> Tuple[str, List[int]]:
"""Expand child chunks to full pages. Only keep pages scoring ≥ 50% of top."""
page_best_score: Dict[int, float] = {}
page_texts: Dict[int, str] = {}
for r in chunks:
pg = r["metadata"].get("page", 0)
score = r.get("score", 0.0)
if score > page_best_score.get(pg, -1.0):
page_best_score[pg] = score
if pg not in page_texts:
page_texts[pg] = r["metadata"].get("page_text") or r["chunk"]
top_score = max(page_best_score.values())
threshold = top_score * 0.5
relevant_pages = {pg for pg, sc in page_best_score.items() if sc >= threshold}
context = "\n\n".join(
f"[Full page {pg + 1}]\n{text}"
for pg, text in sorted(relevant_pages)
)
return context, sorted(pg + 1 for pg in relevant_pages)
Die Extraktionsanweisungen
Jedes Feld enthält eine sorgfältig formulierte Anweisung, die GPT-4o mitteilt, wonach es suchen und wie es antworten soll.
# nodes/extraction_agent.py
prompt = f"""{FIELD_PROMPTS[field]}Contract pages:
{context}
Return JSON:
{{
"value": "<extracted value or null>",
"confidence": <0.0–1.0>,
"page_ref": <page number or null>,
"raw_text": "<short verbatim snippet>"
}}
If not found: {{"value": null, "confidence": 0.0, "page_ref": null, "raw_text": ""}}"""
Die Anfrage payment_timeline ist beispielsweise sehr gezielt ausgerichtet.
Extract only the payment timeline: the specific net payment period or day count.
Look for values like 'Net 30', 'Net 60', '45 days', '30 days from invoice date'.
Return the shortest unambiguous value (e.g. 'Net 30', '60 days').
Return null if no explicit day count or net term is stated.
Es unterscheidet sich von payment_conditions, der die gesamte Darstellung enthält.
Extract the full narrative of payment conditions: everything beyond the simple day count.
This includes: triggers for payment (e.g. delivery, acceptance, milestone completion),
invoice submission rules, acceptance criteria, approval workflows, early payment discounts,
and late payment penalties.
Durch Trennung der Inhalte wird verhindert, dass das LLM eine 400-Wörter lange Zahlungsbedingung mit einem einfachen „Net 30“ verwechselt.
Preistabellen über mehrere Seiten
Preistabellen stellen das schwierigste Feld dar. Sie erstrecken sich über mehrere Seiten, die Spaltenbezeichnungen unterscheiden sich von Seite zu Seite („Item Number“ auf einer Seite, „SKU“ auf einer anderen) und die Zeilenzahl liegt bei Dutzenden. Ein einziger Aufruf des LLM für alle Seiten würde den Kontext übersteigen und die Spaltenschemata verunreinigen. Die Lösung besteht darin, pro Seite zu extrahieren und anschließend die Schemata zu kanonisieren.
# nodes/extraction_agent.py
def _canonicalize_columns(page_schemas, llm):
prompt = f"""These column headers were extracted from different pages of the same pricing document.
Some columns are the same concept named differently across pages
(e.g. "Item Numbers", "Item Number", "SKU").Page schemas with one sample row each:
{json.dumps(page_schemas, indent=2)}
Return a flat JSON object mapping every column name variant to its canonical name.
- Group columns that represent the same concept into one canonical name
- Use the most descriptive and clearest name as the canonical
- Map every column that appears in the input
Return JSON: {{"<variant>": "<canonical>", ...}}"""
return llm.generate_json_tracked(prompt)
Die zurückgegebene Karte sieht ungefähr so aus.
{
"Item Number": "SKU",
"Item Numbers": "SKU",
"SKU": "SKU",
"Price": "Unit Cost",
"Unit Cost": "Unit Cost",
"Amount": "Unit Cost"
}
Jede Zeile jeder Seite wird anschließend mithilfe dieser Zuordnung neu verarbeitet und zu einer einheitlichen Tabelle zusammengeführt.
OCR-Caching
Ein 50-seitiger Vertrag erfordert 5 bis 10 Minuten für die OCR-Verarbeitung, und ein solches Warten bei jeder erneuten Ausführung ist lästig, wenn Anfragen angepasst oder Fehler behoben werden müssen. Das Caching erkennt einen früheren Excel-Bericht für dasselbe Dokument und bietet an, direkt zur Extraktionsphase überzugehen.
# utils/ocr_cache.py
def find_cached_ocr(original_filename: str) -> Optional[Path]:
"""Return the most-recent Excel with matching stem that has a Raw OCR Text sheet."""
stem = Path(original_filename).stem.lower()
candidates = sorted(OUTPUT_DIR.glob("*.xlsx"),
key=lambda p: p.stat().st_mtime, reverse=True)
for xlsx in candidates:
# Match stem: "contract_abc_extracted_20260523_..." → "contract_abc"
xlsx_stem = xlsx.stem.lower()
original_part = xlsx_stem.split("_extracted_")[0] if "_extracted_" in xlsx_stem else xlsx_stem
if original_part != stem:
continue
wb = openpyxl.load_workbook(str(xlsx), read_only=True, data_only=True)
if "Raw OCR Text" in wb.sheetnames:
wb.close()
return xlsx
return None
Wenn die Caching-Funktion aktiviert ist, verläuft der Ablauf wie folgt.
Upload file
│
▼
find_cached_ocr() ── no match ──► run full pipeline (OCR included)
│
▼ match found
Show toggle: "⚡ Skip OCR - reuse existing Excel" (default ON)
│
▼ user clicks "Process Contract"
load_ocr_from_excel() ──► pre-fill raw_text_by_page in state
│
▼
ocr_extraction_node sees full_text is set ──► skips OCR entirely
│
▼
Continues to indexing → field extraction → Excel output
Schritt 9: Die Excel-Datei
Die endgültigen Ergebnisse sowie die Rohdaten werden in einer Excel-Datei gespeichert, die auch in eine Datenbank eingespeist werden kann. Die Farben geben das Vertrauensniveau an: Grün für 80 % und mehr (hoch), Gelb für 50 bis 79 % (mittel) und Rot bei weniger als 50 % (niedrig oder nicht gefunden). Diese Werte zeigen an, welche Attribute einer Überprüfung durch einen Menschen benötigen.
#nodes/excel_generation.py
import json
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List
import openpyxl
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from config.settings import (
FIELD_DISPLAY_NAMES,
HIGH_CONFIDENCE,
LOW_CONFIDENCE,
OUTPUT_DIR,
)
from models.state import ContractState
# ── Colour palette ────────────────────────────────────────────────────────────
_GREEN = PatternFill(start_color="C6EFCE", end_color="C6EFCE", fill_type="solid")
_YELLOW = PatternFill(start_color="FFEB9C", end_color="FFEB9C", fill_type="solid")
_RED = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
_BLUE_HDR = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
_LIGHT_BLUE = PatternFill(start_color="BDD7EE", end_color="BDD7EE", fill_type="solid")
_WHITE_FONT = Font(bold=True, color="FFFFFF")
_BOLD = Font(bold=True)
_WRAP = Alignment(wrap_text=True, vertical="top")
_CENTER = Alignment(horizontal="center", vertical="center")
def _conf_fill(conf: float) -> PatternFill:
if conf >= HIGH_CONFIDENCE:
return _GREEN
if conf >= LOW_CONFIDENCE:
return _YELLOW
return _RED
def _autosize(ws, max_width: int = 80) -> None:
for col in ws.columns:
width = max(len(str(c.value or "")) for c in col)
ws.column_dimensions[get_column_letter(col[0].column)].width = min(width + 2, max_width)
def _header_row(ws, headers: List[str], row: int = 1) -> None:
for col, h in enumerate(headers, 1):
c = ws.cell(row=row, column=col, value=h)
c.fill = _BLUE_HDR
c.font = _WHITE_FONT
c.alignment = _CENTER
# ── Sheet 1: Extracted Fields ─────────────────────────────────────────────────
def _sheet_fields(wb: openpyxl.Workbook, fields: Dict) -> None:
ws = wb.active
ws.title = "Extracted Fields"
_header_row(ws, ["Field", "Extracted Value", "Confidence", "Page Ref"])
row = 2
for key, display in FIELD_DISPLAY_NAMES.items():
if key == "price_details":
continue # handled on Sheet 2
fd = fields.get(key, {})
value = fd.get("value")
conf = float(fd.get("confidence", 0.0))
page_ref = fd.get("page_ref", "")
ws.cell(row=row, column=1, value=display).font = _BOLD
vc = ws.cell(row=row, column=2, value=str(value) if value is not None else "NOT FOUND")
vc.alignment = _WRAP
vc.fill = _conf_fill(conf) if value is not None else _RED
ws.cell(row=row, column=3, value=f"{conf:.0%}").fill = _conf_fill(conf)
ws.cell(row=row, column=4, value=str(page_ref) if page_ref else "—")
row += 1
_autosize(ws)
# ── Sheet 2: Rate Card ────────────────────────────────────────────────────────
def _sheet_rate_card(wb: openpyxl.Workbook, fields: Dict) -> None:
ws = wb.create_sheet("Rate Card")
fd = fields.get("price_details", {})
conf = float(fd.get("confidence", 0.0))
page_ref = fd.get("page_ref", "")
items: Any = fd.get("value", [])
ws.cell(row=1, column=1, value="Confidence").font = _BOLD
ws.cell(row=1, column=2, value=f"{conf:.0%}").fill = _conf_fill(conf)
ws.cell(row=1, column=3, value="Page Reference").font = _BOLD
ws.cell(row=1, column=4, value=str(page_ref) if page_ref else "—")
if not isinstance(items, list) or not items:
ws.cell(row=3, column=1, value="No rate card data extracted.").font = Font(italic=True)
return
# Dynamic headers from union of all item keys
all_keys: List[str] = []
seen_keys: set = set()
for item in items:
for k in item.keys():
if k not in seen_keys:
all_keys.append(k)
seen_keys.add(k)
_header_row(ws, [k.replace("_", " ").title() for k in all_keys], row=3)
for r, item in enumerate(items, 4):
for col, k in enumerate(all_keys, 1):
ws.cell(row=r, column=col, value=str(item.get(k, ""))).alignment = _WRAP
_autosize(ws)
# ── Sheet 3: Raw OCR Text ─────────────────────────────────────────────────────
def _sheet_raw_text(wb: openpyxl.Workbook, raw_text_by_page: Dict) -> None:
ws = wb.create_sheet("Raw OCR Text")
_header_row(ws, ["Page", "Extracted Text"])
for row, (pg_num, text) in enumerate(sorted(raw_text_by_page.items()), 2):
ws.cell(row=row, column=1, value=f"Page {pg_num + 1}").font = _BOLD
ws.cell(row=row, column=2, value=text).alignment = _WRAP
ws.column_dimensions["A"].width = 10
ws.column_dimensions["B"].width = 120
# ── Node ─────────────────────────────────────────────────────────────────────
def excel_generation_node(state: ContractState) -> dict:
log = list(state.get("processing_log", []))
fields = state.get("extracted_fields", {})
raw_text = state.get("raw_text_by_page", {})
wb = openpyxl.Workbook()
_sheet_fields(wb, fields)
_sheet_rate_card(wb, fields)
_sheet_raw_text(wb, raw_text)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
stem = Path(state.get("original_filename", "contract")).stem
filename = f"{stem}_extracted_{ts}.xlsx"
out_path = OUTPUT_DIR / filename
wb.save(str(out_path))
log.append(f"Excel saved: {filename}")
return {
**state,
"excel_output_path": str(out_path),
"processing_log": log,
"current_step": "done",
}
Der Dateinamen enthält ein Zeitstempel zur Nachverfolgbarkeit.
contract_name_extracted_20260523_165458.xlsx
Schritt 10: Erstellung des Graphen
Der LangGraph-Ablauf verbindet alle Schritte miteinander – von der Vorbereitung und OCR-Verarbeitung über die Suche und das aus KI gestützte Extrahieren von Feldern bis hin zur Erstellung in Excel.
#graph.py
from langgraph.graph import StateGraph, END
from models.state import ContractState
from nodes.preprocessing import preprocess_node
from nodes.ocr_extraction import ocr_extraction_node
from nodes.indexing import indexing_node
from nodes.extraction_agent import extraction_agent_node
from nodes.excel_generation import excel_generation_node
def build_graph():
g = StateGraph(ContractState)
g.add_node("preprocess", preprocess_node)
g.add_node("ocr_extract", ocr_extraction_node)
g.add_node("index", indexing_node)
g.add_node("extract", extraction_agent_node)
g.add_node("generate_excel", excel_generation_node)
g.set_entry_point("preprocess")
g.add_edge("preprocess", "ocr_extract")
g.add_edge("ocr_extract", "index")
g.add_edge("index", "extract")
g.add_edge("extract", "generate_excel")
g.add_edge("generate_excel", END)
return g.compile()
# Singleton – imported by app.py
contract_graph = build_graph()
def save_graph_visualization(path: str = "graph_workflow.png") -> None:
png_bytes = contract_graph.get_graph().draw_mermaid_png()
with open(path, "wb") as f:
f.write(png_bytes)
print(f"Graph saved to {path}")
if __name__ == "__main__":
save_graph_visualization()
Schritt 11: Die Streamlit-Oberfläche
Starten Sie sie lokal mit den untenstehenden Befehlen.
cd medium/contract-intelligence
streamlit run app.py
Die Ergebnisse werden in fünf Tabellen angezeigt, wobei die Tabellenseite „Prompt Log“ vollständige Transparenz für das Debuggen und die Prüfung bietet.
Für jedes extrahierte Feld listet diese Tabellenseite jede ausgeführte RAG-Anfrage auf, die Top-K-Teile mit FAISS-Rang, BM25-Rang und RRF-Score, den gesamten Prompt, der an GPT-4o gesendet wurde, die rohe Modellantwort sowie die Tokenzahlen für Eingabe und Ausgabe, damit Kosten nachverfolgt werden können. Für price_details werden außerdem jeder Aufruf pro Seite sowie der Normalisierungsaufruf aufgeführt. In Unternehmensumgebungen ist diese Art von Nachvollziehbarkeit von großer Bedeutung.
Folgen einige Screenshote der Benutzeroberfläche, beginnend mit der Tabellenseite „Extracted Fields“.
Eine animierte Darstellung der in Gebrauch befindlichen Benutzeroberfläche:
Schritt 12: ein Beispiellauf
Das Testdokument ist ein öffentlich zugänglicher Staatsauftragsvertrag – eine gescannte Datei aus der Praxis mit 10 Seiten und hoher Komplexität, einschließlich Tabellen sowie handschriftlichen Zeichen.
Tab 1: extrahierte Felder
Zu den extrahierten Feldern gehören die offiziellen Namen von Partei A und Partei B, Startdatum, Enddatum, Zahlungszeitplan sowie Zahlungsbedingungen. Auch die Preise oder Tarifliste werden extrahiert, doch da ihr Schema sehr dynamisch ist, wird sie getrennt angezeigt. Jeder Wert ist mit einem Quellausschnitt versehen, was bei der Prüfung und Steuerung hilft; hier wird ebenfalls die grüne, gelbe und rote Zuverlässigkeitsfarbgebung verwendet.
Tab 2: Tarifliste
Dynamische Spalten werden nach Kanonisierung aus jeder Preisseite zusammengestellt; die Tabelle bleibt leer, wenn der Vertrag keine Tarifliste enthält.
Tab 3: Roh-OCR-Text
Die vollständige Seite für Seite angezeigte OCR-Ausgabe, die zur Überprüfung der Extraktionen und zur Aufspürung fehlender Felder nützlich ist.
Tab 4: Prompt-Logs
Jeder Aufruf der LLM-API wird zusammen mit seinen wichtigsten Ausgabeabschnitten, dem genauen Eingabeprompt sowie der Ausgabe protokolliert, um Überwachung und Bewertung zu ermöglichen.
Ein grober Überblick über die Gesamtanzahl der Token, die bei allen Aufrufen der KI-Modelle für einen Vertrag verwendet werden: etwa 17.000 Eingabetoken und etwa 4.000 Ausgabetoken. Das entspricht bei gpt-4o-mini etwa 0,005 US-Dollar und bei gpt-4o etwa 0,10 US-Dollar. Selbst zum höheren Preis ist diese Methode günstiger, schneller und strukturierter als manuelle Überprüfungen.
Architekturübersicht
Der vollständige End-to-End-Ablauf des Systems:
Entwurfsentscheidungen
Die hinter dem System stehenden Entscheidungen balancieren Genauigkeit der Extraktion, Verarbeitungsgeschwindigkeit, Skalierbarkeit und Kosten für die Analyse realer Verträge.
Mögliche weitere Entwicklungen
Das aktuelle System ist bereits für die Extraktion aus einzelnen Dokumenten geeignet. Die nächsten Schritte zielen auf mehr Skalierbarkeit, höhere Genauigkeit sowie Analyse auf Portfolioebene ab:
- Mehr Extraktionsfelder, zusätzliche Überprüfungen sowie parallele Batch-Verarbeitung für große Vertragsportfolios.
Durch die Kombination von OCR, Embedding-Modellen und agiler Extraktion entsteht ein effizientes Vertragsintelligenzsystem. Die Dokumentanalyse, semantische Suche und strukturierte Feldextraktion in einem modularen Workflow ermöglichen es, komplexe Verträge mit geringem manuellem Aufwand zu verarbeiten und strukturierte Ausgaben zu erzeugen, die für weitere Analysen bereit sind. Die Architektur ist skalierbar und erweiterbar und bildet eine solide Grundlage für die Automatisierung der Vertragsanalyse sowie zur Reduzierung der Zeit für manuelle Überprüfungen.