So richten Sie ein Suchsystem für Compliance-Dokumente im Fintech-Bereich ein

Fintech-Unternehmen mit Geschäftstätigkeit in Europa können sich bei der Suche nach Compliance-Dokumenten nicht standardmäßig auf Cloud RAG-Systeme (Retrieval-Augmented Generation) verlassen. Artikel 44 der DSGVO schränkt die Übermittlung von regulatorischen Daten außerhalb des Europäischen Wirtschaftsraums (EWR) ohne angemessene Schutzmaßnahmen. Wenn eine neue Verordnung über Märkte für Krypto-Vermögenswerte (MiCA) oder eine Aktualisierung des Consumer Financial Protection Bureau (CFPB) vorliegt, muss das Compliance-Team Zehntausende Seiten an Regulierungstexten, internen Richtlinien und Prüfberichten durchsehen, um festzustellen, ob bestehende AML-Verfahren weiterhin konform sind. Die Stichwortsuche ist zu wörtlich für diese Aufgabe. Cloud würde das Problem der Datenermittlung zwar technisch lösen, doch Gesetze zur Datenhoheit schränken rechtlich ein, wo dieses System betrieben werden darf.

In diesem Blog werden Sie ein vollständig lokal betriebenes RAG-System für die Compliance-Suche aufbauen. Sie werden es vollständig innerhalb einer privaten Infrastruktur betreiben, regulatorische Dokumente verarbeiten, Anfragen in natürlicher Sprache mit Quellenangaben beantworten und jede abfragen Prüfungszwecke protokollieren.

Warum die Cloud ihre Grenzen Cloud

Regulatorische Beschränkungen verhindern, dass Daten die Grenzen des Unternehmens verlassen. Cloud APIs sind für den praktischen Einsatz im Fintech-Bereich ungeeignet. Sie stehen vor drei wesentlichen rechtlichen Hindernissen, die eine lokale Infrastruktur zu einer zwingenden architektonischen Voraussetzung machen.

1. Datenhoheit und grenzüberschreitende Datenübermittlungen

Artikel 44 der DSGVO verbietet die Übermittlung personenbezogener Daten außerhalb des EWR ohne spezifische Schutzmaßnahmen. Wenn Compliance-Dokumente Kundendaten oder Mitarbeiterunterlagen enthalten, führt deren Übermittlung an einen in den USA ansässigen Cloud zu einem unmittelbaren rechtlichen Risiko. PSD2 (überarbeitete Zahlungsdiensterichtlinie) und MiCA schreiben strenge Anforderungen an den Umgang für Zahlungs- und Krypto-Asset-Aufzeichnungen vor. Eine On-Premises stellt sicher, dass Ihre Daten diese festen geografischen und rechtlichen Grenzen niemals überschreiten.

2. Anforderungen an die Nachvollziehbarkeit und Kontrolle

Finanzaufsichtsbehörde (FCA) und das Financial Crimes Enforcement Network (FinCEN) schreiben vor, dass Entscheidungen, die auf der Grundlage von AML-Daten getroffen werden, unter der direkten Kontrolle des regulierten Unternehmens bleiben müssen. Ein Index eines Drittanbieters erfüllt die Anforderung „unter Ihrer Kontrolle“ nicht. Wenn eine Aufsichtsbehörde Einsicht in die Logik hinter einer Compliance-Entscheidung verlangt, müssen Sie die Integrität der Datenquelle nachweisen. Diese Integrität können Sie nicht gewährleisten, wenn sich Ihre Vektor-Embeddings in der Cloud eines Anbieters befinden, wo Sie keinen Einblick in die zugrunde liegende Infrastruktur oder die Richtlinien zur Datenaufbewahrung haben.

3. Einbettung als regulierter Verarbeitungsschritt

