Inteligencia de contratos con OCR, RAG híbrido y LangGraph
Un pipeline de extracción de contratos: limpieza de imágenes, PaddleOCR con fallback a GPT-4o, recuperación mediante FAISS y BM25, RAG para múltiples consultas y un informe en Excel con puntuaciones.
El problema de los contratos
Los contratos son la base de toda organización, desde acuerdos con proveedores y órdenes de servicio hasta licitaciones gubernamentales. En una empresa de tamaño medio pueden haber cientos en vigor al mismo tiempo, y decenas de miles en una gran empresa. Aun así, los datos comerciales que contienen permanecen guardados en archivos no estructurados: PDF escaneados, páginas fotografiadas, exportaciones en XML de herramientas de adquisiciones. Quien quiera conocer las condiciones de pago de un proveedor o la fecha de vencimiento de un contrato debe abrir el documento, leer páginas y páginas de texto legal genérico y luego introducir la respuesta en una hoja de cálculo.
De esta rutina surgen tres problemas:
- Se pasan por alto los plazos y las renovaciones porque la cláusula relevante está oculta en lo profundo del texto.
- A gran escala, las diferencias en los precios de los proveedores pasan desapercibidas y causan pérdidas económicas.
- El seguimiento manual de las condiciones de pago genera riesgos en cuanto al cumplimiento normativo.
El problema fundamental es que los datos de los contratos se encuentran en documentos y no en bases de datos. Una consulta que debería resolverse con una sola consulta SQL lleva a una persona aproximadamente 20 minutos.
El sistema aquí automatiza esa revisión de principio a fin. Un contrato se ingresa en formato PDF, escaneo o XML, y segundos después un informe de Excel muestra los términos comerciales clave, cada uno con una puntuación de confianza y una referencia de página. PaddleOCR, FAISS, BM25, Reciprocal Rank Fusion y GPT-4o se ejecutan todos dentro de una máquina de estados LangGraph.
Diferentes equipos se benefician de maneras distintas. Los departamentos de compras revisan los contratos más rápidamente y integran a los proveedores con mayor celeridad. El área financiera detecta discrepancias en los precios entre los contratos de los proveedores. Los departamentos legales y de cumplimiento mantienen registros auditables con referencias a las fuentes originales. El área operativa hace un seguimiento de las renovaciones y plazos sin necesidad de esfuerzo manual.
La implementación abarca estas técnicas:
- Un DAG LangGraph de múltiples etapas para el procesamiento de documentos.
- Mejora de imágenes (eliminación de ruido, CLAHE, corrección de inclinación) antes del OCR para aumentar la precisión.
- Una configuración de OCR con doble estrategia: PaddleOCR como opción principal y GPT-4V como respaldo.
- RAG híbrido, donde FAISS, BM25 y la fusión de rangos recíprocos juntos superan a cualquiera de los sistemas de recuperación por separado.
- Prompts RAG para múltiples consultas que maximizan la recuperación en diferentes redacciones de contratos.
- Extracción de tarifas de múltiples páginas con normalización de esquemas impulsada por LLM.
- Resultado con puntuación de confianza y auditable, adecuado para uso empresarial.
La aplicación finalizada
La aplicación final puede procesar contratos en formato PDF, imagen (JPG, PNG, TIFF) y XML. Las imágenes de las páginas se limpian mediante eliminación de ruido, ajuste de contraste y corrección de inclinación, y luego PaddleOCR las lee, con una solución alternativa de GPT-4o vision para páginas escasas. La recuperación de información se basa en un índice híbrido FAISS y BM25 fusionado mediante Reciprocal Rank Fusion, y RAG multi-query con evaluación de confianza que extrae siete campos del contrato. Los resultados se presentan en un informe de Excel con codificación por colores que incluye los campos extraídos, una tarifa y el texto OCR en bruto; además, una interfaz Streamlit los muestra junto con un registro detallado de las solicitudes para garantizar total transparencia.
Todo el proceso en una sola vista:
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
└─────────────────────┘
Estructura del proyecto
El repositorio está organizado de la siguiente manera.
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
Paso 0: configuración del entorno
Comience con un entorno virtual para mantener aisladas las dependencias. En macOS o Linux:
python -m venv venv
source venv/bin/activate
En Windows:
python -m venv venv
venv\Scripts\activate
A continuación, instale las bibliotecas que abarcan el OCR, la recuperación de información, los embeddings, la orquestación de LLM y la interfaz Streamlit.
pip install -r requirements.txt
Los paquetes clave en requirements.txt se ven así.
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
Un archivo .env contiene las credenciales de OpenAI y la elección del modelo.
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini # or gpt-4o for higher accuracy
Paso 1: estado del pipeline y configuración central
LangGraph hace que todo pase a través de un objeto de estado compartido, por lo que vale la pena definirlo antes de crear cualquier nodo. Actúa como la única fuente de verdad para el pipeline.
# models/state.py
from typing import TypedDict, Dict, List, Any
class ContractState(TypedDict, total=False):
# ── Input ──────────────────────────────────────────────────────────────
file_path: str
file_type: str # "pdf" | "image" | "xml"
original_filename: str
# ── Pre-processing ─────────────────────────────────────────────────────
page_image_paths: List[str] # enhanced page image paths
deduplicated_page_indices: List[int] # surviving page indices after dedup
# ── OCR / Text ─────────────────────────────────────────────────────────
raw_text_by_page: Dict[int, str] # page_num → extracted text
full_text: str # concatenated text (all pages)
# ── Indexing ───────────────────────────────────────────────────────────
chunks: List[str]
chunk_metadata: List[Dict[str, Any]]
session_faiss_ready: bool
# ── Extraction ─────────────────────────────────────────────────────────
extracted_fields: Dict[str, Dict[str, Any]]
# ── Output ─────────────────────────────────────────────────────────────
excel_output_path: str
errors: List[str]
# ── Meta ───────────────────────────────────────────────────────────────
current_step: str
processing_log: List[str]
prompt_log: List[Dict[str, Any]]
TypedDict ofrece indicaciones de tipo y autocompletado en el IDE, al mismo tiempo que mantiene el estado serializable. Con total=False, cada campo es opcional, ya que cada nodo solo guarda lo que genera.
Todos los parámetros ajustables se encuentran en un único archivo de configuración.
# 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
Paso 2: Servicios de LLM e embedding
El módulo LLM encapsula la API de OpenAI para la generación de texto, imágenes y JSON estructurado. Se encarga de la construcción de prompts, la codificación de imágenes, las reintentos, el seguimiento del uso de tokens y el análisis tolerante de JSON, lo que garantiza la fiabilidad de los flujos de trabajo de extracción.
# 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 {}
El servicio de incrustación utiliza Sentence Transformers para generar vectores densos a partir del texto de los contratos. Las incrustaciones se normalizan para que la búsqueda de similitud semántica siga siendo eficiente, y sirven tanto para el indexado como para la recuperación de información.
# 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]
Paso 3: preprocesamiento
Las imágenes escaneadas de contratos en bruto suelen ser ruidosas, tener bajo contraste y estar ligeramente distorsionadas. Antes del OCR, cada imagen de página pasa por tres etapas de mejora.
Mejora de imágenes
# 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)
Las etapas se realizan en secuencia.
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
¿Por qué CLAHE y no una simple ecualización de histograma? La ecualización estándar amplía el contraste de forma global y puede sobreiluminar las áreas que ya eran claras. CLAHE, abreviatura de Contrast Limited Adaptive Histogram Equalization, funciona en pequeñas secciones. El texto de un sello descolorido o de un encabezado pálido se mejora sin que se diluyan las áreas limpias.
Eliminación de páginas duplicadas
Los contratos largos suelen contener páginas repetidas, como portadas duplicadas, separadores en blanco y copias carbón. El OCR de todas ellas consume tiempo y carga el índice con texto duplicado.
# 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
El hashing perceptual (phash) reduce cada imagen de página a una huella digital de 64 bits formada a partir de componentes de frecuencia. Las páginas cuya distancia de Hamming con una página anterior es 6 o menor se consideran visualmente idénticas, por lo que tanto las copias exactas como las casi idénticas, como una página escaneada dos veces con diferentes calidades, son detectadas. Un ejemplo de ejecución:
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 ✓
También se agrega un analizador XML genérico, junto con la función final que llama a todos los pasos de preprocesamiento en secuencia.
# 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",
}
Paso 4: OCR de doble estrategia
El OCR es el paso más arriesgado, ya que un cuadro o campo de formulario omitido puede arruinar la extracción posterior. El nodo utiliza dos modelos para ampliar la cobertura.
# 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)
Cada página sigue este árbol de decisiones.
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)
¿Por qué 80 caracteres como límite? Las páginas con poca información contienen tablas con pocas palabras por celda, formularios en su mayoría vacíos, así como espacios para sellos y firmas. PaddleOCR funciona bien con texto denso, pero tiene dificultades con diseños altamente estructurados, mientras que GPT-4o Vision puede analizar toda la imagen y reconstruir una tabla utilizando separadores tipo pipe.
La instrucción para el procesamiento visual es deliberadamente específica.
#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])
Luego, las piezas se integran en el pipeline OCR completo.
#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",
}
Paso 5: creación del índice híbrido
Con texto limpio en cada página, lo siguiente es crear una base de conocimientos buscable, utilizando una estrategia de fragmentación padre-hijo.
Fragmentación
# 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",
}
La idea central es la recuperación de padres e hijos.
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)
Los fragmentos pequeños coinciden con las consultas de manera precisa, pero al entregarle al modelo de lenguaje grande un fragmento de 600 caracteres a menudo se pierde el contexto circundante necesario para comprenderlo. Al almacenar las páginas y acceder a ellas posteriormente, se obtiene tanto precisión como contexto.
El almacén híbrido
HybridVectorStore es el núcleo de la recuperación. Combina dos métodos de búsqueda fundamentalmente diferentes.
# 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)
Usar ambos métodos es beneficioso, ya que cada uno compensa las debilidades del otro.
Así funciona la Fusión de Rangos Recíprocos (RRF). La fórmula 1/(k + rank) convierte rangos arbitrarios en puntuaciones comparables. Con k=60, un resultado en primer lugar obtenido por cualquiera de los sistemas suma 1/61, aproximadamente 0,016. Un fragmento que ocupa el primer lugar según FAISS y el tercero según BM25 suma 1/61 + 1/63, unos 0,032, lo que supera a un fragmento detectado solo por un sistema. Esto es útil para contratos con terminología inusual, donde BM25 tiene dificultades, y para textos genéricos copiados y pegados, donde FAISS genera demasiadas coincidencias.
Paso 6: el agente de extracción
Aquí radica la inteligencia del sistema. Para cada uno de los siete campos del contrato, el agente de extracción realiza varias consultas dirigidas, obtiene las páginas más relevantes y pide a GPT-4o que extraiga el valor correspondiente.
Los siete campos configurables
# 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)
]
Recuperación con múltiples consultas
Cada campo recibe cuatro variaciones de consulta para maximizar la tasa de recuperación. Una sola consulta para “fecha de inicio” podría pasar por alto un contrato que indique “Comienzo:” o “Vigencia a partir de:”, mientras que múltiples enfoques permiten capturar estas variantes.
# 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",
],
}
El flujo de recuperación que se muestra a continuación se repite para cada campo.
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
Anclaje en la página cero para los nombres de los partidos
En los contratos gubernamentales, los nombres de los partidos suelen aparecer solo en el membrete o en la página de portada, y no en las cláusulas del cuerpo del documento. Si un umbral de puntuación descarta a la página de portada porque las cláusulas internas tienen mayor prioridad, el nombre del partido desaparece. Un caso especial mantiene activa la página cero para esos campos.
# 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]
Expansión al contexto padre, con filtrado
Antes de que el contexto llegue al LLM, los fragmentos hijos se expanden a páginas padre completas. Las páginas con puntuación baja se descartan para mantener el enfoque de la solicitud.
# 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)
Las instrucciones de extracción
Cada campo cuenta con una instrucción cuidadosamente redactada que indica a GPT-4o qué buscar y cómo responder.
# 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": ""}}"""
Por ejemplo, la pregunta payment_timeline está muy enfocada.
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.
Se diferencia de payment_conditions, que recoge toda la narrativa.
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.
Mantenerlos separados evita que el LLM confunda una cláusula de pago de 400 palabras con un simple “Net 30”.
Fichas de tarifas en varias páginas
Las fichas de tarifas son el campo más complicado. Abarcan varias páginas, los nombres de las columnas varían de una página a otra (“Número de artículo” en una, “SKU” en otra) y las filas suman decenas. Una sola llamada al LLM para procesar todas las páginas sobrecargaría el contexto y complicaría los esquemas de columnas. La solución es extraer la información por página y luego normalizar los esquemas.
# 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)
La asignación que se obtiene se parece más o menos a esto.
{
"Item Number": "SKU",
"Item Numbers": "SKU",
"SKU": "SKU",
"Price": "Unit Cost",
"Unit Cost": "Unit Cost",
"Amount": "Unit Cost"
}
Cada fila de cada página se vuelve a procesar con esa asignación y se combina en una sola tabla unificada.
Almacenamiento en caché de OCR
Un contrato de 50 páginas requiere de 5 a 10 minutos para ser procesado por OCR, y esperar tanto tiempo en cada ejecución repetida es problemático cuando se están ajustando los parámetros o se está corrigiendo un error. El caché detecta un informe anterior de Excel del mismo documento y permite pasar directamente a la fase de extracción.
# 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
Con el caché habilitado, el flujo se ejecuta de esta manera.
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
Paso 9: el libro de trabajo de Excel
La salida final y los registros en bruto se guardan en un archivo de Excel, el cual también puede alimentar una base de datos. Los colores dependen del nivel de confianza: verde para un 80% o más (alto), amarillo para un 50% a un 79% (medio), y rojo para menos del 50% (bajo o no encontrado). Esos valores indican qué atributos requieren revisión por parte de un humano.
#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",
}
El nombre del archivo incluye una marca de tiempo para la trazabilidad.
contract_name_extracted_20260523_165458.xlsx
Paso 10: creación del grafo
El flujo de trabajo de LangGraph conecta todo el proceso, desde la preprocesamiento y el OCR, pasando por la recuperación de información y la extracción de campos basada en IA, hasta la generación de archivos 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()
Paso 11: la interfaz de usuario de Streamlit
Ejecútela localmente con los comandos a continuación.
cd medium/contract-intelligence
streamlit run app.py
Los resultados se muestran en cinco pestañas, y la pestaña “Registro de solicitudes” ofrece total transparencia para la depuración y la auditoría.
Para cada campo extraído, la pestaña enumera cada consulta RAG que se ejecutó, los fragmentos de mayor relevancia según el ranking de FAISS, el ranking de BM25 y la puntuación RRF, toda la solicitud enviada a GPT-4o, la respuesta bruta del modelo, así como los conteos de tokens de entrada y salida para poder hacer un seguimiento de los costos. Para price_details, también se incluyen cada llamada por página y la llamada de canonización. En entornos empresariales, este tipo de trazabilidad es muy importante.
A continuación se presentan algunas capturas de pantalla de la interfaz, comenzando con la pestaña “Campos extraídos”.
Una captura animada de la interfaz en uso:
Paso 12: un ejemplo de ejecución
El documento de prueba es un contrato público de adquisiciones estatales, un archivo escaneado del mundo real de 10 páginas con mucha complejidad, que incluye tablas y caracteres manuscritos.
Pestaña 1: campos extraídos
Los campos obtenidos son los nombres legales de la Parte A y la Parte B, la fecha de inicio, la fecha de finalización, el cronograma de pagos y las condiciones de pago. También se extrae la información de precios o la tarifa, pero su estructura es tan dinámica que se muestra por separado. Cada valor viene acompañado de un Fragmento de origen, lo cual ayuda en la auditoría y la gobernanza; además, aquí también se aplica la codificación de confianza en verde, amarillo y rojo.
Pestaña 2: tarifa
Las columnas dinámicas se generan a partir de cada página de precios tras su canonización, y la pestaña permanece vacía si el contrato no cuenta con tarifa alguna.
Pestaña 3: texto OCR sin procesar
La salida completa del OCR, página por página, útil para verificar los extracciones y localizar campos faltantes.
Pestaña 4: registros de solicitudes
Cada llamada a la API del LLM se registra junto con sus fragmentos más relevantes (top-K), la solicitud de entrada exacta y la salida, con el fin de monitorearla y evaluarla.
Una estimación aproximada del uso total de tokens en todas las llamadas a los LLM para un contrato: unos 17 mil tokens de entrada y unos 4 mil tokens de salida. Eso equivale a aproximadamente $0.005 con gpt-4o-mini y unos $0.10 con gpt-4o. Incluso a un precio más alto, sigue siendo más económico, rápido y estructurado que la validación manual.
Resumen de la arquitectura
El flujo completo de extremo a extremo del sistema:
Decisiones de diseño
Las decisiones detrás del sistema equilibran la precisión de la extracción, la velocidad de procesamiento, la escalabilidad y el costo para el análisis de contratos en el mundo real.
Hacia dónde podría evolucionar
El sistema actual está listo para la extracción de un único documento. Los próximos pasos apuntan a la escalabilidad, la precisión y el análisis a nivel de cartera:
- Más campos de extracción, verificaciones adicionales de validación y procesamiento en lotes paralelo para grandes carteras de contratos.
La combinación de OCR, modelos de embeddings y extracción por agentes genera un sistema eficiente de inteligencia contractual. El análisis de documentos, la recuperación semántica y la extracción estructurada de campos en un único flujo de trabajo modular permiten procesar contratos complejos con poco esfuerzo manual y generar resultados estructurados listos para análisis posteriores. La arquitectura es escalable y extensible, lo que la convierte en una base sólida para automatizar el análisis de contratos y reducir el tiempo necesario para la revisión manual.
Leer más
- Notas prácticas: Construir un agente de IA autocorregible con Self-RAG y LangGraph — Guía paso a paso de las Notas prácticas: Construir un agente de IA autocorregible con Self-RAG y LangGraph: contratos, verificaciones y espacios de código listos para usar para los equipos que implementan este patrón.
- Construir un agente de documentación autocorregible con LangGraph y Ollama — Un pipeline de reflexión local en el que un modelo redacta las documentaciones de la API en Python, un segundo las critica y un tercero armoniza el resultado, conectados como un bucle LangGraph en Ollama.