Аналіз дагавораў з 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 і метод злучэння взаімных рангоў працуюць разам і даюць лепыяся рэзультаты, чым кожны з алгорытмаў окалічна.
- Запросы RAG для калькуляцыі множкі паказаў, якія максымізуюць колькасць знайдзеных дакументаў у разных формулюваннях кантрактаў.
- Экстракцыя даных з калькулятораў у кольках сторунакаў за дапамогою шаблонаў, створаных са адказамі LLM.
- Рэзультаты з паказаннем рэгламентнага балу доверы, якія можна перапрацаваць і падходзяць для выкарыстоўвання ў корпаратывах.
Готовая аплікацыя
Готавыя прыкладніцы можа прыймаць дакументы у фармаце PDF, з’явы (JPG, PNG, TIFF) і XML. З’явы сторанак практычна чысцяцца за дапамою адзьмянення контрасту, усунення шума і скорэктавання нахылу; пасля чаго іх чытае PaddleOCR, а для сторанак з малым колькасцю тэксту выкарыстоўваецца GPT-4o vision fallback. Пошук здзейснюецца за дапамою гібрантнага індэксу FAISS і BM25, з’едыненага методам Reciprocal Rank Fusion, а multi-query RAG з расчытам рэвалюяціі доверлівасці выдэляе сэм полей дакумента. Рэзультаты выкладваюцца у эксел-справаздачы з каляровай кодацыяй, у якім практычна ўсе выданыя поля, карточка стаўкі і неасправлены тэкст, адрэсаваны 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 забягае про тыпы тах, кабеякія маглі буць серыяваныя. За допамогай 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 і embedding
Модуль LLL абранае OpenAI API для генеравання тэксту, візуальных дадзей і структураванага 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: падчырэдзенне
Сканаванні неапдэйтаваных дагоў часта маюць шум, нізкі контраст і лёгкае спяшчэнне. Перад адборамом тэкста кожныя знімкі сторанак праходзяць праз тры этапы падвышэння якосці.
Падвышэнне якосці знімка
# 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.
Разлікватаванне іх адзінальна мера, якая не дазволяе LLM плутаць цяпункт абонента працы на 400 слоў з простым формулюваннем „Net 30“.
Карточкі тарыфаў на разных стороніцах
Карточкі тарыфаў ўважаюцца найскладнейшым элементам. Яны распадаюцца на калькі стороніцаў, назвы столбцоў разлічаюцца з стороніцы на стороніцу („Номер айтэма“ на адной, „SKU“ на іншай), а лініі можа быць дзесяткі. Адна толькі запытка да LLM да ўсіх стороніцаў прыведзе да перапоўнення кантэксту і плутання структуры столбцоў. Рашэннем ёсць выкарыстоўваць даныя за кожной стороніцай, а пасля стандартызаваць структуры.
# 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"
}
Kожэн рад з кожноі сторынкі потым перэпідключаецца за допомогою гэтага мапування і об’еднуецца в одну едыну табліцю.
Кешування 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
Рэзультаты паказваюцца ў пяці вікнах, прычаму вікно «Журнал запытак» забезпечвае пэўную працэйнасць для дыбаггінгу і аўдыту.
Для кожнага выявленага поля вікно паказвае кожны запыт RAG, які быў выкананы, найболей значныя фрагменты з рангамі FAISS, BM25 і RRF, саму запытку, якая была надаслана GPT-4o, первасны адказ мадэлі, а таксама колькасць токэнаў у вхідным і выходным даных, што дапамагае сочыць за віткамі.
Для price_details таксама паказваюцца кожны запыт на адзін старонік і запыт на кананізацыю. У корпаратыўных настройках такая можлівасць аўдыту мае важлівое значэнне.
Праз тут прыводзяцца калькі скрыншотаў з інтэрфейсу, пачынаючы з вікна «Выявленыя поля».
Анімаваны кадр інтэрфейса пад час работы:
Шаг 12: прыклад выканання
Тэстовы дакумент — грамадзкі догавор на закупкі, сканаваны файл з 10 старонікамі, які мае вялікую складнасць, укладаючы ў сябе табліцы і рукапісныя сімвалы.
Табліцы 1: выкарыстоўваныя поля
Выкарыстоўваныя поля — гэта законныя назвы Стороны А і Стороны Б, дата пачатку, дата заканчэння, графік платэжаў і умовы платэжаў. Таксацыі, або карточка тарыфаў, таксама выкарыстоўваюцца, але ўжо таму, што яе структура дынамічная, яна адображаецца окрэма. Кожная значэнне супроводжваецца Фрагментам выхаднага тэксту, які дапамагае у аўдыты і карыстоўванні керавання, а таксама тут прыменяецца зелены, жоўты і чырвоны колеры для паказу прызнакаў надзеі.
Табліцы 2: карточка тарыфаў
Дынамічныя столбцы складаюцца з кожнай стороніцы з тарыфамі пасля кананізацыі, і табліцы застаецца порожней, калі у контракце няма карточкі тарыфаў.
Табліцы 3: неапрануты тэкст OCR
Полны выхадны тэкст пасля аналізу OCR за кожной стороніцай, які ўжоць корыстны для пераканання ў правільнасці выкарыстоўваных даных і выявлення прыгарожаных полей.
Табліцы 4: логі запытаў
Кожны запыт API LLM фіксуецца разам з яго галоўнымі фрагментамі, точным запытам і выхадным рэзультатам, што дапамагае у манітарынгу і ацэнцы.
Апроксыматыўны пацёрк загальнага выкарыстання токэнаў у всіх вызовах LLM для аднаго контракту: прыблізна 17 тыс. токэнаў увайшчання і прыблізна 4 тыс. токэнаў вывайшчання. Гэта дае прыблізна $0,005 з gpt-4o-mini і прыблізна $0,10 з gpt-4o. Нават па вышэйшай цэне гэта дэшэўле, быстрэй і структураванае, чым ручная перакантролька.
Аптака архітэктуры
Полны канцэптуальны ляміт сістэмы:
Рашэння па дызайне
Рашэнні, якія лежачы за сістэмай, збалансаваюць точнасць выкарыстання дадзэнняў, швальнасць апрацоўкі, масштабаванне і косты для аналізу контрактаў у рэальным свеце.
Куды можа развіцца сістэма
Чынная сістэма гатовая для выкарыстання з адным дакументам. Наступныя крокі спрыяюць масштабаванню, падышчу тачнасці і аналітыцы на рэвэлю портфеляў:
- Большае колькасць полей для выкарыстання дадзэнняў, дапаможныя перакантролькі і паралельная апрацоўка пакетаў для большых портфеляў контрактаў.
Спалучэнне OCR, модэляў эмбедынга і агентскага выкарыстоўвання дадзеных стварае эфектываю сістэму інтэлекту для аналіза контрактаў. Аналіз дакументаў, семантычны пошук і выкарыстоўвання структураваных поль у аднам модульным працэсе дазволяюць працаваць з складнымі контрактамі з мінімальным ручным втручанням і ствараць структураваны выхід, гатовы для подальшага аналізу. Архітэктура ёсць масштабаваная і розшыржальная, што ўтварае моцную базу для автоматызаціі аналіза контрактаў і скорачэння часу, які патрабуеяць ручныя перагляды.