Получение данных RAG для банков: PDF, OCR, таблицы и соединители
Как загружать банковские документы в систему RAG: классификация PDF-файлов, выполнение OCR с флагами для проверки, сохранение структуры документов в формате Word, обработка таблиц и изображений, а также создание надежных соединителей.
Источники: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
Система RAG банка работает без сбоев
Представьте себе средний банк, создающий систему усиленной генерации на основе поиска над своей внутренней базой знаний. В рамках проекта учитываются тысячи документов: регуляторные циркуляры RBI, руководства по политике, рекомендации Базельского соглашения III, процедуры обработки кредитов, записи проверки личности и руководства по операционной деятельности. Проект выглядит успешным: один инженер разрабатывает векторную базу данных, другой пишет механизм поиска, третий подключает большую языковую модель и создаёт интерфейс. Затем система запускается в эксплуатацию.
Через три недели пользователи начали жаловаться на неверные ответы, причем причиной были не галлюцинации. Система поиска находила правильный документ, а модель точно отвечала на основе полученной информации. Проблемы начались гораздо раньше. Один из циркуляров RBI, касающихся норм ликвидности NBFC, существовал только в виде отсканированного PDF-файла, и в процессе OCR цифра «10%» превратилась в «lO%» — цифра 1 была заменена на строчную букву L. Парсер принял этот текст без проверки. Модель эмбеддингов кодировала повреждённый текст, хранилище векторов сохраняло его, система поиска выдвигала его в верхнюю часть результатов, а большая языковая модель уверенно сообщала сотруднику, отвечающему за соблюдение норм, что требуемый показатель составляет «lO процентов».
Поиск не сработал с ошибкой, и генерация тоже не сработала с ошибкой. Каждый компонент выполнил свою функцию. Недостаток заключался на первом этапе пайплайна — принятии данных, что является одной из самых распространённых и дорогостоящих проблем в производственных системах RAG. Команды тратят недели на настройку моделей эмбеддингов, разбиения данных на части, алгоритмов ранжирования и промптов, считая процесс принятия данных простой предварительной обработкой. Однако качество поиска определяется задолго до поступления любого запроса. Если исходные данные обработаны неправильно, ни одна модель эмбеддингов, индекс или большая языковая модель не смогут восстановить первоначальные факты.
Далее рассматривается процесс принятия данных в корпоративных системах RAG: обработка структурированных, полуструктурированных и неструктурированных источников, работа с мультимодальными документами, проектирование систем с учётом масштабируемости, выявление ошибок в парсинге и их устранение с использованием Python.
Что делает процесс принятия данных
В пайплайне RAG процесс загрузки берет сырые исходные документы в любом формате и из любого источника, преобразуя их в чистый, структурированный текст, готовый к разбиению на фрагменты и встраиванию.
Этот этап полностью происходит в оффлайн-пайплайне и запускается до поступления любого вопроса от пользователя. Результатом его работы является обработанный и очищенный текст с метаданными, который служит входными данными для всех последующих шагов. Здесь выполняются три операции. Загрузка получает исходный документ из источника — это может быть файловая система, контейнер S3, сайт SharePoint, REST API или база данных. Обработка извлекает чистый текст из исходного формата, будь то декодирование PDF, удаление тегов HTML, десериализация JSON или распознавание символов на изображении. Первоначальный извлечение метаданных фиксирует источник, дату создания, тип документа и любые детали, связанные с форматом, видимые во время обработки. После этого происходят очистка, разбиение на части и встраивание данных. Единственная задача этапа обработки — получить чистый текст из исходного материала.
Сложная реальность банковских документов
База знаний банка не представлена в виде аккуратного формата Markdown. Это десятилетия накопленных форматов: отсканированные PDF-файлы 1990-х годов, файлы Word с отслеживанием изменений, HTML-страницы из внутреннего портала по вопросам соблюдения нормативов, экспортные данные в формате JSON из системы управления политиками, электронные письма с вложениями и сайты SharePoint с сложной структурой папок. Каждый формат требует собственного парсера, и каждый парсер сбоит по-своему. Что ещё хуже, сбои парсеров происходят бесшумно. Парсер, генерирующий искажённый текст, не вызывает исключений — он просто возвращает искажённый текст и продолжает работу.
PDF-файлы: три вида, три способа обработки
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-ключей в переменных окружения или специальных менеджерах секретов, никогда не в виде хардкодированных значений) и измерения задержек на каждом этапе.