Das Einbetten eines Dokuments für RAG ist ein Verarbeitungsschritt, der gemäß Vorschriften wie der DSGVO und dem CCPA eine Rechtsgrundlage erfordert. Compliance-Dokumente enthalten häufig sensible personenbezogene Daten. Wenn Sie einbetten in einen Drittanbieter-Dienst einbetten , kommt es zu einer Datenübermittlung, und Sie müssen den Anbieter gemäß den geltenden Vorschriften als Auftragsverarbeiter bewerten.

Die lokale Ausführung der Einbettungspipeline verringert das Risiko grenzüberschreitender Datenübertragungen und vereinfacht die Risikobewertung durch den Auftragsverarbeiter. Die Daten verbleiben innerhalb der kontrollierten Infrastruktur.

Datenfluss bei On-Premises : Cloud . On-Premises

Architektonische Auswirkungen

Dank des konformen Designs bleibt alles lokal:

• Die Dokumente verbleiben in Ihrem Netzwerk.
• Embeddings werden lokal ausgeführt.
• Der Vektorindex bleibt lokal.
• Abfragen werden lokal ausgeführt.
• Das LLM generiert Antworten lokal.
• Im Prüfprotokoll wird jede Aktion lokal gespeichert.

Jede Architektur, die Daten nach außen übermittelt, verstößt gegen mindestens eine gesetzliche Vorschrift.

Was Sie entwickeln

Sie entwickeln ein dreistufiges Compliance-Suchsystem, das vollständig innerhalb Ihrer Infrastruktur läuft und überprüfbare, mit Quellenangaben versehene Ergebnisse liefert.

Ebene 1: Die Erfassungspipeline

Text aus PDF-Dateien mit behördlichen Unterlagen und internen Richtlinien extrahieren, den Inhalt in Segmente mit jeweils 512 Token aufteilen und Embeddings mithilfe von all-MiniLM-L6-v2. DasMetadatenschema ist der wichtigste Bestandteil dieser Ebene. Sie geben Folgendes ein:

• Behördliche Meldungen (DSGVO, MiCA, CFPB, MAS usw.).
• Interne Richtlinien.
• AML/KYC-Verfahren.
• Prüfungsberichte.

Die Pipeline:

• Text aus PDF-Dateien extrahieren.
• Dokumente in Blöcke unterteilen (512 Token, 50 Token Überlappung).
• Embeddings generieren.
• In der Actian VectorAI-Datenbank zusammen mit Metadaten speichern.

Das Metadaten sollte wie folgt aussehen:

• doc_type: Verordnung | Richtlinie | Prüfung
• Gerichtsbarkeit: EU | USA | Großbritannien | APAC
• Gültigkeitsdatum: ISO-Datum
• Aufsichtsbehörde: z. B. DSGVO, FCA, MAS

Ebene 2: Nach Gerichtsbarkeit gefilterte abfragen

Das System führt eine hybride Suche durch. Wenn ein Compliance-Beauftragter eine Frage stellt, fügt das Backend einen Metadaten ein, der auf der Region des Beauftragten basiert. Dadurch wird verhindert, dass bei DSGVO-bezogenen abfragen Leitlinien der MAS angezeigt werden. Das lokale LLM generiert eine Antwort, die Quellenangaben (Dokumentenname, Abschnitt und Datum) enthalten muss.

Ein Compliance-Beauftragter fragt:

„Was sind die Anforderungen an die Datenübermittlung gemäß Artikel 44 der DSGVO?“

Das System:

• Bindet die abfragen ein.
• Führt eine Hybrid-Suche durch (Vektor + Metadaten ).
• Filtert nach Gerichtsbarkeit und Dokumenttyp.
• Ruft relevante Blöcke ab.
• Sendet die Chunks an ein lokales LLM.
• Gibt eine zitierte Antwort zurück (Dokumentname, Abschnitt, Datum).

Sie wenden Filter an, da die Relevanz von der jeweiligen Rechtsordnung abhängt.

Ebene 3: Die obligatorische Prüfungsebene

