Inteligencja kontraktowa z OCR, hybrydowym RAG i LangGraph
Proces wydobywania danych z umów: oczyszczanie obrazu, PaddleOCR z alternatywą GPT-4o, wyszukiwanie za pomocą FAISS i BM25, RAG z wieloma zapytaniami oraz raport w formacie Excel z ocenami.
Problemy z umowami
Umowy stanowią podstawę każdej organizacji – od umów z dostawcami i zamówień usługowych po przetargi rządowe. W firmie średniej może być aktywnych setki takich umów, a w dużym przedsiębiorstwie dziesiątki tysięcy. Mimo to informacje handlowe zawarte w nich pozostają zamknięte w nieuporządkowanych plikach: zeskanowanych PDF-ach, sfotografowanych stronach, eksportach w formacie XML z narzędzi do zakupów. Każdy, kto chce poznać warunki płatności u dostawcy lub datę wygaśnięcia umowy, musi otworzyć dokument, przeczytać strony pełne standardowych formuł prawnych i wpisać odpowiedź do arkusza kalkulacyjnego.
Z tej rutyny wynikają trzy problemy:
- Koniec terminów i procedur odnowienia umowy umyka uwadze, ponieważ istotne klauzule są ukryte głęboko w tekście.
- Niejednolite ceny u dostawców pozostają niezauważone na większą skalę, co powoduje straty finansowe.
- Ręczne śledzenie warunków płatności stwarza ryzyko nieprzestrzegania przepisów.
Głównym problemem jest to, że dane kontraktowe znajdują się w dokumentach, a nie w bazach danych. Pytanie, które wymagałoby jednego zapytania SQL, zajmuje osobie około 20 minut.
Tutaj system automatyzuje cały proces przeglądania od początku do końca. Kontrakt jest wprowadzany w formie pliku PDF, skanu lub XML, a sekundy później raport w formacie Excel przedstawia kluczowe warunki handlowe, każdy z oceną pewności oraz odniesieniem do strony. PaddleOCR, FAISS, BM25, Reciprocal Rank Fusion i GPT-4o działają wszystkie w ramach jednej maszyny stanowej LangGraph.
Różne zespoły czerpią korzyści na różne sposoby. Dział zakupów szybciej przegląda kontrakty i szybciej włącza dostawców do współpracy. Dział finansowy wykrywa rozbieżności w cenach pomiędzy kontraktami z różnych dostawców. Dział prawny i ds. zgodności przechowuje dokumenty nadające się do audytu wraz z odniesieniami do źródeł. Dział operacyjny śledzi odnowienia i terminy bez konieczności ręcznej pracy.
Budowa obejmuje następujące techniki:
- Multystopniowy graf DAG LangGraph do przetwarzania dokumentów.
- Poprawa jakości obrazu (usuwanie szumów, CLAHE, korygowanie ukośnienia) przed OCR w celu zwiększenia dokładności.
- Układ OCR oparty na dwóch strategiach: PaddleOCR z opcją awaryjną GPT-4V.
- Hibrydowy system RAG, w którym FAISS, BM25 i Reciprocal Rank Fusion łącznie przewyższają efektywność każdego z tych mechanizmów używanego pojedynczo.
- Wskazówki RAG dla wielu zapytań, które maksymalizują stopień odzyskiwania informacji w różnych sformułowaniach umów.
- Ekstrakcja danych z wielu stron kart tarifowych przy użyciu kanonizacji schematu sterowanej przez LLM.
- Wyniki z oceną pewności i możliwością audytu, odpowiednie do użycia w przedsiębiorstwach.
Zakończone rozwiązanie
Zakończona aplikacja obsługuje kontrakty w formacie PDF, obrazów (JPG, PNG, TIFF) oraz XML. Obrazy stron są oczyszczane poprzez usuwanie szumów, dostosowanie kontrastu i korygowanie ukośnienia, a następnie PaddleOCR je odczytuje, przy czym w przypadku rzadko zapisanych stron używany jest GPT-4o vision fallback. Wyszukiwanie odbywa się za pomocą hybrydowego indeksu FAISS i BM25 połączonego metodą Reciprocal Rank Fusion, a multi-query RAG z oceną pewności wyodrębnia siedem pól kontraktu. Wyniki są prezentowane w raporcie Excel w kolorach zawierającym wyodrębnione pola, tabelę stawek oraz surowy tekst odczytany przez OCR, natomiast interfejs Streamlit pokazuje je obok szczegółowego logu zapytań dla pełnej przejrzystości.
Cały proces w jednym widoku:
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
└─────────────────────┘
Struktura projektu
Repozytorium jest zorganizowane w następujący sposób.
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
Krok 0: konfiguracja środowiska
Zacznij od środowiska wirtualnego, aby izolować zależności. W systemach macOS lub Linux:
python -m venv venv
source venv/bin/activate
W systemie Windows:
python -m venv venv
venv\Scripts\activate
Następnie zainstaluj biblioteki obejmujące OCR, wyszukiwanie, embeddingi, orkiestrację LLM oraz interfejs Streamlit.
pip install -r requirements.txt
Kluczowe pakiety w requirements.txt wyglądają w ten sposób.
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
Plik .env zawiera dane logowania do OpenAI oraz wybór modelu.
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini # or gpt-4o for higher accuracy
Krok 1: stan pipeline i konfiguracja centralna
LangGraph przekazuje wszystko przez jeden wspólny obiekt stanu, dlatego warto go zdefiniować przed utworzeniem jakiegokolwiek węzła. Pełni on rolę jedynego źródła prawdy dla całego 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 zapewnia wskazówki typowe oraz automatyczne uzupełnianie w IDE, jednocześnie utrzymując możliwość zapisu stanu w formie serializowalnej. Przy użyciu total=False każde pole jest opcjonalne, ponieważ każdy węzeł zapisuje tylko to, co sam wytwarza.
Wszystkie dostępne do dostosowania parametry znajdują się w jednym pliku konfiguracyjnym.
# 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
Krok 2: Usługi LLM i embeddingów
Moduł LLM obejmuje API OpenAI do generowania tekstu, obrazów oraz strukturyzowanych danych w formacie JSON. Zajmuje się tworzeniem zapytań, kodowaniem obrazów, próbami ponownymi, śledzeniem użycia tokenów oraz odpornym parsowaniem JSON, co zapewnia niezawodność procesów wydobywania informacji.
# 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 {}
Służba embeddingów wykorzystuje Sentence Transformers do tworzenia zwartych wektorów na podstawie tekstu umowy. Embeddingi są normalizowane, aby wyszukiwanie podobieństwa semantycznego pozostawało efektywne, a one służą zarówno do indeksowania, jak i odzyskiwania informacji.
# 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]
Krok 3: przetwarzanie wstępne
Skanowane oryginalne umowy zazwyczaj są szumne, mają niski kontrast i są lekko zniekształcone. Przed procesem OCR każdy obraz strony przechodzi przez trzy etapy poprawy jakości.
Poprawa jakości obrazu
# 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)
Etapy działają kolejno.
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
Dlaczego CLAHE, a nie zwykła ekwalizacja histogramu? Standardowa ekwalizacja zmienia kontrast we wszystkich obszarach i może nadmiernie rozjaśnić te, które już były jasne. CLAHE, skrót od Contrast Limited Adaptive Histogram Equalization, działa na małych obszarach. Tekst na wyblakłym znaczku lub bladym nagłówku zostaje wyostrzony bez utraty jakości czystych obszarów.
Usuwanie duplikatów stron
Długie umowy często zawierają powtarzające się strony, takie jak duplikowane karty tytułowe, puste separatory i kopie. Odczyt OCR wszystkich tych elementów zajmuje dużo czasu i napełnia indeks duplikowanym tekstem.
# 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
Hashowanie percepcyjne (phash) sprowadza obraz każdej strony do 64-bitowego odcisku palca utworzonego na podstawie składowych częstotliwości. Strony, których odległość Hamminga od wcześniejszej strony wynosi 6 lub mniej, są traktowane jako wizualnie identyczne, dzięki czemu zarówno dokładne kopie, jak i niemal identyczne kopie – np. strona zeskanowana dwa razy przy różnej jakości – są wykrywane. Przykład działania:
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 ✓
Dodano również ogólny parser XML wraz z końcową funkcją, która wywołuje wszystkie kroki przetwarzania wstępnego kolejno.
# 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",
}
Krok 4: OCR z podwójną strategią
OCR to najbardziej ryzykowny krok, ponieważ pominięcie tabeli lub pola formularza może zakłócić późniejszą ekstrakcję danych. Węzeł wykorzystuje dwa modele, aby zwiększyć zakres obsługi.
# 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)
Każda strona postępuje zgodnie z tym drzewem decyzyjnym.
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)
Dlaczego granicą jest 80 znaków? Strony o małej objętości zawierają tabele z niewielką liczbą słów w każdej komórce, głównie puste formularze oraz pola na pieczątki i podpisy. PaddleOCR dobrze radzi sobie z gęstym tekstem, ale zawodzi w przypadku silnie skonstruowanych układów, podczas gdy GPT-4o Vision potrafi obejrzeć całe zdjęcie i odtworzyć tabelę przy użyciu separatorów w postaci kropek.
Zapytanie do systemu rozpoznawania obrazów jest celowo konkretne.
#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])
Następnie elementy te są łączone w kompletny pipeline OCR.
#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",
}
Krok 5: tworzenie hybrydowego indeksu
Gdy tekst na każdej stronie jest już oczyszczony, następuje stworzenie bazy wiedzy nadającej się do wyszukiwania, przy użyciu strategii dzielenia na elementy rodzic-dziecko.
Dzielenie na elementy
# 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",
}
Główną ideą jest odzyskiwanie danych między rodzicem a dzieckiem.
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)
Małe fragmenty pasują do zapytań z dużą dokładnością, ale przekazanie modelowi LLM fragmentu o długości 600 znaków często powoduje utratę otaczającego kontekstu, który jest niezbędny do zrozumienia tego fragmentu. Przechowywanie stron, a następnie ich późniejsze wykorzystanie, zapewnia zarówno dokładność, jak i kontekst.
Hybrydowe przechowywanie
HybridVectorStore stanowi istotę mechanizmu wyszukiwania. Łączy on dwa fundamentalnie różne metody wyszukiwania.
# 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)
Wykorzystanie obu metod przynosi korzyści, ponieważ każda z nich kompensuje słabe strony drugiej.
Oto jak działa Reciprocal Rank Fusion (RRF). Wzór 1/(k + rank) przekształca dowolne rangi na wartości porównywalne. Przy k=60 wynik pierwszej pozycji uzyskany za pomocą dowolnego z systemów daje 1/61, czyli około 0,016. Fragment, który zajmuje pierwsze miejsce w FAISS i trzecie w BM25, daje łącznie 1/61 + 1/63, czyli około 0,032, co przewyższa wynik fragmentu znalezionego tylko przez jeden system. To rozwiązanie sprawdza się w przypadku umów z nietypową terminologią, gdzie BM25 ma trudności, oraz w przypadku skopiowanych szablonów, gdzie FAISS tworzy zbyt wiele dopasowań.
Krok 6: agent wydobywający
Tutaj znajduje się element inteligencji. Dla każdego z siedmiu pól umowy agent wydobywający wysyła kilka celowanych zapytań, pobiera najbardziej istotne strony i prosi GPT-4o o wydobycie odpowiedniej wartości.
Siedem konfigurowalnych pól
# 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)
]
Pobieranie danych przy użyciu wielu zapytań
Każde pole otrzymuje cztery warianty zapytań, aby maksymalizować dokładność odnalezienia. Jedno zapytanie dotyczące „daty rozpoczęcia” może przeoczyć umowę, w której podana jest informacja „Rozpoczęcie:” lub „Obowiązuje od:”, natomiast kilka różnych kierunków zapytań pozwala na wykrycie tych wariantów.
# 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",
],
}
Poniższy proces pobierania jest powtarzany dla każdego pola.
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
Używanie strony zero dla nazw partii
W umowach rządowych nazwy partii pojawiają się często jedynie na kopercie lub stronie tytułowej, a nie w treści umowy. Jeśli określony próg oceny sprawia, że strona tytułowa zostaje pominięta ze względu na wyższy priorytet klauzul wewnętrznych, nazwa partii znika. Istnieje specjalny przypadek, który umożliwia użycie strony zero dla tych pól.
# 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]
Rozszerzanie do kontekstu nadrzędnego z filtrowaniem
Zanim kontekst trafi do modelu LLM, fragmenty potomne są rozszerzane na całe strony nadrzędne. Strony o niskiej ocenie są pomijane, aby prompt pozostał skoncentrowany.
# 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)
Wskazówki do ekstrakcji
Każde pole zawiera starannie sformułowaną instrukcję, która mówi GPT-4o, czego ma szukać i jak odpowiadać.
# 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": ""}}"""
Np. pole wprowadzania payment_timeline ma bardzo wąski zakres.
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.
Różni się od payment_conditions, który zawiera całą historię.
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.
Trzymanie ich oddzielnie zapobiega temu, że model językowy pomyli 400-słowną klauzulę płatności z prostym określeniem „Net 30”.
Karty stawek na różnych stronach
Karty stawek to najtrudniejszy obszar do obsługi. Zajmują kilka stron, nazwy kolumn różnią się na poszczególnych stronach („Numer pozycji” na jednej, „SKU” na innej), a liczba wierszy sięga kilkudziesięciu. Jedna tylko próba przetworzenia wszystkich stron przez model językowy spowodowałaby przepełnienie kontekstu i zamieszanie w schematach kolumn. Rozwiązaniem jest wyodrębnianie danych strona po stronie, a następnie kanonizacja tych schematów.
# 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)
Mapowanie, które wraca, wygląda mniej więcej w ten sposób.
{
"Item Number": "SKU",
"Item Numbers": "SKU",
"SKU": "SKU",
"Price": "Unit Cost",
"Unit Cost": "Unit Cost",
"Amount": "Unit Cost"
}
Każdy wiersz z każdej strony jest następnie przekształcany za pomocą tej mapy i łączony w jedną spójną tabelę.
Caching OCR
Przetworzenie 50-stronicowej umowy za pomocą OCR zajmuje od 5 do 10 minut, a czekanie tak długo przy każdym ponownym uruchomieniu jest uciążliwe podczas dostosowywania poleceń lub naprawiania błędów. Cache wykrywa wcześniejszy raport z Excela dla tego samego dokumentu i proponuje przejście bezpośrednio do etapu wydobywania danych.
# 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
Gdy katalogowanie jest włączone, proces przebiega w ten sposób.
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
Krok 9: plik Excel
Ostateczne wyniki oraz surowe zapisy są przechowywane w pliku Excel, który może być również użyty do załadunku danych do bazy. Kolory wskazują poziom pewności: zielony dla 80% i więcej (wysoki), żółty dla 50–79% (średni), a czerwony poniżej 50% (niski lub nieznaleziony). Te wartości pokazują, które atrybuty wymagają ponownej weryfikacji przez człowieka.
#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",
}
Nazwa pliku zawiera datę i godzinę w celu śledzenia.
contract_name_extracted_20260523_165458.xlsx
Krok 10: budowa grafu
Proces LangGraph łączy wszystkie etapy – od przetwarzania wstępnego i OCR, przez wyszukiwanie i wyodrębnianie danych przy użyciu sztucznej inteligencji, aż po generowanie plików 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()
Krok 11: interfejs Streamlit
Zainstaluj go lokalnie za pomocą poniższych poleceń.
cd medium/contract-intelligence
streamlit run app.py
Wyniki są pokazywane w pięciu kartach, a karta „Log zapytań” zapewnia pełną przejrzystość przy debugowaniu i audycji.
Dla każdego wydobytego pola karta wymienia wszystkie wykonane zapytania RAG, najważniejsze fragmenty z rankiem FAISS, rankiem BM25 oraz wynikiem RRF, całe zapytanie wysłane do GPT-4o, surową odpowiedź modelu oraz liczbę tokenów wejściowych i wyjściowych, aby można było śledzić koszty. W przypadku price_details dodawane są informacje o każdym wywołaniu na stronę oraz o operacji kanonizacji. W ustawieniach korporacyjnych taka możliwość audytu ma duże znaczenie.
Poniżej znajdują się kilka zrzutów ekranu z interfejsu, począwszy od karty „Wydobyte pola”.
Animowana prezentacja interfejsu w użyciu:
Krok 12: przykładowe uruchomienie
Dokument testowy to publicznie dostępna umowa zakupu usług państwowych – zeskanowany plik z 10 stron o dużej złożoności, zawierający tabele oraz ręcznie napisane znaki.
Karta 1: wydobyte pola
Pola, które zostały wydobyte, to oficjalne nazwy Strony A i Strony B, data rozpoczęcia, data zakończenia, harmonogram płatności oraz warunki płatności. Wydobywany jest również cennik, ale jego struktura jest tak dynamiczna, że jest wyświetlany oddzielnie. Każda wartość ma dołączony Fragment źródłowy, który ułatwia audyt i zarządzanie, a tutaj również stosuje się kolorystykę pewności w kolorach zielonym, żółtym i czerwonym.
Karta 2: cennik
Kolumny dynamiczne są tworzone na podstawie każdej strony z cenami po kanonizacji, a karta pozostaje pusta, jeśli umowa nie zawiera cennika.
Karta 3: surowy tekst OCR
Pełny wynik OCR strona po stronie, przydatny do sprawdzania wydobytych danych i identyfikacji pominiętych pól.
Karta 4: logi zapytań
Każde wezwanie API LLM jest rejestrowane wraz z jego najważniejszymi fragmentami typu top-K, dokładnym zapytaniem wejściowym i wynikiem, w celu monitorowania i oceny.
Oryginalna ocena całkowitego użycia tokenów we wszystkich wywołaniach modeli LLM dla jednego kontraktu: około 17 tys. tokenów wejściowych i około 4 tys. tokenów wyjściowych. To daje koszt około 0,005 dolara przy użyciu gpt-4o-mini oraz około 0,10 dolara przy użyciu gpt-4o. Nawet przy wyższej cenie jest to tańsze, szybsze i bardziej uporządkowane niż ręczna weryfikacja.
Przegląd architektury
Pełny proces end-to-end systemu:
Decyzje projektowe
Decyzje przyjęte przy tworzeniu systemu uwzględniają dokładność wydobywania informacji, szybkość przetwarzania, skalowalność oraz koszt w kontekście analizy kontraktów w praktyce.
Kierunki dalszego rozwoju
Bieżący system jest gotowy do wydobywania informacji z pojedynczych dokumentów. Kolejne kroki zmierzają do zwiększenia skali, dokładności oraz dostarczania analiz na poziomie całego portfela kontraktów:
Łączenie OCR, modeli embeddingów oraz mechanizmów agencyjnych do ekstrakcji danych tworzy wydajny system analityki umów. Parsowanie dokumentów, wyszukiwanie semantyczne oraz ekstrakcja strukturyzowanych danych w ramach jednego modułowego procesu umożliwiają przetwarzanie złożonych umów przy minimalnym nakładzie pracy ręcznej oraz generowanie gotowych do dalszej analizy danych strukturyzowanych. Architektura ta jest skalowalna i rozszerzalna, stanowiąc solidną podstawę do automatyzacji analizy umów i skrócenia czasu potrzebnego na ręczną weryfikację.
Literatura pokrewna
- Praktyczne notatki: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph — Szczegółowy przewodnik po praktycznych notatkach: Budowanie samokorygującego agenta AI z Self-RAG i LangGraph: umowy, sprawdzenia oraz gotowe fragmenty kodu dla zespołów wdrażających ten model.
- Budowanie samokorygującego agenta do dokumentacji z LangGraph i Ollama — Lokalny pipeline refleksji, w którym jeden model tworzy szkice dokumentacji API w Pythonie, drugi je krytykuje, a trzeci pogodzi wyniki, połączone w pętlę LangGraph na Ollama.