Produktionsbasierte Aufnahme von RAG-Daten für Banken: PDFs, OCR, Tabellen und Anbindungen

Das RAG-System einer Bank versagt leise

Stellen Sie sich eine mittelgroße Bank vor, die über ihre interne Wissensdatenbank ein retrieval-augmented generation System entwickelt. Tausende von Dokumenten fallen in den Anwendungsbereich: Regulierungsverordnungen der RBI, Richtlinienhandbücher, Leitfäden zu Basel III, Verfahren zur Kreditverarbeitung, KYC-Dokumente sowie Betriebsanleitungen. Das Projekt scheint gut voranzukommen. Ein Ingenieur erstellt die Vektordatenbank, ein anderer entwickelt den Abrufprozess, ein Dritter verbindet das LLM und erstellt die Schnittstelle. Anschließend wird das System in Betrieb genommen.

Drei Wochen später beginnen die Nutzer, sich über falsche Antworten zu beschweren – und der Grund liegt nicht in Halluzinationen. Der Suchmechanismus findet das richtige Dokument, und das Modell antwortet zuverlässig auf der Grundlage dessen, was ihm gegeben wurde. Das Problem begann jedoch schon viel früher. Ein RBI-Rundschreiben zu den Liquiditätsvorgaben für NBFCs existierte nur als gescanntes PDF, und bei der OCR-Verarbeitung wurde die Zahl „10%“ als „lO%“ ausgegeben, wobei die Ziffer 1 durch ein kleines L ersetzt wurde. Der Parser akzeptierte dies ohne Überprüfung. Das Embedding-Modell kodierte den fehlerhaften Text, der Vektor-Speicher speicherte ihn, der Suchmechanismus rangierte ihn an die Spitze – und das LLM teilte einem Compliance-Beauftragten selbstbewusst mit, dass der erforderliche Anteil „lO Prozent“ betrage.

Die Datenerfassung ist nicht fehlgeschlagen und auch die Generierung nicht. Jeder Komponente hat seine Aufgabe erfüllt. Der Fehler lag in der ersten Stufe des Pipelines, der Datenerfassung, die zu den häufigsten und kostspieligsten Schwachstellen in produktiven RAG-Systemen gehört. Teams verbringen Wochen damit, Embedding-Modelle, Aufteilung in Blöcke, Neurankierungsalgorithmen und Prompte zu optimieren, während sie die Datenerfassung als einfache Vorbereitungsphase betrachten. Doch die Qualität der Datenerfassung wird bereits lange vor dem Eintreffen einer Anfrage festgelegt. Wenn die Quelldaten falsch analysiert werden, kann kein Embedding-Modell, Index oder LLM die ursprünglichen Fakten wiederherstellen.

Im Folgenden wird die Datenerfassung für enterprise-kompatible RAG-Systeme behandelt: Umgang mit strukturierten, halbstrukturierten und unstrukturierten Quellen, Handhabung multimodaler Dokumente, Skalierbarkeitskonzepte, Erkennung von Analysefehlern sowie die Integration dieser Elemente mithilfe funktionsfähigen Python-Code.

Was die Datenerfassung leistet

In einem RAG-Pipeline-Prozess werden rohe Queldokumente, unabhängig von ihrem Format oder ihrer Herkunft, aufgenommen und in sauberen, strukturierten Text umgewandelt, der zur Aufteilung in Blöcke und Einbettung bereit ist.

Die Verarbeitung findet ausschließlich im Offline-Pipeline-Verfahren statt und läuft bereits vor jeder Anfrage des Benutzers. Das Ergebnis sind strukturierte, bereinigte Texte mit Metadaten, die als Eingang für alle nachfolgenden Schritte dienen. Hier finden drei Aufgaben statt: Das Laden holt das Rohdokument aus seiner Quelle ab – dies kann ein Dateisystem, ein S3-Container, eine SharePoint-Seite, eine REST-API oder eine Datenbank sein. Die Analyse extrahiert den sauberen Text aus dem Rohformat, sei es durch Dekodieren eines PDFs, Entfernen von HTML-Tags, Deserialisieren von JSON oder Durchführung einer OCR-Verarbeitung auf einem Bild. Die Erstextraktion von Metadaten erfasst die Quelle, das Erstellungsdatum, den Dokumenttyp sowie alle formatbezogenen Details, die zum Zeitpunkt der Analyse sichtbar sind. Anschließend folgen die Bereinigung, das Aufteilen in Abschnitte und das Einbetten der Daten. Die einzige Aufgabe der Eingabeverarbeitung besteht darin, aus den Rohdaten sauberen Text zu gewinnen.