Jede abfragen einen lokalen Aufzeichnung der Nutzer , den abgerufenen Datenblöcken und einem Hashwert der generierten Antwort. Dieser Aufzeichnung Prüfern eine klare Möglichkeit, zu überprüfen, ob die Antworten auf den zitierten Daten basieren und ob der Zugriff auf Informationen weiterhin kontrolliert bleibt.

Netzwerkgrenze der Organisation

Hardware-Basisausstattung

Sie können dieses System auf folgenden Systemen ausführen:

• Mindestens 8 GB RAM (16 GB oder mehr empfohlen).
• 10 GB Speicherplatz (100 GB+ empfohlen).

Voraussetzungen

Um die Anleitung nachvollziehen zu können, installieren Sie bitte die folgenden Tools:

• Docker und Docker Compose.
• Python .10 oder höher.
• PIP oder UV: In diesem Leitfaden werden UV aus Gründen der Geschwindigkeit und Zuverlässigkeit.

Projekteinrichtung

Bevor Sie Ihr Compliance-Suchsystem einrichten, konfigurieren Sie Ihre lokale Umgebung so, dass die gesamte Verarbeitung innerhalb Ihrer Netzwerkgrenzen bleibt. Bei dieser Konfiguration wird Folgendes verwendet: uv, ein schnelles Python installationsprogramm, um die Umgebung reproduzierbar und isoliert zu halten.

Laden Sie das Actian VectorAI-Client-Paketherunter. Dadurch wird eine Datei actian_vectorai-0.1.0b2-py3-none-any.whl Datei.

1. Arbeitsbereich einrichten
   Erstellen Sie ein eigenes Verzeichnis für Ihr Fintech-Compliance-Projekt.
   mkdir compliance-rag && cd compliance-rag

uv init .

2. Aktivieren Sie die virtuelle Umgebung
   Erstellen und aktivieren Sie in einem neuen Terminal eine virtuelle Umgebung:

uv venv

3. Abhängigkeiten installieren
   Fügen Sie die erforderlichen Bibliotheken für Vektoroperationen, Textextraktion und lokale Einbettungen hinzu.

# Install the Actian VectorAI Python client (ensure the .whl file is in your directory)
uv pip install actian_vectorai-0.1.0b2-py3-none-any.whl

# Add sentence-transformers, PDF processing tools
uv add sentence-transformers

Aufbau eines Suchsystems für Compliance-Dokumente

In diesem Abschnitt erstellen Sie ein vollständig On-Premises , das regulatorische Dokumente erfasst, eine rechtsgebietsspezifische Suche durchführt, Compliance-Anfragen mit Quellenangaben beantwortet und jede Interaktion zu Prüfungszwecken protokolliert.

Schritt 1: Eine Vektordatenbank bereitstellen

Stellen Sie eine lokale Instanz von Actian VectorAI DB mit persistenter Speicherung sowohl für Vektordaten als auch für Audit-Protokolle bereit.

Erstellen Sie eine docker-compose.yaml Datei:

services:
  vectorai:
    image: actian/vectorai:latest
    platform: linux/amd64
    container_name: vectorai_db
    ports:
      - "50051:50051"
    volumes:
      # vector data persists across restarts
      - ./data:/app/data
      # audit log lives on host -- not inside the container
      - ./audit_logs:/app/audit_logs
    environment:
      - VECTORAI_LOG_LEVEL=info
    restart: unless-stopped

Den Dienst starten:

docker-compose up -d

Erwartetes Ergebnis:

Ausgabe beim Starten des Containers

Die Datenbank wird gestartet und stellt den Port 50051 für den lokalen Zugriff bereit. Vektordaten werden in ./datagespeichert. Audit-Protokolle werden direkt in ./audit_logs auf dem Host geschrieben, wodurch alle Zugriffsaufzeichnungen innerhalb Ihrer Netzwerkgrenzen verbleiben.

Hinweis: 

