Прийом даних RAG для банків: PDF, OCR, таблиці та з’єднувачі
Як імпортувати банківські документи для RAG: класифікувати PDF-файли, виконувати OCR із позначками для перевірки, зберігати структуру документа у форматі Word, обробляти таблиці та зображення, а також створювати надійні з’єднувачі.
Джерела: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
Система RAG банку працює без проблем
Уявіть собі середній банк, який створює систему покращеної генерації інформації на основі власної внутрішньої бази знань. У рамках проекту враховуються тисячі документів: нормативні циркуляри RBI, посібники з політики, рекомендації Basel III, процедури обробки кредитів, записи про перевірку клієнтів та інструкції з роботи. Проєкт виглядає успішним. Один інженер створює базу даних векторів, інший пише механізм пошуку, ще один під’єднує модель LLM та розробляє інтерфейс. Потім система запускається в експлуатацію.
Через три тижні користувачі почали скаржитися на неправильні відповіді, і причиною цього не були галюцинації. Система пошуку знаходила правильний документ, а модель вірно відповідала на основі отриманої інформації. Проблеми почалися значно раніше. Один з циркулярів RBI щодо норм ліквідності NBFC існував лише у вигляді сканованого PDF, і під час обробки OCR цифра „10%“ перетворилася на „lO%“ — цифра 1 була замінена на маленьку літеру L. Парсер прийняв цей текст без перевірки. Модель ембеддингу закодувала пошкоджений текст, система зберігання векторів зберегла його, система пошуку розмістила його майже на вершині списку, а LLM впевнено повідомила спеціаліста з дотримання правил, що необхідний показник становить „lO відсотків“.
Процес отримання даних не зазнав невдачі, як і процес генерації. Кожен компонент виконував те, для чого був створений. Проблема полягала на першому етапі конвеєра — прийманні даних, що є однією з найпоширеніших та найдорожчих проблем у продакшн-системах типу RAG. Команди витрачають тижні на налаштування моделей ембеддингів, розділення даних на частини, алгоритмів ранжування та запитів, водночас сприймаючи процес приймання даних як звичайну попередню обробку. Проте якість отримання даних визначається задовго до надходження будь-якого запиту. Якщо джерело дані оброблено неправильно, жодна модель ембеддингів, індекс чи ШІ не зможуть відновити первинні факти.
У наступному розглядається процес приймання даних у корпоративних системах RAG: робота зі структурованими, напівструктурованими та неструктурованими джерелами, обробка багатомодальних документів, проектування систем для масштабування, виявлення помилок у обробці даних та їх інтеграція за допомогою Python.
Що робить процес приймання даних
У підході RAG процес отримання даних передбачає прийом необроблених вихідних документів, незалежно від їхнього формату чи походження, та перетворення їх на чистий, структурований текст, готовий до поділу на частини та вбудовування.
Цей етап повністю відбувається у офлайн-процесі, ще до того, як з’явиться запит від користувача. Те, що він генерує — це оброблений та очищений текст із метаданими, який стає вхідними даними для кожного наступного кроку. Тут відбуваються три операції. Завантаження отримує необроблений документ з його джерела, яким може бути файлова система, контейнер S3, сайт SharePoint, REST API чи база даних. Обробка вилучає чистий текст із необробленого формату — шляхом декодування PDF, видалення тегів HTML, десеріалізації JSON чи виконання OCR на зображенні. Початкове вилучення метаданих фіксує джерело, дату створення, тип документа та будь-які деталі, специфічні для формату, які є видимими під час обробки. Після цього відбувається очищення, розділення на частини та вбудовування даних. Єдиною функцією цього етапу є отримання чистого тексту з необробленого матеріалу.
Складна реальність банківських документів
База знань банку не має вигляду охайного Markdown. Це десятиліття накопичених форматів: відскановані PDF з 1990-х років, файли Word із слідами змін, HTML-сторінки з внутрішнього порталу дотримання правил, експорти JSON з системи керування політиками, електронні листи з вкладеннями та сайти SharePoint із складними структурами папок. Кожен формат потребує власного парсера, і кожен парсер збивається по-своєму. Ще гірше те, що збій парсера залишається непоміченим. Парсер, який генерує спотворений текст, не створює винятку. Він просто повертає цей спотворений текст та продовжує роботу.
PDF: три види, три способи обробки
PDF домінує у регуляторних та контрольних матеріалах, і саме його найважче парсувати. Три різновиди вимагають різного підходу:
- PDF з текстовим шаром створюються за допомогою текстових процесорів або функцій „друкувати у PDF“. Текст знаходиться у потоці контенту та може бути вилучений безпосередньо.
- Скановані PDF — це файли зображень, створені з фізичного паперу. У них зовсім немає текстового шару, тому OCR є обов’язковим. До цієї категорії переважно належать старі циркуляри та історичні документи.
- Гібридні PDF — це сканування, до яких було додано текстовий шар OCR низької якості. Вони є найбільш ризикованими, оскільки примітивний парсер використовує цей шар, сповнений помилками, та вважає, що все працює коректно.
Відмінність між ними — це перший крок, а наведена нижче функція класифікації аналізує PDF та визначає, до якої категорії він належить.
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
Скановані файли потребують технології OCR, і ця діаграма порівнює три реалістичні варіанти виробництва.
Для банку, який працює з циркулярами RBI, Azure Document Intelligence є найкращим вибором. Він ефективно обробляє скани різної якості, розуміє структуру таблиць у сканованих сторінках та оцінює рівень достовірності кожного слова, що дозволяє позначати елементи з низьким рівнем достовірності для подальшого перегляду людиною. Наведений нижче загрузчик викликає цей інструмент та застосовує відповідну логіку.
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
Флаг needs_review має велике значення у регульованому середовищі. Сторінка з більш ніж п’ятьма словами низької достовірності не повинна проходити далі без уваги. Вона має потрапити до черги людського рецензента або принаймні мати відповідний флаг у своїх метаданих, щоб пізніша оцінка могла врахувати це належним чином.
Документи Word, HTML та структуровані джерела
Файли Word є основним засобом для написання внутрішніх правил та процедур. Вони містять заголовки, таблиці та списки, які допомагають у подальшому пошуку інформації, але зникають під дією примітивних інструментів вилучення тексту. Цей засіб аналізу пройшовся по документу та зберіг його структуру.
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
Відстеження розділів приносить користь. Під час процедури KYC, яка складається з сорока пунктів, знання того, що певна частина походить з „Розділу 3: Поглиблена перевірка“ а не з „Розділу 1: Стандартна процедура KYC“, є метаданими, які значно покращують точність пошуку.
Портали для дотримання норм майже завжди є веб-додатками, тому під час завантаження HTML необхідно пропускати навігацію, рекламу, банери з інформацією про куки та інші стандартні елементи, які знаходяться навколо справжнього контенту.
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,
}
Структуровані джерела, такі як системи керування політиками, бази даних справ та журнали транзакцій, вимагають іншого підходу. Їхній текст вже є охайним, тож питання полягають у тому, які поля потрібно включити та як представити рядки у вигляді доступного тексту.
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
Банки зберігають більшість робочих файлів у SharePoint або Google Workspace, і для їх доступу потрібні OAuth та відповідні API Graph чи 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")
Пакетна обробка чи потокова
Кожен дизайн процесу імпорту починається з одного рішення: обробляти документи в момент їх надходження чи у запланованих пакетах?
- Пакетна імпортна обробка відбувається за графіком, зазвичай щоночі або щотижня. Вона підходить тоді, коли свіжість даних не є критичною, обсяг великий та важливі витрати. Банк, який імпортує щомісячні циркуляри RBI, не потребує реального часу, а щонічна обробка є дешевшою.
- Потокова імпортна обробка працює через чергу подій у момент надходження документів. Вона підходить у випадках, коли свіжість даних є критичною, наприклад у системах контролю дотримання правил, які повинні відображати нові циркуляри протягом кількох хвилин після їх оприлюднення.
Зовнішні з’єднувачі потребують захисту від обмежень швидкості, а наведений нижче клас обгортає запити з можливістю повторних спроб та відстрочок.
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")
Ця схема є обов’язковою. Більшість зовнішніх API, включаючи постачальників фінансових даних, заблокують IP-адресу або скасують ключ, як тільки буде перевищений ліміт. Параметр backoff_factor забезпечує експоненційне затримування спроб: між послідовними спробами очікується 2, 4 та 8 секунд, що є мінімальною політикою повторних спроб для продакшн-з’єднувача.
Таблиці, діаграми та зображення
У 2025 році у продакшн-рішеннях типу RAG відбувся перехід від трубопроводів, що працюють лише з текстом, до багатомодальних. Банківські документи не є суто текстовими — вони містять таблиці коефіцієнтів достатності капіталу, діаграми тенденцій NPA, схеми процесів KYC та зображення підписаних угод. Ігнорування нетекстового контенту призводить до втрати значної частини інформації з кожного документа, а згідно з регуляціями саме ця частина часто є найціннішою. Таблиця в циркулярі RBI, яка вказує точні числові пороги, саме те, що потрібна команда з дотримання правил.
Таблицю можна обробляти двома способами: серіалізувати її у вигляді тексту або видобути як структуровані дані. Правильний вибір залежить від подальшого використання. Функція видобутку нижче витягує таблиці з 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
Серіалізований Markdown підходить для таблиць, які ШІ може читати та аналізувати, наприклад у запитанні «Який коефіцієнт CET1 вимагає Базель III?». Структурований JSON підходить для значень, які будуть запитуватися в коді або порівнюватися числово.
Графіки, діаграми та інфографіка також є важливими. Графік тенденцій NPA за кварталами чи схема процесу схвалення транзакцій FEMA містять інформацію, якої немає в оточуючому тексті. Загальноприйнята практика на 2025 рік полягає у надсиланні кожної зображення до моделі мови та зберіганні створеного опису для подальшого використання та пошуку.
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
Тут важливий формулювання запиту. Загальний запит на кшталт „опишіть це зображення“ дає загальний результат. Інструкція, спрямована на конкретну галузь та що вимагає числових значень, трендів, міток чи кроків процесу, дає описи, які можна знайти. Інженер, який цікавиться лініями трендів NPA у звіті за 3-й квартал, потребує опису, що містить „NPA“, „тренд“ та назву кварталу, а не опису стовпчастої діаграми з синіми стовпцями.
Аудіо та відео
Аудіо та відео є сучасним напрямком розвитку технологій RAG у банківській сфері. Записи зустрічей правління, транскрипції дзвінків аналітиків, прес-конференцій RBI та навчальні відео містять інформацію, до якої системи RAG поки що не можуть отримати доступ.
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,
}
Часові позначки сегментів у результаті є важливими. Коли на запитання відповідають слова, сказані на 34-й хвилині зустрічі, система повинна вказати саме цю позицію, а не просто посилатися на запис зустрічі.
Один маршрутизатор для кожного формату
Окремі завантажувачі недостатні. Система обробки має надсилати кожен документ до відповідного парсера, гідно справлятися з несправностями, відстежувати статус обробки та генерувати один послідовний формат результату, незалежно від вхідних даних.
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
Єдина найважливіша дизайнерська рішення — це послідовна схема виведення даних. Функції розділення на частини, генерація ембеддингів та поповнення метаданих отримують однакову структуру словника, незалежно від того, чи був вхідним даним сканований круглий документ, політика кредитування у форматі DOCX чи аудіозапис з курсу з дотримання правил. Прапорець needs_review направляє проблемні документи до черги для людського розгляду та запобігає поширенню помилок.
Три причини більшості проблем
Більшість проблем із якістю пошуку в банківських системах RAG пов’язані з трьома помилками під час обробки даних.
Ігнорування типу документа під час парсингу. Застосування інструменту для вилучення тексту до сканованого PDF не дає жодних результатів або повертає кілька випадкових символів. У результаті такий файл стає набором порожніх фрагментів, майже нульових векторів та записів у індексі, які потім повертаються під час пошуку. Штучний інтелект бачить порожній контекст та вигадує відповідь. Функція classify_pdf запобігає цьому за допомогою кількох рядків коду, що економить тижні на виправленні помилок.
Обробка таблиць як тексту тіла. Таблиця з оцінкою ризиків, перетворена на прозу, втрачає свої рядки та стовпці, а рядок на кшталт "AAA 0.01% AA 0.05% A 0.10%" не може бути корисно відновлений. Однак якщо його серіалізувати у вигляді таблиці Markdown із заголовками, це стає можливим. Варто вилучати таблиці окремо та зберігати їхню структуру.
Відсутнє автоматичне повторення спроб підключення. Коли коннектор SharePoint досягає тайм-ауту мережі та не надсилає жодних повідомлень, корпус даних стає неповним, і ніхто не знає, які саме документи зникли. Модель RateLimitedConnector спрацьовує після вичерпання кількості спроб та фіксує кожну з них, тому прогалини у даних стають видимими.
Компроміси між інструментами
- PyMuPDF, pdfplumber та PyPDF2. PyMuPDF — найшвидший інструмент для вилучення тексту з PDF-файлів. pdfplumber працює повільніше, але значно краще справляється з таблицями. PyPDF2 є найстарішим та найменш точним інструментом, тому його слід уникати у нових потоках обробки даних. Використовуйте PyMuPDF для тексту та pdfplumber для таблиць у одному й тому ж файлі.
- Самостійне розміщення чи хмарний OCR. Tesseract є безкоштовним, працює локально та досить добре справляється з якісними сканами. Для тисяч історичних циркулярів різної якості перевага у точності Azure Document Intelligence виправдовує його ціну за сторінку. Неправильне розпізнавання нормативних показників — це не та витрата, яку варто економити.
- Вартість стрімування. Стрімування кожного завантаженого файлу в реальному часі — це найдорожча архітектура, і банку, який додає два-три документи на день, вона не потрібна. Використовуйте її лише у випадках, коли свіжість даних впливає на бізнес-результати, наприклад, коли торговельний відділ потребує миттєвого доступу до нових циркулярів SEBI. Бібліотека політик HR не належить до таких випадків.
- Багатомодальні витрати. Надсилання кожної зображення в кожному документі через GPT-4o спричиняє суттєві витрати у масштабному використанні. Необхідно видаляти дуже маленькі зображення, наприклад декоративні, розміром менше 5 КБ, а також ті, чий співвідношення сторін вказує на те, що це логотипи чи іконки. Крім того, слід кешувати описи, щоб зображення, яке повторюється в різних документах, описувалося лише один раз.
Необроблений текст — це не чистий текст
Процес отримання даних передбачає надходження необробленого тексту, а не чистого. Окремий етап очищення відповідає за усунення шуму, дедуплікацію, корекцію помилок OCR, нормалізацію тексту кількома мовами та видалення зайвого форматування; він розташований між отриманням даних та їх поділом на частини. Цей етап має велике значення, оскільки проблеми можуть відрізнятися. Невдачі під час отримання даних призводять до порожнього або спотвореного контенту, тоді як невдачі під час очищення дають контент, який виглядає правильно, але містить прихований шум, що погіршує якість вбудовування, і це проявляється лише під час тестування системи пошуку.
Для використання в промислових умовах зразки коду потребують журналізації подій, керування конфіденційними даними (зберігання API-ключів у змінних середовища або у спеціалізованих менеджерах секретів, ніколи не в коді) та інструментів для вимірювання затримок на кожному етапі.