Die chaotische Realität von Bankdokumenten

Die Wissensdatenbank einer Bank erscheint nicht als ordentlicher Markdown-Text. Es handelt sich um Jahrzehnte an angesammelten Formaten: gescannte PDFs aus den 1990er Jahren, Word-Dateien mit Änderungsverfolgung, HTML-Seiten von einem internen Konformitätsportal, JSON-Exporte aus einem Richtlinienverwaltungssystem, E-Mails mit Anhängen sowie SharePoint-Seiten mit tiefen Ordnerstrukturen. Jedes Format benötigt seinen eigenen Parser, und jeder Parser versagt auf seine eigene Weise. Noch schlimmer ist, dass ein Parserfehler stumm bleibt – ein Parser, der verdrehten Text erzeugt, wirft keine Ausnahme aus. Er gibt den verdrehten Text zurück und macht weiter.

PDFs: drei Arten, drei Behandlungsweisen

PDF dominiert regulatorische und konformitätsbezogene Materialien und ist zugleich am schwierigsten zu parsen. Drei Varianten erfordern unterschiedliche Handhabung:

• Textschicht-PDFs stammen aus Textverarbeitungsprogrammen oder Funktionen zum Erstellen von PDFs. Der Text befindet sich im Inhaltstrom und kann direkt extrahiert werden.
• Gescannte PDFs sind Bilddateien, die aus physischem Papier erstellt wurden. Es gibt überhaupt keine Textschicht, weshalb OCR unerlässlich ist. Ältere Mitteilungen und historische Richtlinien gehören meist in diese Kategorie.
• Hybride PDFs sind Scans, denen eine von schlechter Qualität stammende OCR-Textschicht hinzugefügt wurde. Sie stellen das größte Risiko dar, denn ein einfacher Parser verwendet diese fehlerbehaftete Schicht und geht davon aus, dass alles funktioniert.

Die Unterscheidung zwischen ihnen ist der erste Schritt, und die untenstehende Klassifizierungsfunktion prüft ein PDF und gibt an, um welche Art es sich handelt.

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

Gescannte Dateien benötigen einen OCR-Workflow, und die Übersicht vergleicht die drei realistischen Produktionsmöglichkeiten.

Für eine Bank, die sich mit RBI-Rundschreiben auseinandersetzt, ist Azure Document Intelligence die praktische Wahl. Es kann Scans unterschiedlicher Qualität verarbeiten, erkennt die Tabellenaufbau in den gescannten Seiten und bewertet das Vertrauensniveau pro Wort – dadurch können extrahierte Inhalte mit geringem Vertrauensniveau zur manuellen Überprüfung markiert werden. Der untenstehende Lader ruft diese Funktion auf und wendet diese Logik an.

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

Die Flagge needs_review ist in regulierten Umgebungen von Bedeutung. Eine Seite mit mehr als fünf Wörtern geringer Zuverlässigkeit sollte nicht unbemerkt weitergeleitet werden. Sie gehört in die Warteschlange eines menschlichen Prüfers oder weist zumindest eine Flagge in ihren Metadaten auf, damit eine spätere Bewertung sie entsprechend berücksichtigen kann.

Word-Dokumente, HTML und strukturierte Quellen

Word-Dateien werden häufig zur Erstellung interner Richtlinien und Verfahren verwendet. Sie enthalten Überschriften, Tabellen und Listen, die eine spätere Auffindbarkeit erleichtern, aber bei einem simplen Textextraktor verloren gehen. Der Loader durchläuft hier das Dokument und bewahrt diese Struktur bei.

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

Die Erfassung nach Abschnitten lohnt sich. Bei einem KYC-Verfahren mit vierzig Paragraphen macht es die Genauigkeit der Suche erheblich besser, zu wissen, dass ein bestimmter Teil aus „Abschnitt 3: Erhöhte Sorgfaltspflichten“ stammt und nicht aus „Abschnitt 1: Standard-KYC“ – das sind Metadaten.