• VectorAI DB befindet sich derzeit in der aktiven Entwicklung. Bitte überprüfen Sie die Parameter in der offiziellen Dokumentation, bevor Sie das System im Produktivbetrieb einsetzen.
• Speichern Sie die Audit-Protokolle außerhalb des Containers, um Nachweise für die Einhaltung der Vorschriften zu sichern.

Überprüfen Sie die Deployment:

docker-compose logs

Terminalausgabe, die den erfolgreichen Start des Containers und dessen Betriebszustand anzeigt

Schritt 2: Aufbau der Erfassungspipeline

Führen Sie die Erfassungspipeline aus, um behördliche Unterlagen und interne Richtlinien in Embeddings umzuwandeln und diese in Ihrer lokalen Vektordatenbank zu speichern.

Erstellen ingest.py und geben Sie folgenden Inhalt ein:

import hashlib
import json
from sentence_transformers import SentenceTransformer
from actian_vectorai import VectorAIClient, VectorParams, Distance, PointStruct, FilterBuilder, Field

# ── Config ────────────────────────────────────────────────────────────────────
BASE_ADDRESS = "localhost:50051"
COLLECTION = "compliance_docs"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
VECTOR_DIM = 384
CHUNK_TOKENS = 512
OVERLAP_TOKENS = 50

# ── Sample Regulatory Corpus ───────────────────────────────────────────────────
REG_DOCS = [
    {
        "document_id": "MAS_637_2025",
        "jurisdiction": "SG",
        "doc_type": "regulation",
        "section": "Section 2.1: Capital Adequacy Ratios",
        "date": "2025-01-15",
        "text": "MAS Notice 637 on Risk Based Capital Adequacy Requirements for Banks..."
    },
    {
        "document_id": "GDPR_ART_44",
        "jurisdiction": "EU",
        "doc_type": "regulation",
        "section": "Chapter 5: Transfers of Personal Data",
        "date": "2018-05-25",
        "text": "Any transfer of personal data to a third country shall take place only if..."
    }
]

# ── Processing Logic ───────────────────────────────────────────────────────────
def chunk_text(text, size=CHUNK_TOKENS, overlap=OVERLAP_TOKENS):
    tokens = text.split()
    chunks, start = [], 0
    while start < len(tokens):
        end = min(start + size, len(tokens))
        chunks.append(" ".join(tokens[start:end]))
        if end == len(tokens):
            break
        start += size - overlap
    return chunks


model = SentenceTransformer(EMBED_MODEL)


def ingest(docs):
    with VectorAIClient(BASE_ADDRESS) as client:
        # Create collection if it doesn't exist
        try:
            client.collections.create(
                COLLECTION,
                vectors_config=VectorParams(size=VECTOR_DIM, distance=Distance.Cosine),
            )
        except Exception as e:
            if "already exists" not in str(e).lower():
                raise

        points = []
        for doc in docs:
            chunks = chunk_text(doc["text"])
            vectors = model.encode(chunks).tolist()

            for i, chunk in enumerate(chunks):
                point_id = int(
                    hashlib.sha256(f"{doc['document_id']}:{i}".encode()).hexdigest()[:15],
                    16,
                )
                points.append(PointStruct(
                    id=point_id,
                    vector=vectors[i],
                    payload={
                        "document_id": doc["document_id"],
                        "jurisdiction": doc["jurisdiction"],
                        "doc_type": doc["doc_type"],
                        "section": doc.get("section", ""),
                        "date": doc.get("date", ""),
                        "text": chunk,
                    }
                ))

        client.points.upsert(COLLECTION, points)
        print(f"✓ Ingested {len(points)} chunks from {len(docs)} documents")


if __name__ == "__main__":
    ingest(REG_DOCS)

Dieses Skript führt die folgenden Aktionen aus:

