Przyjmowanie danych RAG do produkcji dla banków: pliki PDF, OCR, tabele i łączniki
Jak wprowadzać dokumenty bankowe do systemu RAG: klasyfikować pliki PDF, uruchamiać OCR z flagami do sprawdzenia, zachowywać strukturę plików Word, obsługiwać tabele i obrazy oraz tworzyć odpornych łączników.
Źródła: Data Ingestion for Production RAG: Building Reliable Enterprise Data Pipelines
System RAG banku przestaje działać bez żadnych objawów
Załóżmy, że średniej wielkości bank buduje system generowania treści wzbogaconego o możliwości wyszukiwania na podstawie swojej wewnętrznej bazy wiedzy. W zakres prac wchodzą tysiące dokumentów: obwieszczenia regulacyjne RBI, podręczniki polityk, wytyczne Basel III, procedury przetwarzania pożyczek, zapisy KYC oraz instrukcje operacyjne. Projekt wygląda obiecująco – jeden inżynier tworzy bazę wektorową, drugi opracowuje proces wyszukiwania, a trzeci łączy model językowy z interfejsem. Następnie system zostaje uruchomiony.
Po trzech tygodniach użytkownicy zaczynają skarżyć się na błędne odpowiedzi, a przyczyną nie są halucynacje. System wyszukiwania znajduje właściwy dokument, a model odpowiada wiernie na podstawie tego, co otrzymał. Problemy zaczęły się znacznie wcześniej. Jeden z obwieszczeń RBI dotyczących norm płynności w NBFC istniał jedynie w formie zeskanowanego pliku PDF, a podczas procesu OCR liczba „10%” została przetłumaczona jako „lO%”, przy czym cyfra 1 została zastąpiona małą literą L. Parser przyjął to bez żadnej weryfikacji. Model embeddingowy zakodował uszkodzony tekst, magazyn wektorów go zapisał, system wyszukiwania umieścił go w czołówce, a model LLM z pewnością siebie poinformował pracownika ds. zgodności, że wymagany wskaźnik to „lO percent.”
Pobieranie danych nie zawiodło, podobnie jak generowanie. Każdy komponent wykonał to, do czego został stworzony. Wada znajdowała się na pierwszym etapie procesu, przyjmowania danych, który należy do najczęstszych i najdroższych problemów w systemach RAG w produkcji. Zespoły spędzają tygodnie na optymalizacji modeli embeddingów, dzieleniu danych na fragmenty, narzędzi do ponownego rankowania oraz promptów, traktując przy tym proces przyjmowania danych jako zwykłe przetwarzanie wstępne. Tymczasem jakość pobierania danych jest ustalana na długo przed nadejściem jakiegokolwiek zapytania. Jeśli źródło danych zostanie błędnie przetworzone, żaden model embeddingów, indeks ani LLM nie będzie w stanie przywrócić oryginalnych faktów.
Następne rozdziały omawiają proces przyjmowania danych w systemach RAG dla przedsiębiorstw: obsługę źródeł strukturalnych, półstrukturalnych i niestrukturalnych, radzenie sobie z dokumentami multimodalnymi, projektowanie systemów skalowalnych, wykrywanie błędów w przetwarzaniu oraz łączenie tych elementów za pomocą Pythona.
Jakie zadania pełni proces przyjmowania danych
W pipeline RAG proces pobierania obejmuje surowe dokumenty źródłowe, niezależnie od ich formatu czy pochodzenia, i przekształca je w czysty, ustrukturyzowany tekst gotowy do podziału na fragmenty i włączenia.
Proces ten odbywa się wyłącznie w pipeline’ie offline, zanim pojawi się jakiekolwiek pytanie użytkownika. To, co jest wytwarzane – tekst po przetworzeniu i oczyszczeniu wraz z metadanymi – stanowi dane wejściowe dla każdego kolejnego kroku. Tutaj odbywają się trzy zadania. Ładowanie pobiera surowy dokument z jego źródła, którym może być system plików, pojemnik S3, strona SharePoint, REST API lub baza danych. Przetwarzanie wydobywa czysty tekst z surowego formatu, czy to poprzez dekodowanie pliku PDF, usuwanie tagów HTML, deserializację danych JSON lub wykonywanie OCR na obrazie. Pierwotne wyodrębnianie metadanych obejmuje zapisanie źródła, daty utworzenia, typu dokumentu oraz wszelkich szczegółów specyficznych dla formatu widocznych w momencie przetwarzania. Dopiero potem następuje oczyszczanie, dzielenie na fragmenty i włączanie metadanych do tekstu. Jedynym zadaniem tego procesu jest uzyskanie czystego tekstu z surowych danych.
Skomplikowana rzeczywistość dokumentów bankowych
Baza wiedzy banku nie występuje w postaci uporządkowanego markdowna. To dziesięciolecia gromadzonych formatów: skanowane pliki PDF z lat 90., pliki Word z zapisem zmian, strony HTML z wewnętrznego portalu ds. zgodności, eksporty JSON z systemu zarządzania politykami, e-maile z załącznikami oraz strony SharePoint z rozbudowanymi drzewami folderów. Każdy format wymaga własnego parsera, a każdy z nich zawodzi na swój sposób. Co gorsza, awaria parsera przebiega bez żadnych objawów. Parser, który generuje zniekształcony tekst, nie wywołuje wyjątku – po prostu zwraca ten zniekształcony tekst i przechodzi dalej.
Pliki PDF: trzy rodzaje, trzy sposoby obsługi
PDF dominuje w materiałach regulacyjnych i związanych z zgodnością, a jednocześnie jest najtrudniejszy do parsowania. Trzy jego odmiany wymagają różnego podejścia:
- Pliki PDF z warstwą tekstową pochodzą z programów do edycji tekstu lub funkcji drukowania na PDF. Tekst znajduje się w strumieniu treści i można go bezpośrednio wyekstrahować.
- Skanowane pliki PDF to pliki obrazowe utworzone z fizycznego papieru. W ogóle nie ma warstwy tekstowej, dlatego OCR jest obowiązkowe. Do tej kategorii należą głównie starsze obwieszczenia i polityki historyczne.
- Hybrydowe pliki PDF to skany, do których dodano warstwę tekstową OCR niskiej jakości. Są one najbardziej ryzykowne, ponieważ prosty parser wykorzystuje tę pełną błędów warstwę i zakłada, że wszystko działa poprawnie.
Rozróżnienie ich jest pierwszym krokiem, a poniższa funkcja klasyfikacji analizuje plik PDF i podaje, do której kategorii należy.
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
Zeskanowane pliki wymagają procesu OCR, a wykres porównuje trzy realistyczne opcje produkcji.
Dla banku, który musi postępować zgodnie z wytycznymi RBI, Azure Document Intelligence jest praktycznym rozwiązaniem. Radzi sobie ze zrzutami o różnej jakości, rozumie strukturę tabel w zeskanowanych stronach oraz ocenia poziom pewności dla każdego słowa, co umożliwia oznaczenie fragmentów o niskiej pewności do dalszej weryfikacji przez człowieka. Poniższy loader wywołuje tę usługę i stosuje tę logikę.
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
Znak needs_review ma znaczenie w środowisku regulowanym. Strona zawierająca ponad pięć słów o niskiej wiarygodności nie powinna trafić dalej bez żadnej uwagi. Powinna znaleźć się w kolejce do przeglądu przez człowieka lub przynajmniej mieć flagę w swoich metadanych, aby późniejsza ocena mogła ją odpowiednio uwzględnić.
Dokumenty Word, HTML i źródła strukturyzowane
Pliki Word dominują przy tworzeniu wewnętrznych polityk i procedur. Zawierają nagłówki, tabele i listy, które ułatwiają późniejsze wyszukiwanie informacji, ale znikają przy użyciu prostego narzędzia do wydobywania tekstu. Ten loader przegląda dokument i zachowuje jego strukturę.
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
Śledzenie sekcji przynosi korzyści. W procedurze KYC składającej się z czterdziestu akapitów wiedza o tym, że dany fragment pochodzi z „Sekcji 3: Wzmożona weryfikacja” a nie z „Sekcji 1: Standardowe KYC”, stanowi metadane, które znacznie poprawiają dokładność wyszukiwania.
Portale związane z przestrzeganiem regulacji to niemal zawsze aplikacje internetowe, dlatego ładowanie HTML musi pomijać menu nawigacyjne, reklamy, banery z informacjami o plikach cookie oraz inne elementy standardowe znajdujące się wokół rzeczywistej treści.
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,
}
Zastrukturyzowane źródła, takie jak systemy zarządzania politykami, bazy danych przypadków oraz dzienniki transakcji, wymagają innego podejścia. Ich tekst jest już uporządkowany, więc pytania dotyczą tego, które pola należy przyjąć oraz w jaki sposób przedstawić wiersze jako tekst dostępny do wyświetlenia.
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
Banki przechowują większość plików roboczych w SharePoint lub Google Workspace, a dostęp do nich wymaga OAuth oraz odpowiednich API Graph lub 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")
Przetwarzanie partiami czy w strumieniu
Każdy projekt przetwarzania danych zaczyna się od jednej decyzji: czy przetwarzać dokumenty w momencie ich przybycia, czy w zaplanowanych partiach?
- Przetwarzanie partiami odbywa się według harmonogramu, zwykle co noc lub co tydzień. Jest odpowiednie w sytuacjach, gdy aktualność nie jest kluczowa, a objętość danych duża i koszty mają znaczenie. Bank przetwarzający miesięczne obwieszczenia RBI nie potrzebuje danych w czasie rzeczywistym, a cotygodniowe zadanie jest tańsze.
- Przetwarzanie w strumieniu polega na przetwarzaniu dokumentów w miarę ich przybywania za pośrednictwem kolejki zdarzeń. Nadaje się do przypadków, gdy aktualność jest kluczowa, np. w systemach zgodności, które muszą pokazać nowe obwieszczenia w ciągu kilku minut od ich publikacji.
Konektory zewnętrzne wymagają ochrony przed ograniczeniami szybkości, a poniższa klasa otacza żądania mechanizmami ponawiania prób i opóźnień.
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")
To rozwiązanie nie jest opcjonalne. Większość zewnętrznych API, w tym dostawcy danych finansowych, zablokuje adres IP lub cofnie klucz po przekroczeniu limitu. backoff_factor umożliwia eksponencjalne opóźnianie prób – kolejne próby odbywają się po 2, 4 i 8 sekundach – co stanowi minimalną politykę ponawiania prób dla połączeń produkcyjnych.
Tabele, wykresy i obrazy
W 2025 roku w rozwiązaniach RAG produkcyjnych nastąpiła zmiana z pipeline’ów opartych wyłącznie na tekście na te wielomodalne. Dokumenty bankowe nie są czysto tekstowe – zawierają one tabele wskaźników kapitałowych, wykresy trendów NPA, schematy procesów KYC oraz obrazy podpisanych umów. Pomijanie treści nie tekstowych oznacza utratę dużej części informacji zawartych w każdym dokumencie, a zgodnie z przepisami ta część jest często najcenniejsza. Tabela w obwieszczeniu RBI, która podaje dokładne wartości liczbowe progów, jest dokładnie tym, czego potrzebuje zespół ds. zgodności.
Tabelę można przetwarzać na dwa sposoby: jako tekst zserwowany lub jako dane strukturalne. Odpowiedni wybór zależy od dalszego zastosowania. Poniższa funkcja wydobywania pobiera tabele z pliku 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
Serializowany markdown nadaje się do tabel, które model językowy może przeczytać i przeanalizować, na przykład w pytaniu „Jaki wskaźnik CET1 wymaga Basel III?”. Strukturyzowany JSON jest odpowiedni dla wartości, które będą wyszukiwane w kodzie lub porównywane liczbowo.
Rysunki, diagramy i infografiki również nie mogą zostać pominęte. Wykres trendów NPA w poszczególnych kwartałach lub schemat procesu zatwierdzania transakcji FEMA zawierają informacje, których nie znajdziemy w otaczającym tekście. Powszechny model z 2025 roku polega na wysyłaniu każdego obrazu do modelu języka wizualnego oraz przechowywaniu wygenerowanego opisu w celu dalszego wykorzystania i wyszukiwania.
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
Tutaj kluczowe jest sformułowanie zapytania. Ogólne polecenie typu „opisz ten obraz” daje ogólne wyniki. Instrukcja specyficzna dla danej dziedziny, wymagająca wartości liczbowych, trendów, etykiet lub kroków procesu, przynosi opisy, które można znaleźć. Inżynier pytający o linie trendów NPA w raporcie za III kwartał potrzebuje opisu zawierającego słowa „NPA”, „trend” oraz nazwę kwartału, a nie opisu dotyczącego wykresu słupkowego z niebieskimi słupkami.
Audio i wideo
Audio i wideo stanowią obecnie nowe granice w zastosowaniach RAG w bankowości. Nagrania ze spotkań zarządu, transkrypcje rozmów z analitykami, konferencje prasowe RBI oraz filmy szkoleniowe zawierają materiały, do których systemy RAG jeszcze nie mają dostępu.
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,
}
Czasowe oznaczenia segmentów w wyniku są istotne. Gdy na pytanie odpowiada coś powiedzianego po 34 minutach od rozpoczęcia spotkania, system powinien wskazać dokładną tę pozycję, a nie jedynie odnieść się do nagrania spotkania.
Jeden router na każdy format
Odrębne narzędzia do ładowania nie wystarczą. Proces produkcji musi wysyłać każdy dokument do odpowiedniego parsera, sprawnie radzić sobie z błędami, śledzić stan przetwarzania i generować jeden spójny format wyniku, niezależnie od tego, co zostało wprowadzone.
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
Jednolity schemat wyjściowy jest najważniejszą decyzją projektową. Narzędzie dzielące na fragmenty, generator embeddingów oraz narzędzie wzbogacania metadanych otrzymują zawsze tę samą strukturę słownika, niezależnie od tego, czy danymi wejściowymi była zeskanowana karta kredytowa, plik DOCX z warunkami pożyczki, czy transkrypcja audio z kursu szkoleniowego. Flaga needs_review kieruje problematyczne dokumenty do kolejki do przeglądu przez człowieka i zapobiega rozprzestrzenianiu się błędów.
Trzy przyczyny większości problemów
Większość problemów z jakością wyszukiwania w systemach RAG bankowych wynika z trzech błędów podczas przetwarzania danych.
Ignowowanie typu dokumentu podczas analizy. Narzędzie do wydobywania tekstu uruchomione na zeskanowanym pliku PDF nie zwraca żadnych wyników lub tylko kilku przypadkowych znaków. Taki plik staje się zbiorem pustych fragmentów, wektorów bliskich zeru oraz wpisów indeksu, które następnie są zwracane. Sztuczna inteligencja widzi pusty kontekst i wymyśla odpowiedź. Funkcja classify_pdf zapobiega temu za pomocą zaledwie kilku linijek kodu, co oszczędza tygodnie pracy nad debugowaniem.
Traktowanie tabel jako tekstu ciała. Tabela oceny ryzyka przekształcona w prozę traci swoje wiersze i kolumny, a ciąg znaków taki jak „AAA 0.01% AA 0.05% A 0.10%” nie może być użytecznie odzyskany. Gdy jest zserializowana jako tabela w formacie Markdown z nagłówkami, można to zrobić. Należy wydobywać tabele oddzielnie i zachować ich strukturę.
Gdy connector do SharePoint doświadcza przeciążenia sieciowego i nie reaguje, korpus zawiera luki, a nikt nie wie, które dokumenty zniknęły. Wzorzec RateLimitedConnector uruchamia się po wyczerpaniu możliwości powtórzeń i rejestruje każdą próbę, dzięki czemu luki stają się widoczne.
Kompromisy między narzędziami
- PyMuPDF, pdfplumber i PyPDF2. PyMuPDF to najszybszy narzędzie do wydobywania tekstu z plików PDF. pdfplumber jest wolniejszy, ale znacznie lepszy do obsługi tabel. PyPDF2 jest najstarszy i najmniej dokładny, dlatego należy unikać go w nowych procesach przetwarzania. Użyj PyMuPDF do tekstu, a pdfplumber do tabel w tym samym pliku.
- Samodzielnie zarządzany lub chmurowy OCR. Tesseract jest darmowy, działa lokalnie i dobrze radzi sobie z czystymi skanami. W przypadku tysięcy historycznych obwieszczeń o zróżnicowanej jakości przewaga dokładności Azure Document Intelligence usprawiedliwia jego cenę za stronę. Błędne odczytanie progów regulacyjnych to koszt, którego nie warto oszczędzać.
- Koszt transmisji strumieniowej. Transmisja każdego załadowanego pliku w czasie rzeczywistym to najdroższa architektura, a bank przesyłający dwa lub trzy dokumenty dziennie nie ma takiej potrzeby. Należy ją stosować tylko w sytuacjach, gdy aktualność ma wpływ na wyniki biznesowe, np. gdy dział handlowy potrzebuje natychmiastowego dostępu do nowych obwieszczeń SEBI. Biblioteka polityk HR nie należy do takich przypadków.
- Koszt multimodalny. Wysyłanie każdego obrazka z każdego dokumentu za pośrednictwem GPT-4o powoduje realne wydatki w dużych skali. Należy usunąć małe obrazki, takie jak te dekoracyjne o rozmiarze poniżej 5 KB, odrzucić te, których stosunek boków wskazuje na to, że są to logotypy lub ikony, oraz zapisać opisy w pamięci tymczasowej, aby obrazek powtarzający się w różnych dokumentach był opisywany tylko raz.
Tekst surowy to nie tekst czysty
Proces pobierania dostarcza tekst surowy, a nie tekst czysty. Odrębna faza oczyszczania zajmuje się usuwaniem szumów, eliminacją duplikatów, korektą błędów OCR, normalizacją w językach obcych oraz usuwaniem elementów standardowych, i znajduje się pomiędzy procesem pobierania a narzędziem dzielenia na fragmenty. To rozróżnienie jest istotne, ponieważ błędy się różnią. Błędy w procesie pobierania powodują puste lub zniekształcone treści, natomiast błędy w oczyszczaniu dają treści, które wyglądają poprawnie, ale zawierają subtelne szumy, które pogarszają jakość embeddingów i ujawniają się tylko podczas testów wyszukiwania.
Dla użycia w produkcji przykłady kodu wymagają logowania, zarządzania sekretami (przechowywanie kluczy API w zmiennych środowiskowych lub w narzędziu do zarządzania sekretami, nigdy w formie hardkodowanej) oraz pomiaru opóźnień na każdym etapie.
Literatura pokrewna
- Czyszczenie danych dla RAG w produkcji: kodowanie, OCR, usuwanie duplikatów i ochrona danych osobowych — Jak przekształcić zinterpretowane dokumenty korporacyjne w tekst gotowy do wyszukiwania: naprawa kodowania, usuwanie szablonów i duplikatów, flagi wskazujące na ryzykowne operacje OCR, obsługa języków oraz maskowanie danych osobowych.
- Strategie dzielenia na fragmenty w RAG produkcyjnym: o ustalonej wielkości, semantyczne i rodzic-dziecko — Praktyczny przegląd procesu dzielenia na fragmenty w RAG, od podziału o ustalonej wielkości i rekurencyjnego dzielenia po metody rodzic-dziecko, propozycje, późne dzielenie na fragmenty, RAPTOR, tabele, kod oraz testowanie wydajności.
- Hybrydowy RAG dla tekstów prawniczych: dzielenie na fragmenty, BM25, BGE i ponowna klasyfikacja — Proces RAG w dziedzinie prawa oparty na Konstytucji Indii: dzielenie na fragmenty rodzic-dziecko, wyszukiwanie za pomocą BM25 w połączeniu z BGE, fuzja rankingów, ponowna klasyfikacja za pomocą cross-encodera oraz rozszerzanie zapytań, przetestowane na benchmarkach.
- Wybór fragmentów bez zapytań: wykorzystanie metody K-Means, gdy RAG nie ma żadnego pytania do zadania — Gdy nie ma zapytania użytkownika, RAG nie może wybrać odpowiedniego kontekstu. Użyj liczby K jako określenia liczby wyników, zgrupuj wektory fragmentów, wybierz te o środkowym położeniu i zachowaj informacje o zakresie oraz cytatach źródłowych.
- Projektowanie RAG dla 100 milionów dokumentów: shardowanie, wyszukiwanie hybrydowe i koszty — Co zmienia się, gdy system RAG rośnie do miliardów fragmentów: asynchroniczne pobieranie danych, kierowanie ruchem, shardowanie, hybrydowe odzyskiwanie informacji, ponowna sortacja wyników, autoryzacja, cacheowanie oraz kontrola kosztów.
- Trzy przyczyny niepowodzeń systemów RAG w produkcji: fragmentaryczny kontekst, dewiacje i niewłaściwe użycie — Dlaczego systemy RAG pomijają subtelności, tracą jakość w miarę rozwoju baz wiedzy lub udzielają błędnych odpowiedzi pomimo skutecznego wyszukiwania, wraz z praktycznymi rozwiązaniami na poziomie warstwy wyszukiwania.
- Notatki praktyczne: MCP 2026: Protokół dorósł – o tym, jak wygląda sztuczna inteligencja w produkcji — Szczegółowy przewodnik po notatkach praktycznych: MCP 2026: Protokół dorósł – o tym, jak wygląda sztuczna inteligencja w produkcji: umowy, sprawdzania oraz gotowe elementy kodu dla zespołów wdrażających ten model.
- Inżynieria płaszczyzny kontroli oceny dla agentów AI w produkcji — Jak budować ciągłą ocenę agentów: oddzielenie płaszczyzn wykonawczych i kontrolnych, twierdzenia oparte na śledzeniu, złożone zestawy danych, bramy CI, testy całego cyklu życia oraz lista kontrolna przed uruchomieniem.