Konformitätsportale sind fast immer Webanwendungen, weshalb das Laden von HTML den Navigationsteilen, Werbeanzeigen, Cookie-Bannern sowie anderen Standardelementen um den eigentlichen Inhalt herum ausweichen muss.

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,
    }

Strukturierte Quellen wie Policy-Management-Systeme, Falldatenbanken und Transaktionsprotokolle erfordern einen anderen Ansatz. Ihr Text ist bereits sauber, daher stellt sich die Frage, welche Felder aufgenommen werden sollen und wie Zeilen als auffindbarer Text dargestellt werden können.

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

Banken speichern die meisten Arbeitsdateien in SharePoint oder Google Workspace, und um darauf zuzugreifen, sind OAuth sowie die entsprechenden Graph- oder Drive-APIs erforderlich.

import httpx
from typing import Generator

def load_sharepoint_library(
    tenant_id: str,
    client_id: str,
    client_secret: str,
    site_id: str,
    drive_id: str,
) -> Generator[dict, None, None]:
    """
    Iterate over all DOCX and PDF files in a SharePoint document library.
    Yields file metadata dicts; caller is responsible for downloading and parsing.
    """
    # Get access token
    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
    token_response = httpx.post(token_url, data={
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default",
    })
    access_token = token_response.json()["access_token"]
    headers = {"Authorization": f"Bearer {access_token}"}
    base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
    # List files recursively
    next_link = f"{base_url}/root/children"
    while next_link:
        response = httpx.get(next_link, headers=headers)
        data = response.json()
        for item in data.get("value", []):
            if item.get("file") and item["name"].endswith((".pdf", ".docx")):
                yield {
                    "name": item["name"],
                    "download_url": item.get("@microsoft.graph.downloadUrl"),
                    "created": item.get("createdDateTime"),
                    "modified": item.get("lastModifiedDateTime"),
                    "size": item.get("size"),
                    "web_url": item.get("webUrl"),
                }
        next_link = data.get("@odata.nextLink")

Batch oder Streaming

Jedes Design zur Dateneingabe beginnt mit einer Entscheidung: Sollen Dokumente sofort verarbeitet werden, sobald sie ankommen, oder in geplanten Chargen?

• Batch-Dateneingabe findet zu einem festgelegten Zeitpunkt statt, meist nachts oder wöchentlich. Sie eignet sich dann, wenn die Aktualität nicht entscheidend ist, die Menge groß und Kosten eine Rolle spielen. Eine Bank, die monatlich RBI-Bekanntmachungen einliest, benötigt keine Echtzeitverarbeitung – eine nächtliche Verarbeitung ist kostengünstiger.
• Streaming-Dateneingabe arbeitet über eine Ereigniswarteschlange, sobald Dokumente ankommen. Sie ist geeignet für Fälle, in denen die Aktualität entscheidend ist, wie beispielsweise ein Compliance-System, das neue Bekanntmachungen unmittelbar nach ihrer Veröffentlichung anzeigen muss.

Externe Anbindungen benötigen Schutz vor Rate-Limits, und die untenstehende Klasse umhüllt die Anfragen mit Wiederholungsversuchen und Zeitverzögerungen.

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")

Dieses Muster ist nicht optional. Die meisten externen APIs, einschließlich der Anbieter von Finanzdaten, blockieren eine IP-Adresse oder entziehen einen Schlüssel, sobald die Grenze überschritten wird. Der backoff_factor sorgt für ein exponentielles Warten, wobei bei aufeinanderfolgenden Versuchen 2, 4 und 8 Sekunden gewartet werden – das ist die absolute Mindestversuchsstrategie für einen Produktionsconnector.

Tabellen, Diagramme und Bilder

Der Wandel im Produktions-RAG im Jahr 2025 ging von rein textbasierten Pipelines zu multimodalen über. Bankdokumente sind nicht ausschließlich textbasiert; sie enthalten Tabellen mit Kapitalausstattungsquoten, Diagramme zu Trends bei Forderungsausfällen, Darstellungen der KYC-Abläufe sowie Bilder von unterzeichneten Vereinbarungen. Das Überspringen nicht-textuellen Inhalts führt dazu, dass ein großer Teil der Informationen in jedem Dokument weggelassen wird – und in regulatorischen Zusammenhängen ist genau dieser Teil oft der wertvollste. Die Tabelle in einer RBI-Rundmail, die genaue numerische Schwellenwerte angibt, ist genau das, was ein Compliance-Team benötigt.