• Teilt die Texte in Segmente auf: Das System unterteilt Rechtsdokumente in Segmente mit jeweils 512 Token und einer Überlappung von 50 Token, um den rechtlichen Kontext über die Segmentgrenzen hinweg zu gewährleisten.
• Bindet die Chunks ein: Das Modell wandelt Segmente mithilfe eines lokalen Modells in numerische Vektoren um und stellt so sicher, dass Ihre Daten Ihre Infrastruktur niemals verlassen.
• Speichert die Metadaten: Jeder Vektor wird mit seiner Zuständigkeitsbereich und Dokumenttyp. Diese Felder sind erforderlich, um Ihren regulatorischen Geltungsbereich bei der Suche durchzusetzen.

Erfassung ausführen:

uv run ingest.py

Erwartetes Ergebnis:

Aufnahmeprotokolle, aus denen die Anzahl der indizierten Chunks hervorgeht

Schritt 3: Führen Sie Ihre Abfragen aus

Führen Sie Abfragen in Ihrem lokalen RAG-System durch und überprüfen Sie die Datenextraktion, die zuständigkeitsbasierte Filterung sowie die Protokollierung von Prüfvorgängen.

Erstelle eine Datei abfragen.py mit folgendem Inhalt:

import json
import datetime
import hashlib
import requests # type: ignore
from pathlib import Path
from sentence_transformers import SentenceTransformer # type: ignore
from actian_vectorai import VectorAIClient, Field, FilterBuilder

# ── Config ────────────────────────────────────────────────────────────────────
BASE_ADDRESS = "localhost:50051"
COLLECTION = "compliance_docs"
AUDIT_LOG  = Path("./audit_logs/compliance_queries.jsonl")

# ── Jurisdiction Permissions ──────────────────────────────────────────────────
OFFICER_PERMISSIONS = {
    "sg_compliance_officer": ["SG"],
    "eu_data_privacy_lead":  ["EU"],
    "global_auditor":        ["SG", "EU", "UK", "US"]
}

model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")


def write_audit(record):
    AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
    with open(AUDIT_LOG, "a") as f:
        f.write(json.dumps(record) + "\n")

def run_compliance_query(user_id, role, query_text):
    timestamp = datetime.datetime.now().isoformat()
    allowed_jurisdictions = OFFICER_PERMISSIONS.get(role, [])

    if not allowed_jurisdictions:
        write_audit({
            "timestamp": timestamp,
            "user_id": user_id,
            "role": role,
            "query": query_text,
            "docs": [],
            "access": "DENIED"
        })
        return "Access Denied."

    # Embed query
    q_vec = model.encode([query_text])[0].tolist()

    # Build filter for jurisdiction access control (database-level enforcement)
    # Using Filter DSL: for single jurisdiction use must(), for multiple use should() with OR semantics
    if len(allowed_jurisdictions) == 1:
        # Single jurisdiction: use must() for strict enforcement
        jurisdiction_filter = FilterBuilder().must(
            Field("jurisdiction").eq(allowed_jurisdictions[0])
        ).build()
    else:
        # Multiple jurisdictions: OR them together using should()
        filter_builder = FilterBuilder()
        for jurisdiction in allowed_jurisdictions:
            filter_builder = filter_builder.should(Field("jurisdiction").eq(jurisdiction))
        jurisdiction_filter = filter_builder.build()

    with VectorAIClient(BASE_ADDRESS) as client:
        # Search with database-level jurisdiction filter
        # The filter is applied at the database level, not after retrieval
        search_results = client.points.search(
            COLLECTION,
            vector=q_vec,
            limit=3,  # Get top 3 results
            filter=jurisdiction_filter  # Database enforces access control here
        )

    # Convert results to our expected format
    results = [
        {
            "id": result.id,
            "score": result.score,
            "metadata": result.payload
        }
        for result in search_results
    ]

    doc_refs = [
        {
            "doc_id": r["metadata"].get("document_id"),
            "jurisdiction": r["metadata"].get("jurisdiction")
        }
        for r in results
    ]

    # Audit ONLY filtered (authorized) results
    write_audit({
        "timestamp": timestamp,
        "user_id": user_id,
        "role": role,
        "query": query_text,
        "jurisdictions_accessed": allowed_jurisdictions,
        "docs": doc_refs,
        "access": "ALLOWED"
    })

    return results

