Інтелект контрактів із OCR, гібридним RAG та LangGraph
Потік обробки контрактів: очищення зображень, PaddleOCR із альтернативою GPT-4o, пошук за допомогою FAISS та BM25, багатозапитний RAG та звіт у форматі Excel із оцінками.
Проблеми з контрактами
Контракти є основою кожної організації — від угод з постачальниками та замовлень на послуги до державних тендерів. У середнього за розміром підприємства одночасно може бути сотні таких контрактів, а у великій корпорації — десятки тисяч. Проте комерційні дані, які вони містять, залишаються у неструктурованих файлах: відскановані PDF-файли, сфотографовані сторінки, експорт дані у форматі XML з інструментів закупівель. Кожен, хто хоче дізнатися умови оплати від постачальника чи термін закінчення контракту, змушений відкривати документ, читати сторінки юридичних формулювань та вводити інформацію у електронну таблицю.
Через таку рутину виникають три проблеми:
- Терміни та процедури продовження контрактів пропускаються, оскільки важлива клозула ховається глибоко в тексті.
- Несуворість у ціноутворенні постачальників залишається непоміченою у масштабних операціях, що призводить до втрат грошей.
- Ручний відстежування умов оплати створює ризики для дотримання правил.
Основна проблема полягає у тому, що дані контрактів зберігаються у документах, а не в базах даних. Завдання, яке можна було б вирішити одним SQL-запитом, займає у людини близько 20 хвилин.
Ця система автоматизує весь процес перегляду від початку до кінця. Контракт подається у форматі PDF, скану чи XML, і через кілька секунд у звіті Excel вказуються ключові комерційні умови, кожна з яких супроводжується показником надійності та посиланням на сторінку. PaddleOCR, FAISS, BM25, Reciprocal Rank Fusion та GPT-4o працюють усі разом у машинному стані LangGraph.
Різні команди отримують переваги по-різному: відділ закупівель швидше переглядає контракти та швидше інтегрує постачальників; фінансовий відділ виявляє розбіжності у цінах між контрактами з різними постачальниками; юридичний та відділ комплаєнсу зберігають записи, які можна перевірити, з посиланнями на джерело; відділ операцій відстежує продовження термінів дії контрактів без додаткових зусиль.
У проекті використано такі техніки:
- Багатоетапна структура LangGraph DAG для обробки документів.
- Покращення якості зображень (удалення шуму, CLAHE, виправлення нахилу) перед OCR для підвищення точності.
- Система OCR з подвійною стратегією: PaddleOCR з можливістю переходу на GPT-4V у разі проблем.
- Гібридний підхід RAG, де FAISS, BM25 та метод Reciprocal Rank Fusion разом дають кращі результати, ніж будь-який з цих інструментів окремо.
- Запити RAG для кількох запитань, які максимізують кількість знайдених даних у різних формулюваннях контрактів.
- Витягування інформації з багатосторінкових документів із канонізацією схеми за допомогою LLM.
- Результати з оцінкою надійності та можливістю аудиту, придатні для використання в корпоративних проектах.
Готове застосування
Готовий додаток приймає контракти у форматах PDF, зображень (JPG, PNG, TIFF) та XML. Зображення сторінок очищуються шляхом усунення шуму, корекції контрасту та вирівнювання, після чого PaddleOCR їх аналізує, а для рідкісних сторінок використовується GPT-4o vision fallback. Пошук здійснюється за допомогою гібридного індексу FAISS та BM25, об’єднаного методом Reciprocal Rank Fusion, а multi-query RAG із оцінкою надійності витягує сім полів контракту. Результати зберігаються у звіті Excel у кольоровому форматі з витягнутими полями, таблицею тарифів та необробленим текстом OCR; крім того, Streamlit UI відображає їх разом із детальним журналом запитів для повної прозорості.
Увесь процес у одному перегляді:
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
└─────────────────────┘
Структура проекту
Репозиторій організований наступним чином.
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
Крок 0: налаштування середовища
Почніть із віртуального середовища, щоб ізолювати залежності. У macOS або Linux:
python -m venv venv
source venv/bin/activate
У Windows:
python -m venv venv
venv\Scripts\activate
Далі встановіть бібліотеки, які покривають OCR, пошук інформації, ембеддинги, оркестрацію LLM та інтерфейс Streamlit.
pip install -r requirements.txt
Ключові пакети у requirements.txt виглядають так.
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
Файл .env містить облікові дані OpenAI та вибір моделі.
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini # or gpt-4o for higher accuracy
Крок 1: стан конвеєра та центральна конфігурація
LangGraph передає всю інформацію через один спільний об’єкт стану, тому варто його визначити ще до створення будь-якого вузла. Він є єдиним джерелом істини для конвеєра.
# 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 забезпечує підказки щодо типів та автодоповнення в IDE, водночас зберігаючи можливість серіалізації стану. З параметром total=False кожне поле є необов’язковим, оскільки кожен вузол записує лише те, що він генерує.
Усі налаштовувані параметри знаходяться в одному файлі конфігурації.
# 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
Крок 2: Сервіси LLM та ембеддингів
Модуль LLM інтегрує API OpenAI для обробки тексту, зображень та створення структурованих даних у форматі JSON. Він забезпечує формування запитів, кодування зображень, повторні спроби виконання завдань, відстеження кількості використаних токенів та стійке парсування JSON, що робить процеси вилучення даних надійними.
# 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 {}
Сервіс імбеддингу використовує Sentence Transformers для створення щільних векторів з тексту контракту. Імбеддинги нормалізуються, щоб пошук за семантичною схожістю залишався ефективним, і вони використовуються як для індексування, так і для пошуку інформації.
# 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]
Крок 3: попередня обробка
Сканувані оригінали договорів зазвичай мають високий рівень шуму, низьку контрастність та трохи спотворену форму. Перед OCR кожне зображення сторінки проходить через три етапи покращення якості.
Покращення якості зображення
# 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)
Етапи виконуються по черзі.
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
Чому саме CLAHE, а не звичайна еквіалізація гістограми? Стандартна еквіалізація змінює контраст у всьому зображенні, що може надмірно освітлити вже світлі ділянки. CLAHE, скорочення від Contrast Limited Adaptive Histogram Equalization, працює з невеликими фрагментами зображення. Текст на вицвілому штампі чи світлому заголовку покращується без зниження якості чистих ділянок.
Видалення дублікатів сторінок
Довгі контракти часто містять повторювані сторінки – дублікати титульних аркушів, порожні розділовики та копії на папері. Операція OCR з усіма ними займає багато часу та заповнює індекс дубльованим текстом.
# 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
Перцептивне хешування (phash) перетворює зображення кожної сторінки на 64-бітний відбиток, створений на основі частотних складових. Сторінки, відстань Хеммінга яких від попередньої сторінки дорівнює 6 або менше, вважаються візуально ідентичними, тож фіксуються як точні копії, так і майже ідентичні копії, наприклад сторінка, відсканована двічі з різною якістю. Приклад виконання:
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 ✓
Також додано універсальний парсер XML разом із функцією, яка послідовно викликає всі кроки попередньої обробки.
# 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",
}
Крок 4: OCR з подвійною стратегією
OCR є найбільш ризикованим кроком, оскільки пропущена таблиця чи поле форми може завадити подальшій обробці даних. Для покращення охоплення використовуються два моделі.
# 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)
Кожна сторінка дотримується цього дерева рішень.
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)
Чому обмеження становить 80 символів? На простих сторінках є таблиці з небагатьма словами в кожній клітинці, переважно порожні форми, блоки для штампів та підписів. PaddleOCR добре справляється з щільним текстом, але не ефективно працює з сильно структурованими макетами, тоді як GPT-4o Vision може розгледіти всю зображення та відновити таблицю за допомогою роздільників у вигляді трубок.
Запит до моделі для розпізнавання зображень є навмисно конкретним.
#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])
Потім ці елементи об’єднуються у повний 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",
}
Крок 5: створення гібридного індексу
Після того, як на кожній сторінці буде чистий текст, наступним кроком є створення пошукової бази знань за допомогою стратегії розділення на батьківські та дочірні частини.
Розділення на частини
# 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",
}
Основна ідея — пошук між батьком та дитиною.
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)
Малі фрагменти точно відповідають запитам, але передача LLM уривка довжиною 600 символів часто призводить до втрати навколишнього контексту, необхідного для його розуміння. Зберігання сторінок та подальше їх використання забезпечує як точність, так і контекст.
Гібридний сховище
HybridVectorStore є основою процесу пошуку. Він поєднує два принципово різних методи пошуку.
# 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)
Використання обох підходів є вигідним, адже кожен з них компенсує слабкі сторони іншого.
Ось як працює метод Reciprocal Rank Fusion (RRF). Формула 1/(k + rank) перетворює довільні ранги на оцінки, які можна порівнювати. При k=60 результат першого місця від будь-якої з систем додає 1/61, що становить приблизно 0,016. Елемент, який посідає перше місце за допомогою FAISS та третє — за допомогою BM25, дає суму 1/61 + 1/63, що становить приблизно 0,032, що краще за елемент, який знайшла лише одна з систем. Це корисно для контрактів із незвичайною термінологією, де BM25 має труднощі, а також для шаблонного тексту, скопійованого з інших джерел, де FAISS створює занадто багато результатів.
Крок 6: агент вилучення
Саме тут знаходиться інтелектуальний елемент. Для кожного з семи полів контракту агент вилучення формулює кілька цілеспрямованих запитів, отримує найбільш релевантні сторінки та просить GPT-4o витягти потрібну інформацію.
Сім налаштовуваних полів
# 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)
]
Отримання даних за кількома запитами
Кожному полю надається чотири варіанти запиту для максимізації кількості знайдених результатів. Один запит щодо „дати початку“ може пропустити контракт, у якому зазначено „Початок:“ або „Набуває чинності з:“; тоді як кілька запитів дозволяють виявити різні варіанти формулювань.
# 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",
],
}
Процес отримання даних нижче повторюється для кожного поля.
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
Використання сторінки номер нуль для назв партій
У державних контрактах назви партій часто зустрічаються лише на титульному аркуші чи обкладинці, а не в основних положеннях договору. Якщо певний поріг оцінки призводить до того, що обкладинка відсувається на другий план через більш високий пріоритет внутрішніх положень, назва партії зникає. Для таких полів існує особливий випадок, який дозволяє залишити у використанні сторінку номер нуль.
# 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]
Розширення до батьківського контексту з фільтрацією
Перш ніж контекст потрапить до LLM, дочірні фрагменти розширюються до повних батьківських сторінок. Сторінки з низькими оцінками видаляються, щоб запит залишався зосередженим.
# 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)
Інструкції для екстракції
У кожному полі є ретельно складена інструкція, яка пояснює GPT-4o, що шукати та як відповідати.
# 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": ""}}"""
Наприклад, запит payment_timeline має дуже вузьку спрямованість.
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.
Він відрізняється від payment_conditions, який описує всю ситуацію повністю.
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.
Тримання їх окремо запобігає тому, що ШІ-модель сплутає 400-слівний пункт умов оплати з простим формулюванням „Net 30“.
Карти тарифів на кількох сторінках
Карти тарифів є найскладнішим форматом. Вони охоплюють кілька сторінок, назви стовпців змінюються зі сторінки на сторінку („Номер товару“ на одній, „SKU“ — на іншій), а кількість рядків сягає десятків. Одне запитання до ШІ-моделі для всіх сторінок призведе до переповнення контексту та заплутаності в схемах стовпців. Рішенням є видобування даних для кожної сторінки окремо, а потім нормалізація схем.
# 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)
Отриманий результат мапування виглядає приблизно так.
{
"Item Number": "SKU",
"Item Numbers": "SKU",
"SKU": "SKU",
"Price": "Unit Cost",
"Unit Cost": "Unit Cost",
"Amount": "Unit Cost"
}
Кожен рядок з кожної сторінки потім переформатовується за допомогою цього мапування та об’єднується в одну єдину таблицю.
Кешування OCR
Обробка 50-сторінкового контракту за допомогою OCR займає від 5 до 10 хвилин, і чекати так довго під час кожної повторної обробки є незручним, коли налаштовуються запити чи виправляються помилки. Кеш виявляє попередній Excel-звіт для того самого документа та пропонує перейти безпосередньо до етапу вилучення інформації.
# 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
З увімкненим кешуванням процес виконується таким чином.
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
Крок 9: робочий аркуш Excel
Кінцеві результати та первинні журнали зберігаються у файлі Excel, який також може використовуватися для бази даних. Колір відображає рівень надійності: зелений — для 80% та більше (високий), жовтий — для 50–79% (середній), а червоний — для менше 50% (низький або не знайдено). Ці показники вказують, які атрибути потребують перевірки людиною.
#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",
}
У назві файлу вказано час створення для можливості відстеження.
contract_name_extracted_20260523_165458.xlsx
Крок 10: створення графа
Робочий процес LangGraph об’єднує всі етапи — від попередньої обробки та OCR через пошук інформації та вилучення даних за допомогою ШІ до створення файлу 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()
Крок 11: інтерфейс Streamlit
Запустіть його локально за допомогою наведених нижче команд.
cd medium/contract-intelligence
streamlit run app.py
Результати відображаються у п’яти вкладках, причому вкладка Prompt Log забезпечує повну прозорість для дебаггінгу та аудиту.
Для кожного вилученого поля вкладка перелічує кожне запитання RAG, яке було виконано, найкращі K фрагментів із рейтингами FAISS, BM25 та RRF, повний запит, надісланий до GPT-4o, первинну відповідь моделі, а також кількість токенів у вхідному та вихідному тексті для відстеження витрат. Для price_details додано інформацію про кожен запит на сторінку та про операцію канонізації. У корпоративних налаштуваннях така можливість аудиту є дуже важливою.
Далі наведено кілька скріншотів інтерфейсу, починаючи з вкладки Вилучені поля.
Анімоване зображення інтерфейсу під час роботи:
Крок 12: зразковий запуск
Тестовим документом є публічно доступний договір про державні закупівлі — сканований файл реального документа обсягом 10 сторінок із великою складністю, що включає таблиці та рукописні символи.
Вкладка 1: вилучені поля
Вилученими полями є офіційні назви Сторони А та Сторони Б, дата початку, дата закінчення, графік оплати та умови оплати. Також вилучається інформація про ціни чи тарифну таблицю, але її структура настільки динамічна, що вона відображається окремо. Кожне значення супроводжується фрагментом джерела, що допомагає під час аудиту та керування, а тут також застосовується зелений, жовтий та червоний колір для позначення рівня надійності.
Вкладка 2: тарифна таблиця
Динамічні стовпці формуються з кожної сторінки з цінами після канонізації, і вкладка залишається порожньою, якщо у контракті немає тарифної таблиці.
Вкладка 3: необроблений текст OCR
Повний результат OCR для кожної сторінки, що корисний для перевірки вилученої інформації та пошуку пропущених полів.
Вкладка 4: журнали запитів
Кожен виклик API ШІ фіксується разом із його найважливішими фрагментами, точним вхідним запитом та результатом для моніторингу та оцінки.
Орієнтовна кількість загальних токенів, використаних під час усіх запитів до ШІ для одного контракту: близько 17 тис. вхідних токенів та близько 4 тис. вихідних токенів. Це коштує приблизно 0,005 долара з використанням gpt-4o-mini та близько 0,10 долара з використанням gpt-4o. Навіть за вищою ціною це дешевше, швидше та більш структуровано, ніж ручна перевірка.
Огляд архітектури
Повний кінцево-кінцевий процес роботи системи:
Рішення щодо проектування
Рішення, які лежать в основі системи, забезпечують баланс між точністю вилучення інформації, швидкістю обробки, масштабованістю та витратами для аналізу контрактів у реальному світі.
Можливий подальший розвиток
Нинішня система готова до вилучення інформації з одного документа. Наступні кроки спрямовані на збільшення масштабу, покращення точності та створення аналітики на рівні всього портфеля:
- Більше полів для вилучення інформації, додаткові перевірки та паралельна обробка пакетів документів для великих портфелів контрактів.
Поєднання OCR, моделей ембеддингів та агентного вилучення інформації створює ефективну систему інтелекту контрактів. Парсинг документів, семантичне пошукове забезпечення та вилучення структурованих даних у одному модульному робочому процесі дозволяють обробляти складні контракти з мінімальними зусиллями та отримувати структурований результат, готовий до подальшого аналізу. Архітектура є масштабованою та розширюваною, що є міцною основою для автоматизації аналізу контрактів та скорочення часу ручної перевірки.