Eine Tabelle kann auf zwei Arten verarbeitet werden: als Text serialisiert oder als strukturierte Daten extrahiert. Die richtige Wahl hängt von der weiteren Verwendung ab. Die untenstehende Extraktionsfunktion zieht Tabellen aus einem PDF heraus.

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

Serielles Markdown eignet sich für Tabellen, die von einem LLM gelesen und analysiert werden können, wie zum Beispiel bei der Frage „Welches CET1-Verhältnis verlangt Basel III?“ Strukturiertes JSON ist geeignet für Werte, die in Code abgefragt oder numerisch verglichen werden sollen.

Auch Diagramme, Grafiken und Infografiken dürfen nicht weggelassen werden. Ein Diagramm mit Trendlinien zu NPA-Werten nach Quartalen oder ein Flussdiagramm des FEMA-Transaktionszulassungsprozesses enthalten Informationen, die im umliegenden Text nicht zu finden sind. Das gängige Vorgehen für das Jahr 2025 besteht darin, jedes Bild an ein Vision-Language-Modell zu senden und die erzeugte Beschreibung zur Verwendung bei der Einbettung und Suche zu speichern.

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

Die Anfrage ist hier entscheidend. Eine allgemeine Anweisung wie „Beschreiben Sie dieses Bild“ liefert allgemeine Ergebnisse. Eine branchenspezifische Anweisung, bei der nach numerischen Werten, Trends, Etiketten oder Prozessschritten gefragt wird, ergibt beschreibende Ergebnisse, die auffindbar sind. Ein Ingenieur, der nach NPA-Trendlinien im Q3-Bericht fragt, benötigt eine Beschreibung, die „NPA“, „Trend“ sowie das Quartal enthält – nicht eine Beschreibung eines Balkendiagramms mit blauen Balken.

Audio und Video

Audio und Video stellen derzeit die Grenzgebiete für RAG im Bankwesen dar. Aufnahmen von Vorstandssitzungen, Transkripte von Analystenrufen, Pressekonferenzen der RBI sowie Schulungsvideos enthalten Material, das RAG-Systeme derzeit noch nicht erreichen können.

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,
    }

Die Zeitstempel der Segmente im Ausgabeformat sind wichtig. Wenn eine Frage durch etwas beantwortet wird, das 34 Minuten in einer Board-Call-Sitzung gesagt wird, sollte das System genau diese Position angeben und nicht einfach nur auf eine Aufzeichnung der Sitzung verweisen.

Ein Router für jedes Format

Getrennte Lader reichen nicht aus. Ein Produktionspipeline muss jedes Dokument an den richtigen Parser senden, Fehlern gelassen begegnen, den Verarbeitungsstatus überwachen und unabhängig von der Eingabe immer ein einheitliches Ausgabeformat liefern.

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

Das konsistente Ausgabeformat ist die wichtigste Gestaltungsempfehlung überhaupt. Der Chunker, der Embedding-Generator sowie der Metadaten-Anreicher erhalten alle denselben Dictionary-Strukturtyp – unabhängig davon, ob die Eingabe ein gescanntes Dokument, eine DOCX-Vertragsvorlage oder eine Audiodatei aus einem Compliance-Kurs war. Die needs_review-Flagge leitet problematische Dokumente in eine menschliche Bearbeitungskette weiter und verhindert so die Ausbreitung von Fehlern.

Drei Fehlerquellen hinter den meisten Problemen

Die meisten Probleme bei der Dokumentenabrufqualität in bankbasierten RAG-Systemen gehen auf drei Fehler bei der Dateneingabe zurück.

Ablehnung des Dokumenttyps bei der Analyse. Ein Textextraktor, der auf einem gescannten PDF ausgeführt wird, liefert entweder nichts oder nur einige verirrte Zeichen. Das Dateiinhalt wird zu leeren Blöcken, nahe-null-Werten sowie Indexeinträgen, die anschließend zurückgegeben werden. Das große Sprachmodell sieht einen leeren Kontext und erfindet eine Antwort. Die classify_pdf-Funktion verhindert dies mit nur wenigen Zeilen und spart so wochenlange Debugging-Arbeit.