if __name__ == "__main__":
    print("\n--- Query 1: Authorized SG Officer ---")
    res = run_compliance_query(
        "officer_tan",
        "sg_compliance_officer",
        "What are capital adequacy rules?"
    )
    for r in res:
        print(f"Found: {r['metadata']['document_id']} ({r['metadata']['jurisdiction']})")

    print("\n--- Query 2: EU trying SG data ---")
    res2 = run_compliance_query(
        "officer_schmidt",
        "eu_data_privacy_lead",
        "MAS Notice 637"
    )
    print(f"Results returned: {len(res2)}")

Das Skript führt drei Kernvorgänge aus:

1. Sichert die Zuständigkeitskontrolle: Das System überprüft Ihre Rolle vor dem Abruf. Die Zuständigkeitskontrolle wird auf Datenbankebene durch Einfügen einer Filter-Nutzlast in die Vektorsuchanfrage durchgesetzt. Die Datenbank gibt nur Dokumente zurück, die innerhalb des zulässigen regulatorischen Rahmens liegen.
2. Ruft gefilterte Daten ab: Die Vektorähnlichkeitssuche ist auf Ihren spezifischen regulatorischen Rahmen beschränkt.
3. Schreibt das vorgeschriebene Prüfprotokoll: Jedes abfragen wird lokal protokolliert, um Ihre Compliance-Anforderungen zu erfüllen.

Ausführen:

uv run query.py

Erwartetes Ergebnis:

Abfrageergebnis mit den abgerufenen Chunks, deren Bewertungen und Metadaten

Schritt 4: Antworten mit einem lokalen LLM generieren

Die Antworten werden ausschließlich anhand der abgerufenen Dokumente generiert.

Ollama starten:

ollama run mistral:7b

Fügen Sie Generierungslogik zu Ihrer abfragen.pyhinzu:

import requests

OLLAMA_URL = "http://127.0.0.1:11434/api/generate"
OLLAMA_MODEL = "mistral:7b"

def build_context(results):
    """Build context string from retrieved documents for LLM prompt."""
    blocks = []
    for idx, result in enumerate(results, start=1):
        md = result["metadata"]
        doc_id = md.get("document_id", "unknown")
        section = md.get("section") or "(section not specified)"
        date = md.get("date") or "(date not specified)"
        text = md.get("text", "")

        blocks.append(
            f"Document {idx}: {doc_id}\n"
            f"Section: {section}\n"
            f"Date: {date}\n"
            f"Text: {text}"
        )
    return "\n---\n".join(blocks)


def generate_answer(results, query_text):
    """
    Generate an answer using Ollama's mistral:7b model.
    Note: First generation may take 2-3 minutes as the model loads into memory.
    Subsequent calls are faster once the model is warm.
    """
    if not results:
        return "No documents retrieved to generate answer from."
    
    context = build_context(results)
    prompt = f"""
Answer using only the context below.
Include document name, section, and date.

Context:
{context}

Question:
{query_text}
"""

    try:
        # Increase timeout to 600s (10 minutes) for mistral:7b generation
        # First run may take 2-3 minutes as the model loads
        response = requests.post(
            OLLAMA_URL,
            json={
                "model": OLLAMA_MODEL,
                "prompt": prompt,
                "stream": False,  # Ensure non-streaming response
            },
            timeout=600,  # 10 minutes to account for model warmup
        )
        response.raise_for_status()
        payload = response.json()
        # Ollama /api/generate endpoint returns 'response' field
        answer = payload.get("response", "")
        if not answer:
            return f"⚠️  Ollama returned empty response. Payload: {payload}\n\nContext for manual review:\n{context}"
        return answer
    except requests.exceptions.Timeout as e:
        return f"⚠️  Ollama generation timed out after 10 minutes.\n\nMistral:7b can take 2-3 minutes on first run. Try again--it will be faster next time.\n\nContext for manual review:\n{context}"
    except requests.exceptions.ConnectionError as e:
        return f"⚠️  Ollama server not running on {OLLAMA_URL}\n\nTo start Ollama:\n  docker compose up -d ollama\n  docker exec ollama_inference ollama pull mistral:7b\n\nContext for manual review:\n{context}"
    except Exception as e:
        return f"Error calling Ollama: {e}\n\nContext sent:\n{context}"

