Production RAG Data Ingestion for Banks: PDFs, OCR, Tables and Connectors
How to ingest bank documents for RAG: classify PDFs, run OCR with review flags, keep Word structure, handle tables and images, and build resilient connectors.
This article is published in English.
Sources: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
A bank's RAG system fails quietly
Picture a mid-sized bank building a retrieval-augmented generation system over its internal knowledge base. Thousands of documents are in scope: RBI regulatory circulars, policy manuals, Basel III guidance, loan-processing procedures, KYC records and operational playbooks. The project looks healthy. One engineer stands up the vector database, another writes the retrieval pipeline, a third connects the LLM and builds the interface. Then the system goes live.
Three weeks in, users start complaining about wrong answers, and the cause isn't hallucination. The retriever finds the right document, and the model answers faithfully from what it was given. The trouble began much earlier. One RBI circular on NBFC liquidity norms existed only as a scanned PDF, and during OCR the figure "10%" came out as "lO%", with the digit 1 swapped for a lowercase L. The parser accepted it without a check. The embedding model encoded the damaged text, the vector store saved it, the retriever ranked it near the top, and the LLM confidently told a compliance officer the required ratio was "lO percent."
Retrieval didn't fail and generation didn't fail. Every component did what it was built to do. The defect sat in the first stage of the pipeline, data ingestion, which is among the most common and costly blind spots in production RAG. Teams spend weeks tuning embeddings, chunking, rerankers and prompts while treating ingestion as plain preprocessing. Yet retrieval quality is settled well before any query arrives. If the source is parsed wrongly, no embedding model, index or LLM can bring the original facts back.
What follows covers ingestion for enterprise RAG: handling structured, semi-structured and unstructured sources, dealing with multimodal documents, designing for scale, spotting parse failures and wiring it together with working Python.
What ingestion does
In a RAG pipeline, ingestion takes raw source documents, whatever their format or origin, and turns them into clean, structured text ready to be chunked and embedded.
The stage lives wholly in the offline pipeline, running ahead of any user question. What it emits, parsed and cleaned text with metadata, is the input for every later step. Three jobs happen here. Loading fetches the raw document from its home, which might be a file system, an S3 bucket, a SharePoint site, a REST API or a database. Parsing extracts clean text from the raw format, whether by decoding a PDF, stripping HTML tags, deserializing JSON or running OCR on an image. Initial metadata extraction captures the source, creation date, document type and any format-specific details visible at parse time. Cleaning, chunking and embedding come afterward. Ingestion's sole duty is getting clean text out of raw material.
The messy reality of bank documents
A bank's knowledge base doesn't show up as tidy markdown. It's decades of accumulated formats: scanned PDFs from the 1990s, Word files with tracked changes, HTML pages from an internal compliance portal, JSON exports from a policy-management system, emails with attachments and SharePoint sites with deep folder trees. Every format needs its own parser, and every parser fails in its own way. Worse, parser failure is silent. A parser that produces garbled text doesn't raise an exception. It returns the garble and moves on.
PDFs: three kinds, three treatments
PDF dominates regulatory and compliance material, and it's also the hardest to parse. Three varieties need different handling:
- Text-layer PDFs come from word processors or print-to-PDF. The text sits in the content stream and can be extracted directly.
- Scanned PDFs are image files made from physical paper. There's no text layer at all, so OCR is mandatory. Older circulars and historical policies mostly fall here.
- Hybrid PDFs are scans that received an OCR text layer of poor quality. They're the riskiest, because a naive parser pulls the error-filled layer and assumes everything worked.
Telling them apart is the first step, and the classification function below inspects a PDF and reports which kind it is.
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
Scanned files need an OCR pipeline, and the chart compares the three realistic production options.
For a bank working through RBI circulars, Azure Document Intelligence is the practical pick. It copes with scans of mixed quality, understands table structure inside scanned pages and scores confidence per word, which lets low-confidence extractions be flagged for human review. The loader below calls it and applies that logic.
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
The needs_review flag matters in a regulated setting. A page with more than five low-confidence words shouldn't slide downstream unnoticed. It belongs in a human reviewer's queue, or at least carries a flag in its metadata so later evaluation can weight it accordingly.
Word documents, HTML and structured sources
Word files dominate internal policy and procedure writing. They carry headings, tables and lists that help retrieval later but vanish under a naive text extractor. The loader here walks the document and keeps that structure.
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
Section tracking pays off. In a KYC procedure with forty paragraphs, knowing that a chunk came from "Section 3: Enhanced Due Diligence" and not "Section 1: Standard KYC" is metadata that sharply improves retrieval precision.
Compliance portals are nearly always web apps, so HTML loading must skip navigation, ads, cookie banners and other boilerplate around the real content.
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,
}
Structured sources such as policy-management systems, case databases and transaction logs call for a different approach. Their text is already clean, so the questions are which fields to ingest and how to render rows as retrievable text.
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
Banks keep most working files in SharePoint or Google Workspace, and reaching them needs OAuth plus the relevant Graph or Drive APIs.
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")
Batch or streaming
Every ingestion design starts with one decision: process documents as they land, or in scheduled batches?
- Batch ingestion runs on a schedule, usually nightly or weekly. It fits when freshness isn't critical, volume is high and cost matters. A bank ingesting monthly RBI circulars doesn't need real time, and a nightly job is cheaper.
- Streaming ingestion works through an event queue as documents arrive. It suits cases where freshness is critical, such as a compliance system that must surface a new circular within minutes of publication.
External connectors need protection against rate limits, and the class below wraps requests with retries and backoff.
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")
That pattern isn't optional. Most external APIs, financial data providers included, will block an IP or revoke a key once the limit is exceeded. The backoff_factor gives exponential backoff, with waits of 2, 4 and 8 seconds on successive retries, which is the bare minimum retry policy for a production connector.
Tables, charts and images
The 2025 shift in production RAG moved from text-only pipelines to multimodal ones. Banking documents aren't purely textual. They hold tables of capital adequacy ratios, charts of NPA trends, diagrams of KYC workflows and images of signed agreements. Skipping non-text content discards a large share of each document's information, and in regulation that share is often the most valuable. The table in an RBI circular that states exact numeric thresholds is precisely what a compliance team needs.
A table can be handled in two ways: serialized as text, or extracted as structured data. The right choice depends on downstream use. The extraction function below pulls tables out of a 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
Serialized markdown suits tables an LLM will read and reason over, as in "what CET1 ratio does Basel III require?" Structured JSON suits values that will be queried in code or compared numerically.
Charts, diagrams and infographics can't be skipped either. A chart of NPA trend lines by quarter, or a flowchart of the FEMA transaction approval process, holds information found nowhere in the surrounding text. The common 2025 pattern sends each image to a vision-language model and stores the generated description for embedding and retrieval.
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
The prompt matters here. A generic "describe this image" yields generic output. A domain-specific instruction, asking for numeric values, trends, labels or process steps, yields descriptions that can be found. An engineer asking about NPA trend lines in the Q3 report needs a description containing "NPA," "trend" and the quarter, not one about a bar chart with blue bars.
Audio and video
Audio and video are the current frontier for banking RAG. Recordings of board meetings, analyst call transcripts, RBI press conferences and training videos hold material that RAG systems can't yet reach.
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,
}
Segment timestamps in the output are important. When a question is answered by something said 34 minutes into a board call, the system should cite that exact position and not just credit a board recording.
One router for every format
Separate loaders aren't enough. A production pipeline must send each document to the right parser, handle failures gracefully, track processing status and emit one consistent output format whatever went in.
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
The consistent output schema is the single most important design choice. The chunker, embedding generator and metadata enricher all receive the same dictionary shape, whether the input was a scanned circular, a DOCX loan policy or an audio transcript from a compliance course. The needs_review flag sends problem documents to a human queue and keeps errors from spreading.
Three failures behind most problems
Most retrieval-quality trouble in banking RAG traces back to three ingestion mistakes.
Ignoring document type at parse time. A text extractor run on a scanned PDF returns nothing, or a few stray characters. That file becomes empty chunks, near-zero vectors and index entries that retrieval then returns. The LLM sees empty context and invents an answer. The classify_pdf function prevents it with a handful of lines that save weeks of debugging.
Treating tables as body text. A risk-rating table flattened into prose loses its rows and columns, and a string like "AAA 0.01% AA 0.05% A 0.10%" can't be retrieved usefully. Serialized as a markdown table with headers, it can. Extract tables separately and keep their structure.
Missing retries on connectors. When a SharePoint connector hits a network timeout and says nothing, the corpus ends up with holes and nobody knows which documents are gone. The RateLimitedConnector pattern raises once retries run out and logs each attempt, so the gaps are visible.
Tool trade-offs
- PyMuPDF, pdfplumber and PyPDF2. PyMuPDF is the fastest extractor for text-native PDFs. pdfplumber is slower but far better at tables. PyPDF2 is oldest and least accurate, so avoid it in new pipelines. Use PyMuPDF for text and pdfplumber for tables from the same file.
- Self-hosted or cloud OCR. Tesseract is free, runs locally and handles clean scans well enough. For thousands of historic circulars of mixed quality, Azure Document Intelligence's accuracy edge justifies its per-page price. Misread regulatory thresholds aren't a cost worth saving on.
- Streaming cost. Streaming every upload in real time is the most expensive architecture, and a bank adding two or three documents a day has no need for it. Keep it for cases where freshness changes business outcomes, like a trading desk needing instant access to new SEBI circulars. An HR policy library isn't one of those.
- Multimodal cost. Sending every image in every document through GPT-4o adds real expense at scale. Filter out tiny images, such as decorative ones under 5KB, drop those whose aspect ratio marks them as logos or icons, and cache descriptions so an image repeated across documents gets described once.
Raw text isn't clean text
What ingestion delivers is raw text, not clean text. A separate cleaning stage handles noise removal, deduplication, OCR error correction, multilingual normalization and boilerplate stripping, sitting between ingestion and the chunker. The split matters because the bugs differ. Ingestion failures give empty or garbled content, while cleaning failures give content that looks right but carries subtle noise that hurts embedding quality, and that only shows up when a retrieval benchmark runs.
For production use, the code samples need logging, secrets management (keep API keys in environment variables or a secrets manager, never hardcoded) and latency instrumentation at each stage.