Behandlung von Tabellen als Textinhalt. Eine Risikobewertungstabelle, die in Prosa umgewandelt wird, verliert ihre Zeilen und Spalten, und ein String wie „AAA 0,01% AA 0,05% A 0,10%“ kann nicht nützlich abgerufen werden. Wenn sie als Markdown-Tabelle mit Überschriften serialisiert wird, ist das möglich. Extrahieren Sie Tabellen getrennt und bewahren Sie ihre Struktur bei.

Fehlende Wiederholungsversuche bei Connectoren. Wenn ein SharePoint-Connector auf eine Netzwerkzeitüberschreitung stößt und nichts meldet, entstehen im Korpus Lücken, und niemand weiß, welche Dokumente fehlen. Das RateLimitedConnector-Muster löst sich aus, sobald die Wiederholungsversuche erschöpft sind, und protokolliert jeden Versuch, sodass die Lücken sichtbar werden.

Abwägungen bei Tools

• PyMuPDF, pdfplumber und PyPDF2. PyMuPDF ist der schnellste Extraktor für PDF-Dateien mit reinem Textinhalt. pdfplumber ist langsamer, aber weitaus besser bei der Verarbeitung von Tabellen. PyPDF2 ist am ältesten und am ungenauesten, daher sollte es in neuen Pipelines vermieden werden. Verwenden Sie PyMuPDF für den Text und pdfplumber für Tabellen derselben Datei.
• Selbst gehosteter oder cloud-basierter OCR. Tesseract ist kostenlos, läuft lokal und bewältigt saubere Scans recht gut. Bei Tausenden historischer Circulars unterschiedlicher Qualität rechtfertigt die höhere Genauigkeit von Azure Document Intelligence den Preis pro Seite. Falsch interpretierte regulatorische Schwellenwerte sind kein Kostenfaktor, bei dem man sparen sollte.
• Kosten durch Streaming. Das Echtzeit-Streaming jeder hochgeladenen Datei ist die teuerste Architektur, und eine Bank, die täglich nur zwei oder drei Dokumente hinzufügt, benötigt dies nicht. Nutzen Sie diese Methode nur in Fällen, in denen Aktualität Geschäftsergebnisse beeinflusst – beispielsweise, wenn ein Handelsteam sofortigen Zugang zu neuen SEBI-Circulars benötigt. Eine HR-Richtlinienbibliothek gehört nicht dazu.
• Multimodaler Kostenfaktor. Das Versenden jeder Bilddatei in jedem Dokument über GPT-4o verursacht im großen Maßstab echte Kosten. Filtern Sie kleine Bilder aus, wie z. B. dekorative Bilder unter 5 KB, entfernen Sie solche, deren Seitenverhältnis sie als Logos oder Icons kennzeichnet, und speichern Sie Beschreibungen im Cache, damit ein in mehreren Dokumenten wiederkehrendes Bild nur einmal beschrieben wird.

Rohtext ist kein sauberer Text

Was die Eingabe liefert, ist Rohtext, kein sauberer Text. Eine separate Reinigungsphase kümmert sich um die Entfernung von Störungen, die Deduplizierung, die Korrektur von OCR-Fehlern, die Multilingual-Normalisierung sowie das Entfernen von Standardtexten und befindet sich zwischen der Eingabe und dem Aufteilungsprozess. Die Aufteilung ist wichtig, weil sich die Fehler unterscheiden: Fehlschläge bei der Eingabe führen zu leeren oder verdorbenen Inhalten, während Fehlschläge bei der Reinigung zu Inhalten führen, die auf den ersten Blick in Ordnung erscheinen, aber subtile Störungen enthalten, die die Qualität der Einbettung beeinträchtigen – und diese zeigen sich erst bei der Ausführung eines Abruftests.

Zur Produktivnutzung benötigen die Codebeispiele Logging, eine Verwaltung von Geheimnissen (API-Schlüssel sollten in Umgebungsvariablen oder einem Secrets-Manager aufbewahrt werden, niemals hardcodiert) sowie Latenzmessung in jeder Phase.