if res:
        print("\n--- Generated answer for SG officer ---")
        print(generate_answer(res, "What are capital adequacy rules?"))

Dieser Schritt gewährleistet:

• Keine externen API-Aufrufe.
• Antworten, die auf Ihren Unterlagen basieren.
• Quellenangaben zur Nachvollziehbarkeit.

Noch einmal ausführen:

<>uv run query.py

Erwartetes Ergebnis:

Von Ollama generierte Antwortprotokolle

Hinweis: Die lokale Ausführung von „mistral:7b“ ohne GPU-Beschleunigung dauert bei der ersten Inferenz 2–3 Minuten, nachfolgende Abfragen Nutzen jedoch Nutzen Modell-Caching.

Schritt 5: Konfigurieren Sie das Überwachungsprotokoll

Speichern Sie alle abfragen , indem Sie die bei Deployment definierte Datenträgerzuordnung verwenden.

Die Docker-Konfiguration bindet ./audit_logs von Ihrem Host in den Container ein. Jede Interaktion mit dem System erzeugt einen Eintrag in compliance_queries.jsonl.

Beispiel für einen Protokolleintrag:

{"timestamp": "2026-04-16T20:08:11.929639", "user_id": "officer_tan", "role": "sg_compliance_officer", "query": "What are capital adequacy rules?", "jurisdictions_accessed": ["SG"], "docs": [{"doc_id": "MAS_637_2025", "jurisdiction": "SG"}], "access": "ALLOWED"}
{"timestamp": "2026-04-16T20:08:13.397832", "user_id": "officer_schmidt", "role": "eu_data_privacy_lead", "query": "MAS Notice 637", "jurisdictions_accessed": ["EU"], "docs": [{"doc_id": "GDPR_ART_44", "jurisdiction": "EU"}], "access": "ALLOWED"}

Jeder Eintrag enthält:

• Wer hat den Antrag gestellt?
• Was sie gefragt haben.
• Auf welche Dokumente wurde zugegriffen?

Diese Datei verbleibt vollständig in Ihrer Infrastruktur und erfüllt die Anforderungen an die Rechnungsprüfung.

Umgang mit Änderungen der gesetzlichen Vorschriften

Aktualisieren Sie Regulierungsdokumente schrittweise, ohne den Index neu zu erstellen. Wenn eine Aufsichtsbehörde eine neue Version veröffentlicht, ersetzen Sie lediglich die betroffenen Dokumentvektoren. Dadurch bleibt die Integrität der Prüfkette gewahrt und Ihr System bleibt auf dem neuesten Stand.

Wenn sich ein Leitfaden ändert, löschen Sie alle Vektoren, die mit dem bestehenden document_id, und fügen Sie anschließend die aktualisierten Abschnitte mit dem neuen Gültigkeitsdatum ein.

Aktualisieren Sie Ihre Erfassungspipeline

Füge diese Funktion zu ingest.pyhin:

