Загрузка даных для прадакшн-RAG у банку: PDF, OCR, табліцы і канектары
Як загружаць банкаўскія дакументы ў RAG: класіфікаваць PDF, запускаць OCR з пазнакамі для праверкі, захоўваць структуру Word, апрацоўваць табліцы і выявы і будаваць устойлівыя канектары.
Крыніцы: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
Сістэма RAG банку працюе без проблем
Уявіце сярэдні банк, які стварае сістэму падсіленай генеравання інфармацыі на адміністрацыйной базе знаёмых. У ўзглядзе знаходзяцца тыясячы дакументаў: рэгуляторныя цыркуляры RBI, інструкцыі па політыкі, рэкамендаціі Базеля III, процедуры апрацоўкі крэдытавання, дакументы KYC і методыкі працы. Праект выглядае адносна стабільным. Аднін інжынер стварае базу дадзеных у векторны фармат, іншы пішае алгорытм пошуку, трэці пад’яоўвае LLM і стварае інтэрфейс. Пасля чаго сістэма запускаецца у працу.
Чырвоўце дзён пасля запуску корыстуначыя пачалі скаржыцца на некарэктныя адпаведзі, і прычынаю была не галюцынэція. Сістэма пошуку знаходзіла правы дакумент, а мадэль адпавядала точна на адпаведны матэрыял. Проблемы пачаліся ўжо раней. Аднам з цыркуляраў RBI пра нормы ліквіднасці NBFC існаваў толькі у варыянцы сканаванага PDF, і пад час працэсу OCR цыфра „10%“ ператворылася на „lO%“, дзе цыфра 1 была заменена на малая літэра L. Сістэма парсінгу прыняла ўсё гэта без пераканальнага перагляду. Мадэль эмбедынгу закодавала паспаленый тэкст, хранільнік вектараў яго зафіксав, сістэма пошуку пасунула яго на першыя месцы, а LLM абавязкова сказала спеціялісту з дапамогі ў сарабатке, што неабходны пакэт станавіць „lO percent.“
Працэс адаптавання дакументаў не выклікаў проблем, як і працэс стварэння рэзультатаў. Кожны элемент сістэмы выканаў тое, для чаго быў створаны. Недакладнасць было ў першай стадзіі працы сістэмы — прыйманні дакументаў, якая ёсць аднай з найпашырэнейшых і найдорожэйшых прычын проблем у практычным вікорыстанні тэхналогій 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
Серыяваны маркдаун падходзіць для табелей, якія ШМ можа прачытаць і аналізаваць, напрыклад, у запытанні "Какі парадокс 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 у звярненні за трэці квартал, патрэбуе апісанні, якое мячыць слова „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 направляе документы з проблемамі у чаты для перагледу людзьмі і запобiegае распространэнню памылак.
Тры прычыны большасці проблем
Большасць прыбуткаў у якосці выкарыстоўвання дадзеных у банкавасці вынікае з трох памылак падчас ўваходжэння дадзеных.
Ігнараванне тыпу дакумента пад час парсавання. Адзінстрактар тэксту, які працюе з сканаваным 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, нормалізацыёй кантэнту для разных мов і адсутнечай зайвага тэксту, і ён распалоўваецца межы прыемкі і процэсам разбівання на часткі. Разбіўка мае значэнне, таму што багі разныя. Памілкі прыемкі даюць порожні чыста збэдзёрваны кантэнт, тады як памілкі чышчэння даюць кантэнт, які выглядае правільна, але мае тонкі шум, які паграджае якасць імбеддынгу, і гэты шум праявляецца толькі пад час адпаведных тэстаў.
Для практычнага выкарыстоўвання прыклады коду трэбуюць логавання, калектываванне секрэтных дадзей (хранення ключаў API ў змэнных сяродавысці або у спецыяльным калектары секрэтных дадзей, ніколі не ў фармаце жорсткага коду) і відстэжэння затрымакоў на кожным этапе.