Ingestión de datos RAG para la producción en bancos: PDF, OCR, tablas y conectores
Cómo incorporar documentos bancarios para RAG: clasificar PDFs, ejecutar OCR con indicadores de revisión, mantener la estructura de Word, manejar tablas e imágenes, y crear conectores resistentes.
Fuentes: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
El sistema RAG de un banco falla en silencio
Imagínese a un banco de tamaño medio que desarrolla un sistema de generación mejorada mediante recuperación de información a partir de su base de conocimientos interna. Hay miles de documentos incluidos: circulares regulatorias del RBI, manuales de políticas, directrices de Basilea III, procedimientos de procesamiento de préstamos, registros de KYC y guías operativas. El proyecto parece ir bien. Un ingeniero crea la base de datos vectorial, otro diseña el pipeline de recuperación, y un tercero conecta el LLM y desarrolla la interfaz. Luego el sistema entra en funcionamiento.
Después de tres semanas, los usuarios comienzan a quejarse de respuestas incorrectas, y la causa no es una alucinación. El recuperador encuentra el documento correcto y el modelo responde fielmente basándose en lo que le fue proporcionado. El problema comenzó mucho antes. Una circular del RBI sobre las normas de liquidez para las NBFC existía únicamente en formato PDF escaneado, y durante el proceso de OCR la cifra “10%” apareció como “lO%”, con el dígito 1 reemplazado por una L minúscula. El analizador la aceptó sin realizar ninguna verificación. El modelo de embedding codificó el texto dañado, el almacén de vectores lo guardó, el recuperador lo posicionó cerca de la cima, y el LLM le dijo con confianza a un oficial de cumplimiento que la tasa requerida era “lO por ciento.”
La recuperación de información no falló ni tampoco la generación. Cada componente realizó exactamente lo para lo que fue diseñado. El defecto se encontraba en la primera etapa del proceso, la ingesta de datos, que es uno de los puntos ciegos más comunes y costosos en los sistemas RAG en producción. Los equipos pasan semanas ajustando modelos de embeddings, procesos de fragmentación, algoritmos de reclasificación y prompts, mientras consideran que la ingesta es solo un paso previo de preprocesamiento. Sin embargo, la calidad de la recuperación ya está determinada mucho antes de que llegue cualquier consulta. Si la fuente se analiza incorrectamente, ningún modelo de embeddings, índice o LLM podrá recuperar los hechos originales.
A continuación se aborda la ingesta en los sistemas RAG empresariales: cómo manejar fuentes estructuradas, semiestructuradas y no estructuradas, lidiar con documentos multimodales, diseñar soluciones escalables, detectar errores en el análisis y integrarlo todo mediante Python funcional.
Funciones de la ingesta
En un pipeline RAG, la fase de ingesta toma los documentos de origen en bruto, independientemente de su formato o procedencia, y los convierte en texto limpio y estructurado, listo para ser dividido en fragmentos e incrustado.
La etapa se lleva a cabo íntegramente en el pipeline offline, ejecutándose antes de que surja cualquier consulta del usuario. Lo que genera son textos analizados y limpiados junto con metadatos, los cuales sirven como entrada para cada paso posterior. Aquí tienen lugar tres tareas: Carga obtiene el documento en bruto desde su origen, que puede ser un sistema de archivos, un bucket de S3, un sitio de SharePoint, una API REST o una base de datos. Análisis extrae el texto limpio del formato original, ya sea decodificando un PDF, eliminando etiquetas HTML, deserializando JSON o aplicando OCR a una imagen. Extracción inicial de metadatos captura la fuente, la fecha de creación, el tipo de documento y cualquier detalle específico del formato visible en el momento del análisis. Después vienen las tareas de limpieza, fragmentación e incrustación. La única función de la etapa de ingestión es obtener texto limpio a partir del material en bruto.
La complicada realidad de los documentos bancarios
La base de conocimientos de un banco no se presenta en formato Markdown ordenado. Se trata de décadas de formatos acumulados: PDF escaneados de la década de 1990, archivos Word con cambios registrados, páginas HTML de un portal interno de cumplimiento, exportaciones JSON de un sistema de gestión de políticas, correos electrónicos con archivos adjuntos y sitios de SharePoint con estructuras de carpetas complejas. Cada formato requiere su propio analizador, y cada analizador falla a su manera. Peor aún, las fallas de los analizadores son silenciosas. Un analizador que genera texto corrupto no lanza ninguna excepción; simplemente devuelve ese texto corrupto y continúa con su trabajo.
PDF: tres tipos, tres formas de tratarlos
Los PDF dominan el material regulatorio y de cumplimiento, y también son los más difíciles de analizar. Tres variedades requieren un manejo diferente:
- PDFs de capa de texto provienen de procesadores de textos o herramientas de impresión a PDF. El texto se encuentra en el flujo de contenido y puede extraerse directamente.
- PDF escaneados son archivos de imagen creados a partir de papel físico. No cuentan con capa de texto en absoluto, por lo que el OCR es obligatorio. La mayoría de los circulares antiguos y políticas históricas pertenecen a esta categoría.
- PDF híbridos son escaneos que han recibido una capa de texto OCR de mala calidad. Son los más riesgosos, ya que un analizador ingenuo utiliza esa capa llena de errores y asume que todo funcionó correctamente.
Distinguirlos es el primer paso, y la función de clasificación a continuación examina un PDF e indica de qué tipo se trata.
import fitz # PyMuPDF
from pathlib import Path
def classify_pdf(pdf_path: str) -> str:
"""
Classify a PDF as text-native, scanned, or hybrid.
Returns: 'text', 'scanned', or 'hybrid'
"""
doc = fitz.open(pdf_path)
total_pages = len(doc)
pages_with_text = 0
total_text_length = 0
for page in doc:
text = page.get_text("text").strip()
if text:
pages_with_text += 1
total_text_length += len(text)
doc.close()
# No text layer at all: scanned PDF
if pages_with_text == 0:
return "scanned"
# Text exists but average is suspiciously short: likely hybrid with poor OCR
avg_text_per_page = total_text_length / total_pages
if avg_text_per_page < 100 and pages_with_text < total_pages * 0.5:
return "hybrid"
return "text"
def load_text_pdf(pdf_path: str) -> list[dict]:
"""
Extract text from a text-native PDF using PyMuPDF.
Returns a list of page dicts with text and metadata.
"""
doc = fitz.open(pdf_path)
pages = []
for page_num, page in enumerate(doc, start=1):
text = page.get_text("text")
pages.append({
"page_number": page_num,
"text": text.strip(),
"source": pdf_path,
"total_pages": len(doc),
})
doc.close()
return pages
Los archivos escaneados requieren un proceso de OCR, y el gráfico compara las tres opciones realistas de producción.
Para un banco que debe cumplir con las circulares del RBI, Azure Document Intelligence es la opción más práctica. Puede manejar escaneos de calidad variable, entender la estructura de las tablas dentro de las páginas escaneadas y evaluar el nivel de confianza de cada palabra, lo que permite marcar las extracciones con bajo nivel de confianza para su revisión por parte de un humano. El cargador que se muestra a continuación lo utiliza y aplica esa lógica.
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
def load_scanned_pdf_azure(
pdf_path: str,
endpoint: str,
api_key: str,
) -> list[dict]:
"""
Extract text from a scanned PDF using Azure Document Intelligence.
Returns pages with text, confidence scores, and table data.
"""
client = DocumentAnalysisClient(
endpoint=endpoint,
credential=AzureKeyCredential(api_key),
)
with open(pdf_path, "rb") as f:
poller = client.begin_analyze_document("prebuilt-read", f)
result = poller.result()
pages = []
for page in result.pages:
page_text = ""
low_confidence_words = []
for line in page.lines:
page_text += line.content + "\n"
# Flag words with confidence below threshold
for word in page.words:
if word.confidence < 0.85:
low_confidence_words.append({
"word": word.content,
"confidence": word.confidence,
})
pages.append({
"page_number": page.page_number,
"text": page_text.strip(),
"low_confidence_words": low_confidence_words,
"needs_review": len(low_confidence_words) > 5,
"source": pdf_path,
})
return pages
La bandera needs_review es importante en entornos regulados. Una página con más de cinco palabras de baja confiabilidad no debería pasar desapercibida. Debe formar parte de la cola de revisión humana, o al menos llevar una bandera en sus metadatos para que una evaluación posterior pueda ponderarla en consecuencia.
Documentos de Word, HTML y fuentes estructuradas
Los archivos de Word son los más utilizados para redactar políticas y procedimientos internos. Contienen encabezados, tablas y listas que facilitan su recuperación posterior, pero desaparecen con un extractor de texto simplista. El cargador analiza el documento y mantiene esa estructura.
from docx import Document as DocxDocument
from docx.oxml.ns import qn
def load_docx(docx_path: str) -> list[dict]:
"""
Load a DOCX file preserving heading structure and tables.
Returns a list of content blocks with type and text.
"""
doc = DocxDocument(docx_path)
blocks = []
current_section = "Introduction"
for element in doc.element.body:
# Paragraphs
if element.tag == qn("w:p"):
para = element
style = para.style.name if hasattr(para, "style") else ""
text = para.text.strip()
if not text:
continue
if style.startswith("Heading"):
level = int(style.replace("Heading ", "")) if style != "Heading" else 1
current_section = text
blocks.append({
"type": "heading",
"level": level,
"text": text,
"section": current_section,
"source": docx_path,
})
else:
blocks.append({
"type": "paragraph",
"text": text,
"section": current_section,
"source": docx_path,
})
# Tables
elif element.tag == qn("w:tbl"):
table_data = []
for row in element.findall(f".//{qn('w:tr')}"):
row_data = []
for cell in row.findall(f".//{qn('w:tc')}"):
cell_text = " ".join(
p.text for p in cell.findall(f".//{qn('w:t')}")
if p.text
)
row_data.append(cell_text.strip())
table_data.append(row_data)
blocks.append({
"type": "table",
"data": table_data,
"section": current_section,
"source": docx_path,
})
return blocks
El seguimiento por secciones rinde frutos. En un procedimiento de KYC con cuarenta párrafos, saber que una parte proviene de “Sección 3: Diligencia debida mejorada” y no de “Sección 1: KYC estándar” constituye metadatos que mejoran significativamente la precisión en la recuperación de información.
Los portales de cumplimiento suelen ser aplicaciones web, por lo que la carga de HTML debe omitir la navegación, los anuncios, las banderas de cookies y otros elementos genéricos que rodean al contenido real.
import httpx
from bs4 import BeautifulSoup
def load_html_page(url: str, headers: dict | None = None) -> dict:
"""
Load an HTML page and extract main content, stripping boilerplate.
"""
response = httpx.get(url, headers=headers or {}, follow_redirects=True)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
# Remove boilerplate elements common in compliance portals
for tag in soup(["nav", "header", "footer", "script", "style",
"aside", "advertisement", ".cookie-banner",
".sidebar", ".breadcrumb"]):
tag.decompose()
# Find main content area (adjust selectors for your portal)
main_content = (
soup.find("main")
or soup.find("article")
or soup.find(id="content")
or soup.find(class_="content")
or soup.find("body")
)
text = main_content.get_text(separator="\n", strip=True) if main_content else ""
return {
"url": url,
"title": soup.title.string if soup.title else "",
"text": text,
"source": url,
}
Las fuentes estructuradas como los sistemas de gestión de políticas, las bases de datos de casos y los registros de transacciones requieren un enfoque diferente. Su texto ya está estructurado, por lo que las preguntas son qué campos incorporar y cómo presentar las filas como texto recuperable.
import json
import psycopg2
from typing import Any
def load_json_policies(json_path: str, text_fields: list[str]) -> list[dict]:
"""
Load a JSON array of policy documents, extracting specified text fields.
Banking example: a JSON export from a policy management system.
"""
with open(json_path, "r", encoding="utf-8") as f:
policies = json.load(f)
documents = []
for policy in policies:
# Combine relevant text fields into a single document
text_parts = []
for field in text_fields:
if field in policy and policy[field]:
text_parts.append(f"{field.replace('_', ' ').title()}: {policy[field]}")
documents.append({
"text": "\n\n".join(text_parts),
"policy_id": policy.get("id"),
"policy_type": policy.get("type"),
"effective_date": policy.get("effective_date"),
"jurisdiction": policy.get("jurisdiction"),
"source": json_path,
})
return documents
def load_from_postgres(
connection_string: str,
query: str,
text_column: str,
metadata_columns: list[str],
) -> list[dict]:
"""
Load documents from a PostgreSQL database.
Banking example: regulatory requirement records from a compliance database.
"""
conn = psycopg2.connect(connection_string)
cursor = conn.cursor()
cursor.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
cursor.close()
conn.close()
documents = []
for row in rows:
row_dict = dict(zip(columns, row))
documents.append({
"text": str(row_dict.get(text_column, "")),
**{col: row_dict.get(col) for col in metadata_columns},
"source": "postgresql",
})
return documents
Los bancos almacenan la mayoría de sus archivos de trabajo en SharePoint o Google Workspace, y para acceder a ellos se necesita OAuth junto con las API correspondientes de Graph o Drive.
import httpx
from typing import Generator
def load_sharepoint_library(
tenant_id: str,
client_id: str,
client_secret: str,
site_id: str,
drive_id: str,
) -> Generator[dict, None, None]:
"""
Iterate over all DOCX and PDF files in a SharePoint document library.
Yields file metadata dicts; caller is responsible for downloading and parsing.
"""
# Get access token
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
token_response = httpx.post(token_url, data={
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
"scope": "https://graph.microsoft.com/.default",
})
access_token = token_response.json()["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
# List files recursively
next_link = f"{base_url}/root/children"
while next_link:
response = httpx.get(next_link, headers=headers)
data = response.json()
for item in data.get("value", []):
if item.get("file") and item["name"].endswith((".pdf", ".docx")):
yield {
"name": item["name"],
"download_url": item.get("@microsoft.graph.downloadUrl"),
"created": item.get("createdDateTime"),
"modified": item.get("lastModifiedDateTime"),
"size": item.get("size"),
"web_url": item.get("webUrl"),
}
next_link = data.get("@odata.nextLink")
Por lotes o en tiempo real
Cada diseño de ingesta comienza con una decisión: ¿procesar los documentos a medida que llegan o en lotes programados?
- Ingesta por lotes se ejecuta según un horario, generalmente nocturno o semanal. Es adecuada cuando la actualidad no es crítica, el volumen es alto y el costo importa. Un banco que ingiere circulares mensuales del RBI no necesita información en tiempo real, y un trabajo nocturno es más económico.
- Ingesta en tiempo real funciona a través de una cola de eventos a medida que llegan los documentos. Es adecuada para casos en los que la actualidad es crítica, como un sistema de cumplimiento que debe mostrar una nueva circular minutos después de su publicación.
Los conectores externos necesitan protección contra límites de tasa, y la clase a continuación envuelve las solicitudes con intentos repetidos y retrasos.
import time
import httpx
from pathlib import Path
class RateLimitedConnector:
"""
A base connector with retry logic and rate limiting.
Use this pattern for any external API that imposes rate limits.
Banking example: connecting to an RBI document portal or a Bloomberg data feed.
"""
def __init__(
self,
base_url: str,
api_key: str,
requests_per_minute: int = 60,
max_retries: int = 3,
backoff_factor: float = 2.0,
):
self.base_url = base_url
self.api_key = api_key
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0.0
self.max_retries = max_retries
self.backoff_factor = backoff_factor
def _wait_for_rate_limit(self):
elapsed = time.monotonic() - self.last_request_time
if elapsed < self.min_interval:
time.sleep(self.min_interval - elapsed)
self.last_request_time = time.monotonic()
def fetch(self, endpoint: str, params: dict | None = None) -> dict:
url = f"{self.base_url}/{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(self.max_retries):
self._wait_for_rate_limit()
try:
response = httpx.get(url, headers=headers, params=params or {})
if response.status_code == 429:
# Rate limited: back off and retry
wait_time = self.backoff_factor ** attempt
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
if attempt == self.max_retries - 1:
raise
time.sleep(self.backoff_factor ** attempt)
raise RuntimeError(f"Failed to fetch {url} after {self.max_retries} attempts")
Ese patrón no es opcional. La mayoría de las APIs externas, incluidos los proveedores de datos financieros, bloquearán una IP o revocarán una clave una vez se supere el límite. El backoff_factor permite un retraso exponencial, con esperas de 2, 4 y 8 segundos en los intentos sucesivos, lo cual constituye la política mínima de reintentos para un conector en entorno de producción.
Tablas, gráficos e imágenes
El cambio hacia el uso de RAG en producción para el año 2025 ha pasado de pipelines basados únicamente en texto a ones multimodales. Los documentos bancarios no son puramente textuales; contienen tablas de ratios de solvencia, gráficos de tendencias de NPA, diagramas de flujos de trabajo KYC e imágenes de acuerdos firmados. Omitir el contenido no textual descarta una gran parte de la información de cada documento, y según las regulaciones esa parte suele ser la más valiosa. La tabla que aparece en una circular del RBI y que indica los umbrales numéricos exactos es precisamente lo que necesita un equipo de cumplimiento.
Una tabla puede manejarse de dos maneras: serializada como texto o extraída como datos estructurados. La elección adecuada depende del uso posterior. La función de extracción a continuación extrae las tablas de un PDF.
import fitz
import pdfplumber
def extract_tables_from_pdf(pdf_path: str) -> list[dict]:
"""
Extract tables from a PDF using pdfplumber.
Returns tables as both structured data and serialised text.
Banking example: extracting capital ratio tables from Basel III compliance reports.
"""
tables_output = []
with pdfplumber.open(pdf_path) as pdf:
for page_num, page in enumerate(pdf.pages, start=1):
tables = page.extract_tables()
for table_idx, table in enumerate(tables):
if not table or len(table) < 2:
continue
# First row as headers
headers = [str(h).strip() if h else f"col_{i}"
for i, h in enumerate(table[0])]
rows = table[1:]
# Structured data representation
structured = []
for row in rows:
row_dict = {}
for header, cell in zip(headers, row):
row_dict[header] = str(cell).strip() if cell else ""
structured.append(row_dict)
# Text serialisation (markdown table format for LLM consumption)
header_row = " | ".join(headers)
separator = " | ".join(["---"] * len(headers))
data_rows = [
" | ".join(str(cell or "").strip() for cell in row)
for row in rows
]
serialised = "\n".join([header_row, separator] + data_rows)
tables_output.append({
"page_number": page_num,
"table_index": table_idx,
"headers": headers,
"structured_data": structured,
"serialised_text": serialised,
"source": pdf_path,
})
return tables_output
El Markdown serializado es adecuado para tablas que un LLM podrá leer y analizar, como en “¿Qué ratio CET1 exige Basilea III?”. El JSON estructurado sirve para valores que se consultarán en código o se compararán numéricamente.
Tampoco se pueden pasar por alto los gráficos, diagramas e infografías. Un gráfico de las tendencias de NPA por trimestre, o un diagrama de flujo del proceso de aprobación de transacciones de FEMA, contiene información que no se encuentra en el texto circundante. El patrón común para 2025 consiste en enviar cada imagen a un modelo de lenguaje visual y almacenar la descripción generada para su uso en integraciones y búsquedas.
import fitz
import base64
import httpx
from io import BytesIO
def extract_and_describe_images(
pdf_path: str,
openai_api_key: str,
min_image_size: int = 5000, # bytes, to skip tiny decorative images
) -> list[dict]:
"""
Extract images from a PDF and generate text descriptions using GPT-4V.
Banking example: describing NPA trend charts and regulatory workflow diagrams.
"""
doc = fitz.open(pdf_path)
image_docs = []
for page_num in range(len(doc)):
page = doc[page_num]
image_list = page.get_images(full=True)
for img_idx, img_info in enumerate(image_list):
xref = img_info[0]
base_image = doc.extract_image(xref)
image_bytes = base_image["image"]
# Skip tiny decorative images
if len(image_bytes) < min_image_size:
continue
# Encode to base64 for the vision API
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
image_ext = base_image["ext"]
# Generate description with GPT-4V
response = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {openai_api_key}"},
json={
"model": "gpt-4o",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": (
"This image is from a banking regulatory document. "
"Describe its content precisely, including any numerical "
"values, trends, labels, axes, or process steps shown. "
"If it is a chart, describe what metric is shown, the "
"time period, and the key data points. If it is a "
"diagram, describe the process or relationship shown."
),
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/{image_ext};base64,{image_b64}"
},
},
],
}
],
"max_tokens": 400,
},
)
description = response.json()["choices"][0]["message"]["content"]
image_docs.append({
"type": "image_description",
"page_number": page_num + 1,
"image_index": img_idx,
"description": description,
"source": pdf_path,
})
doc.close()
return image_docs
Aquí el enunciado es crucial. Una solicitud genérica como “describe esta imagen” produce resultados genéricos. Una instrucción específica para un dominio, que solicite valores numéricos, tendencias, etiquetas o pasos del proceso, genera descripciones útiles. Un ingeniero que pregunta por las líneas de tendencia de NPA en el informe del tercer trimestre necesita una descripción que contenga “NPA”, “tendencia” y el trimestre, no una sobre un gráfico de barras con barras azules.
Audio y video
El audio y el video representan la vanguardia actual en los sistemas RAG bancarios. Las grabaciones de reuniones del consejo, transcripciones de llamadas con analistas, conferencias de prensa del RBI y videos de capacitación contienen información a la que los sistemas RAG aún no pueden acceder.
import httpx
import time
def transcribe_audio(
audio_path: str,
openai_api_key: str,
language: str = "en",
) -> dict:
"""
Transcribe an audio file using OpenAI Whisper.
Banking example: transcribing an RBI monetary policy press conference recording.
"""
with open(audio_path, "rb") as audio_file:
response = httpx.post(
"https://api.openai.com/v1/audio/transcriptions",
headers={"Authorization": f"Bearer {openai_api_key}"},
data={
"model": "whisper-1",
"language": language,
"response_format": "verbose_json", # includes timestamps per segment
"timestamp_granularities[]": "segment",
},
files={"file": (audio_path, audio_file, "audio/mpeg")},
)
result = response.json()
return {
"transcript": result.get("text", ""),
"segments": [
{
"start": seg["start"],
"end": seg["end"],
"text": seg["text"],
}
for seg in result.get("segments", [])
],
"language": result.get("language", language),
"source": audio_path,
}
Los marcos de tiempo de los segmentos en la salida son importantes. Cuando una pregunta se responde con algo dicho a los 34 minutos de una llamada del consejo, el sistema debe indicar esa posición exacta y no limitarse a citar una grabación del consejo.
Un router para cada formato
Los cargadores separados no son suficientes. Una cadena de producción debe enviar cada documento al analizador adecuado, manejar las fallas de manera adecuada, seguir el estado del procesamiento y generar un formato de salida consistente independientemente de la entrada.
from pathlib import Path
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class DocumentType(Enum):
PDF_TEXT = "pdf_text"
PDF_SCANNED = "pdf_scanned"
DOCX = "docx"
HTML = "html"
JSON = "json"
AUDIO = "audio"
UNKNOWN = "unknown"
def detect_document_type(file_path: str) -> DocumentType:
path = Path(file_path)
suffix = path.suffix.lower()
if suffix == ".pdf":
classification = classify_pdf(file_path)
return DocumentType.PDF_SCANNED if classification == "scanned" else DocumentType.PDF_TEXT
elif suffix == ".docx":
return DocumentType.DOCX
elif suffix in (".html", ".htm"):
return DocumentType.HTML
elif suffix == ".json":
return DocumentType.JSON
elif suffix in (".mp3", ".mp4", ".wav", ".m4a"):
return DocumentType.AUDIO
return DocumentType.UNKNOWN
def ingest_document(
file_path: str,
azure_endpoint: str | None = None,
azure_api_key: str | None = None,
openai_api_key: str | None = None,
) -> list[dict]:
"""
Route a document to the correct parser and return a list of content blocks
with a consistent schema regardless of input format.
Every output block contains:
- text: the extracted text content
- source: origin path or URL
- doc_type: the detected document type
- page_number: where applicable
- needs_review: flag for human review queue
"""
doc_type = detect_document_type(file_path)
results = []
try:
if doc_type == DocumentType.PDF_TEXT:
pages = load_text_pdf(file_path)
for page in pages:
page["doc_type"] = "pdf_text"
page["needs_review"] = False
results.append(page)
elif doc_type == DocumentType.PDF_SCANNED:
if azure_endpoint and azure_api_key:
pages = load_scanned_pdf_azure(file_path, azure_endpoint, azure_api_key)
for page in pages:
page["doc_type"] = "pdf_scanned"
results.append(page)
else:
logger.warning(
f"Scanned PDF detected but no OCR credentials provided: {file_path}"
)
results.append({
"text": "",
"source": file_path,
"doc_type": "pdf_scanned",
"needs_review": True,
"error": "No OCR credentials available",
})
elif doc_type == DocumentType.DOCX:
blocks = load_docx(file_path)
for block in blocks:
block["doc_type"] = "docx"
block["needs_review"] = False
results.append(block)
elif doc_type == DocumentType.AUDIO:
if openai_api_key:
transcript = transcribe_audio(file_path, openai_api_key)
for segment in transcript["segments"]:
results.append({
"text": segment["text"],
"source": file_path,
"doc_type": "audio_transcript",
"start_seconds": segment["start"],
"end_seconds": segment["end"],
"needs_review": False,
})
else:
logger.warning(f"Unknown document type, skipping: {file_path}")
except Exception as e:
logger.error(f"Ingestion failed for {file_path}: {e}")
results.append({
"text": "",
"source": file_path,
"doc_type": str(doc_type.value),
"needs_review": True,
"error": str(e),
})
return results
El esquema de salida consistente es la elección de diseño más importante. El procesador de bloques, el generador de embeddings y el enriquecedor de metadatos reciben todos la misma estructura de diccionario, independientemente de que la entrada sea un documento circular escaneado, una póliza de préstamo en formato DOCX o una transcripción de audio de un curso de cumplimiento. La bandera needs_review envía los documentos con problemas a una cola de revisión humana y evita que los errores se propaguen.
Tres causas de los problemas más comunes
La mayoría de los problemas relacionados con la calidad de las búsquedas en los sistemas RAG bancarios se deben a tres errores en la ingesta de datos.
Ignorar el tipo de documento al momento del análisis. Un extractor de texto ejecutado en un PDF escaneado no devuelve nada, o solo algunos caracteres sueltos. Ese archivo se convierte en fragmentos vacíos, vectores cercanos a cero y entradas de índice que luego se devuelven al realizar búsquedas. El modelo de lenguaje grande ve un contexto vacío e inventa una respuesta. La función classify_pdf lo evita con unas pocas líneas de código que ahorran semanas de depuración.
Tratar las tablas como texto del cuerpo. Una tabla de calificación de riesgos convertida a prosa pierde sus filas y columnas, y una cadena como “AAA 0.01% AA 0.05% A 0.10%” no puede recuperarse de manera útil. Si se serializa como una tabla en Markdown con encabezados, sí es posible hacerlo. Extraiga las tablas por separado y mantenga su estructura.
Faltan intentos de reintentado en los conectores. Cuando un conector de SharePoint sufre un tiempo de espera de red y no emite ninguna señal, el corpus acaba con huecos y nadie sabe qué documentos han desaparecido. El patrón RateLimitedConnector se activa una vez se agotan los intentos de reintentado y registra cada intento, de modo que los espacios en blanco quedan visibles.
Compromisos entre herramientas
- PyMuPDF, pdfplumber y PyPDF2. PyMuPDF es el extractor más rápido para PDFs con texto nativo. pdfplumber es más lento pero mucho mejor para tablas. PyPDF2 es el más antiguo y menos preciso, por lo que se debe evitar en nuevos flujos de trabajo. Utilice PyMuPDF para el texto y pdfplumber para las tablas del mismo archivo.
- OCR autohospedado o en la nube. Tesseract es gratuito, funciona localmente y maneja bien las escaneos de buena calidad. Para miles de circulares históricas de calidad variable, la mayor precisión de Azure Document Intelligence justifica su precio por página. No vale la pena ahorrar en umbrales regulatorios mal interpretados.
- Costo de transmisión en tiempo real. Transmitir cada archivo subido en tiempo real es la arquitectura más costosa, y un banco que agrega dos o tres documentos al día no la necesita. Úsela solo en casos donde la actualidad inmediata afecta los resultados comerciales, como cuando un departamento de trading necesita acceso instantáneo a nuevas circulares de SEBI. Una biblioteca de políticas de recursos humanos no es uno de esos casos.
- Costo multimodal. Enviar cada imagen de cada documento a través de GPT-4o genera gastos reales a gran escala. Filtre las imágenes pequeñas, como las decorativas de menos de 5 KB, descarte aquellas cuyo ratio de aspecto las identifique como logotipos o íconos, y almacene en caché las descripciones para que una imagen repetida en varios documentos sea descrita solo una vez.
El texto en bruto no es texto limpio
Lo que proporciona la fase de ingesta es texto en bruto, no texto limpio. Una etapa separada de limpieza se encarga de eliminar ruidos, eliminar duplicados, corregir errores de OCR, normalizar texto multilingüe y eliminar texto genérico, situada entre la ingesta y el procesador de fragmentos. La división es importante porque los problemas difieren: los fallos en la ingesta generan contenido vacío o corrupto, mientras que los fallos en la limpieza producen contenido que parece correcto pero contiene ruidos sutiles que afectan la calidad de los embeddings, y que solo se manifiestan al realizar pruebas de recuperación.
Para su uso en producción, los ejemplos de código necesitan registro de actividades, gestión de secretos (guardar las claves API en variables de entorno o en un gestor de secretos, nunca codificadas directamente) e instrumentación de latencia en cada etapa.
Lecturas relacionadas
- Limpieza de datos para RAG en producción: codificación, OCR, eliminación de duplicados y protección de PII — Cómo convertir documentos empresariales analizados en texto listo para su recuperación: reparación de codificaciones, eliminación de textos genéricos y duplicados, marcadores de OCR riesgosos, manejo de idiomas y enmascaramiento de PII.
- Estrategias de Particionamiento para RAG de Producción: de Tamaño Fijo, Semántico y Padre-Hijo — Un recorrido práctico por el particionamiento en RAG, desde la división de tamaño fijo y recursiva hasta los enfoques padre-hijo, proposiciones, particionamiento tardío, RAPTOR, tablas, código y pruebas de rendimiento.
- RAG Híbrido para Textos Jurídicos: Particionamiento, BM25, BGE y Reclasificación — Un pipeline RAG para textos jurídicos basado en la Constitución de la India: particionamiento padre-hijo, recuperación con BM25 y BGE, fusión de rangos, reclasificación mediante codificador cruzado y expansión de consultas, probado en pruebas de rendimiento.
- Selección de fragmentos sin consultas: uso de K-Means cuando RAG no tiene preguntas que hacer — Cuando no hay consulta del usuario, RAG no puede seleccionar el contexto. Considere K como la cantidad de resultados deseados, agrupe las incrustaciones de los fragmentos, elija los fragmentos con centroide y mantenga la cobertura además de las citaciones de origen.
- Diseño de RAG para 100 millones de documentos: particionamiento, búsqueda híbrida y costos — Qué cambia cuando un sistema RAG crece hasta alcanzar miles de millones de fragmentos: ingestión asíncrona, enrutamiento, particionamiento, recuperación híbrida, reclasificación, autorización, caché y control de costos.
- Tres formas en que falla RAG en entornos de producción: contexto fragmentado, desviación y uso incorrecto — Por qué los sistemas RAG pierden matices, degradan a medida que crecen las bases de conocimiento o dan respuestas erróneas a pesar de una buena recuperación de información, con soluciones prácticas en la capa de recuperación.
- Notas prácticas: MCP 2026: El protocolo creció así: esto es la IA en producción — Guía práctica de Notas prácticas: MCP 2026: El protocolo creció así: esto es la IA en producción: contratos, verificaciones y espacios para código listos para usar para los equipos que implementan este patrón.
- Ingeniería de un plano de control de evaluación para agentes de IA en producción — Cómo construir una evaluación continua de los agentes: planos de ejecución y control separados, afirmaciones basadas en trazas, conjuntos de datos de referencia, puertas de integración continua, pruebas durante todo el ciclo de vida y una lista de verificación para el lanzamiento.