def update_regulatory_document(document_id, updated_doc):
    """
    Update a regulatory document by deleting old vectors and inserting new ones.
    This maintains 'Effective Date' integrity by replacing the entire document's chunks.
    """
    with VectorAIClient(BASE_ADDRESS) as client:
        # Delete existing vectors for the old document
        delete_filter = FilterBuilder().must(Field("document_id").eq(document_id)).build()
        client.points.delete(COLLECTION, filter=delete_filter)
        print(f"✓ Deleted old vectors for document {document_id}")

        # Ingest the updated document
        chunks = chunk_text(updated_doc["text"])
        vectors = model.encode(chunks).tolist()

        points = []
        for i, chunk in enumerate(chunks):
            point_id = int(
                hashlib.sha256(f"{updated_doc['document_id']}:{i}".encode()).hexdigest()[:15],
                16,
            )
            points.append(PointStruct(
                id=point_id,
                vector=vectors[i],
                payload={
                    "document_id": updated_doc["document_id"],
                    "jurisdiction": updated_doc["jurisdiction"],
                    "doc_type": updated_doc["doc_type"],
                    "section": updated_doc.get("section", ""),
                    "date": updated_doc.get("date", ""),
                    "text": chunk,
                }
            ))

        client.points.upsert(COLLECTION, points)
        print(f"✓ Updated document {document_id} with {len(points)} chunks (new effective date: {updated_doc.get('date', 'N/A')})")

Anwendungsbeispiel

# Update MAS_637_2025 with new content and date

updated_mas_doc = {
    "document_id": "MAS_637_2025",
    "jurisdiction": "SG",
    "doc_type": "regulation",
    "section": "Section 2.1: Capital Adequacy Ratios (Updated)",
    "date": "2026-04-16",  # New effective date
    "text": "Updated MAS Notice 637 text here..."
}

update_regulatory_document("MAS_637_2025", updated_mas_doc)

Führen Sie das Update aus

uv run ingest.py

Erwartetes Ergebnis:

Aufnahme + Aktualisierungsprotokolle

Das:

• Die ersten Dokumente einlesen
• Aktualisierung MAS_637_2025 mit neuen Inhalten und Datum (16.04.2026)

Überprüfen Sie das Update

uv run query.py

Erwartetes Ergebnis:

Abfrageergebnis mit aktualisiertem Dokument

Das System:

• Gibt das aktualisierte Dokument (16.04.2026) für SG-Abfragen zurück.
• Zeigt den neuen Abschnitt in den abgerufenen Ergebnissen an.
• Erzeugt Antworten, die auf den aktualisierten Inhalten basieren.

Zum Abschluss

Aufgrund gesetzlicher Auflagen ist die Übermittlung von Compliance-Daten an Cloud untersagt. Sie haben dieses Problem gelöst, indem Sie eine vollständig lokale Pipeline aufgebaut haben, in der die Erfassung, das Abrufen, die Generierung und die Protokollierung von Prüfprotokollen vollständig innerhalb Ihrer Infrastruktur erfolgen.
Sie verfügen nun über:

• Ein durchsuchbares Compliance-Korpus.
• Kontextsensitive Filterung.
• Der lokale LLM antwortet mit Quellenangaben.
• Ein vollständiges Protokoll jede abfragen.

Wenden Sie diese Architektur auf andere regulierte Anwendungsbereiche an, wie beispielsweise Betrugsanalysen oder interne Risikosysteme. Informationen zu Updates und Details zur Implementierung finden Sie in der VectorAI-DB-Dokumentation und auf GitHub Lager.

Werden Sie Teil der Community und erfahren Sie mehr über Actian.

Über den Autor

Über Tahiya Chowdhury

Tahiya Chowdhury ist Produktmanagerin für Actian Zen und VectorAI DB und leitet dort die Strategie für die branchenweit leistungsstärkste Edge-Datenplattform. Auf der Grundlage ihrer Erfahrungen bei MongoDB und Goldman Sachs hat sich Tahiya auf die Entwicklung von Produkten spezialisiert, die die Schnittstelle zwischen den Anforderungen großer Unternehmen und der Arbeitsgeschwindigkeit moderner Entwickler bilden. Ihr großes Anliegen ist es, die Komplexität der Dateninfrastruktur zu reduzieren und Entwicklungsteams dabei zu unterstützen, schneller vom Prototyp zur Produktion zu gelangen.