Cómo crear un sistema de búsqueda de documentos de cumplimiento normativo para el sector fintech

Las empresas de tecnología financiera que operan en Europa no pueden recurrir de forma predeterminada a sistemas de «Generación Aumentada por Recuperación» (RAG) alojados en la nube para buscar documentos de cumplimiento normativo. El artículo 44 del RGPD restringe la transferencia de datos normativos fuera del Espacio Económico Europeo (EEE) sin las garantías adecuadas. Cuando se apruebe la nueva normativa sobre mercados de criptoactivos (MiCA) o una actualización de la Oficina para la Protección Financiera del Consumidor (CFPB), el equipo de cumplimiento normativo debe revisar decenas de miles de páginas de texto normativo, políticas internas e informes de auditoría para determinar si los procedimientos existentes contra el blanqueo de capitales siguen cumpliendo con la normativa. La búsqueda por palabras clave resulta demasiado literal para esta tarea. El RAG en la nube resolvería técnicamente el problema de la recuperación de información, pero las leyes de soberanía de datos imponen una restricción legal sobre dónde puede ejecutarse ese sistema.

En este blog, crearás un sistema RAG totalmente local para la búsqueda de cumplimiento normativo. Lo ejecutarás íntegramente en una infraestructura privada, procesarás documentos normativos, responderás a consultas en lenguaje natural con referencias y registrarás cada consulta para su auditoría.

Por qué la nube tiene sus limitaciones

Las restricciones normativas impiden que los datos salgan de los límites de la organización. Las API alojadas en la nube no son adecuadas para implementaciones prácticas de tecnología financiera. Te enfrentas a tres obstáculos legales principales que hacen que la infraestructura local sea un requisito arquitectónico imprescindible.

1. Soberanía de los datos y transferencias transfronterizas

Artículo 44 del RGPD prohíbe la transferencia de datos personales fuera del EEE sin garantías específicas. Si los documentos de cumplimiento contienen datos de clientes o expedientes de empleados, enviarlos a un índice de vectores en la nube con sede en EE. UU. supone un riesgo jurídico inmediato. La PSD2 (Directiva revisada sobre servicios de pago) y la MiCA imponen requisitos estrictos de tratamiento para los registros de pagos y criptoactivos. Una arquitectura local garantiza que sus datos nunca traspasen estas estrictas fronteras geográficas y legales.

2. Requisitos de auditabilidad y control

Autoridad de Conducta Financiera (FCA) y la Red de Control de Delitos Financieros (FinCEN) exigen que las decisiones tomadas a partir de datos sobre la lucha contra el blanqueo de capitales permanezcan bajo el control directo de la entidad regulada. Un índice de terceros no cumple el requisito de «estar bajo su control». Si un organismo regulador solicita inspeccionar la lógica en la que se basa una decisión de cumplimiento, deberá demostrar la integridad de la fuente de datos. No puede garantizar esta integridad cuando sus representaciones vectoriales residen en el entorno en la nube de un proveedor, donde carece de visibilidad sobre la infraestructura subyacente o las políticas de retención de datos.

3. La incrustación como fase regulada del tratamiento

La integración de un documento para RAG es un paso del proceso que requiere una base jurídica conforme a normativas como el RGPD y la CCPA. Los documentos de cumplimiento suelen contener datos personales sensibles. Al integrar datos en un servicio de terceros, se produce una transferencia de datos y es necesario evaluar al proveedor como encargado del tratamiento con arreglo a la normativa aplicable.

La ejecución local del proceso de integración reduce la exposición a las transferencias transfronterizas y simplifica las evaluaciones de riesgo de los encargados del tratamiento. Los datos permanecen dentro de la infraestructura controlada.

Flujo de datos de RAG en la nube frente al flujo de datos de RAG en las instalaciones

Repercusiones arquitectónicas

El diseño conforme a la normativa mantiene todo a nivel local:

• Los documentos permanecen dentro de tu red.
• Las representaciones se ejecutan de forma local.
• El índice vectorial sigue siendo local.
• Las consultas se ejecutan localmente.
• El LLM genera respuestas de forma local.
• El registro de auditoría almacena cada acción de forma local.

Cualquier arquitectura que envíe datos al exterior incumple al menos una restricción normativa.

Lo que estás construyendo

Estás desarrollando un sistema de búsqueda de cumplimiento normativo de tres capas que funciona íntegramente dentro de tu infraestructura y genera respuestas auditables y con referencias.

Capa 1: El proceso de ingesta

Extraer texto de documentos normativos en formato PDF y de políticas internas, dividir el contenido en segmentos de 512 tokens y generar representaciones utilizando all-MiniLM-L6-v2. El esquema de metadatos es la parte más importante de esta capa. Se introducen:

• Documentación regulatoria (RGPD, MiCA, CFPB, MAS, etc.).
• Políticas internas.
• Procedimientos de lucha contra el blanqueo de capitales y de identificación de clientes (KYC).
• Informes de auditoría.

La cartera de proyectos:

• Extraer texto de archivos PDF.
• Dividir los documentos en fragmentos (512 tokens, 50 de solapamiento).
• Generar representaciones.
• Almacenar en la base de datos Actian VectorAI junto con los metadatos.

El esquema de metadatos debería tener este aspecto:

• doc_type: normativa | política | auditoría
• jurisdicción: UE | EE. UU. | Reino Unido | APAC
• fecha_de_entrada_en_vigor: fecha ISO
• organismo regulador: p. ej., el RGPD, la FCA, la MAS

Capa 2: Consulta filtrada por jurisdicción

El sistema realiza una búsqueda híbrida. Cuando un responsable de cumplimiento formula una pregunta, el backend aplica un filtro de metadatos basado en la región del responsable. Esto evita que aparezcan las directrices de MAS en una consulta relacionada con el RGPD. El modelo de lenguaje grande (LLM) local genera una respuesta que debe incluir referencias (nombre del documento, sección y fecha).

Un responsable de cumplimiento normativo pregunta:

«¿Cuáles son los requisitos en materia de transferencia de datos establecidos en el artículo 44 del RGPD?»

El sistema:

• Incorpora la consulta de forma local.
• Realiza una búsqueda híbrida (vectorial + filtro de metadatos).
• Filtra por jurisdicción y tipo de documento.
• Recupera fragmentos relevantes.
• Envía los fragmentos a un LLM local.
• Devuelve una respuesta citada (nombre del documento, sección, fecha).

Aplicáis el filtrado porque la relevancia depende de la jurisdicción.

Nivel 3: El nivel de auditoría obligatorio

Cada consulta genera un registro de auditoría local que incluye el ID de usuario, los fragmentos recuperados y un hash de la respuesta generada. Este registro ofrece a los auditores una forma clara de verificar que las respuestas se basan en los datos citados y que el acceso a la información sigue estando controlado.

Límites de la red de la organización

Configuración básica de hardware

Puedes ejecutar este sistema en:

• Al menos 8 GB de RAM (se recomiendan 16 GB o más).
• 10 GB de espacio en disco (se recomiendan 100 GB o más).

Requisitos previos

Para seguir el tutorial, instala las siguientes herramientas:

• Docker y Docker Compose.
• Python 3.10 o superior.
• PIP o UV: En esta guía se utilizan UV por su rapidez y fiabilidad.

Configuración del proyecto

Antes de crear tu sistema de búsqueda de cumplimiento normativo, configura tu entorno local de modo que todo el procesamiento se realice dentro de los límites de tu red. Esta configuración utiliza uv, un instalador rápido de paquetes de Python, para mantener el entorno reproducible y aislado.

Descarga el paquete de cliente de Actian VectorAI. Esto crea un archivo actian_vectorai-0.1.0b2-py3-none-any.whl .

1. Inicializa tu espacio de trabajo
   Crea un directorio específico para tu proyecto de cumplimiento normativo en el sector fintech.
   mkdir compliance-rag && cd compliance-rag

uv init .

2. Activar el entorno virtual
   En un nuevo terminal, crea y activa un entorno virtual:

uv venv

3. Instalar las dependencias
   Añade las bibliotecas necesarias para operaciones vectoriales, extracción de texto e incrustaciones locales.

# 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

Creación de un sistema de búsqueda de documentos de cumplimiento normativo

En esta sección, crearás un sistema RAG totalmente local que recopile documentos normativos, garantice una búsqueda que tenga en cuenta la jurisdicción, responda a consultas de cumplimiento normativo con referencias y registre todas las interacciones con fines de auditoría.

Paso 1: Implementar una base de datos vectorial

Implementa una instancia local de Actian VectorAI DB con almacenamiento persistente tanto para los datos vectoriales como para los registros de auditoría.

Crea un archivo archivo docker-compose.yaml :

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

Ejecuta el servicio:

docker-compose up -d

Resultado esperado:

Salida del proceso de inicio del contenedor

La base de datos se inicia y abre el puerto 50051 para el acceso local. Los datos vectoriales se almacenan en ./data. Los registros de auditoría se escriben directamente en ./audit_logs en el servidor, lo que mantiene todos los registros de acceso dentro de los límites de su red.

Nota: 

• VectorAI DB se encuentra en fase de desarrollo activo. Comprueba los parámetros en la documentación oficial antes de utilizarlo en entorno de producción.
• Guarda los registros de auditoría fuera del contenedor para conservar las pruebas de cumplimiento.

Comprueba la implementación:

docker-compose logs

Salida del terminal que muestra que el contenedor se ha iniciado correctamente y su estado de funcionamiento

Paso 2: Crear el canal de ingesta

Ejecuta el proceso de ingestión para convertir los documentos normativos y las políticas internas en representaciones vectoriales y almacénalas en tu base de datos vectorial local.

Crear ingest.py e introduce el siguiente contenido:

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)

Este script realiza las siguientes acciones:

• Divide los textos en fragmentos: El sistema divide los documentos normativos en segmentos de 512 tokens con un solapamiento de 50 tokens para mantener el contexto jurídico entre los límites de los segmentos.
• Incorpora los fragmentos: El modelo convierte los segmentos en vectores numéricos utilizando un modelo local, lo que garantiza que tus datos nunca salgan de tu infraestructura.
• Almacena los metadatos: Cada vector se almacena con su jurisdicción y tipo de documento. Estos campos son obligatorios para garantizar el cumplimiento de su perímetro normativo durante la búsqueda.

Ejecutar la ingesta:

uv run ingest.py

Resultado esperado:

Registros de ingesta que muestran el número de fragmentos indexados

Paso 3: Ejecuta tus consultas

Ejecuta consultas en tu sistema RAG local y comprueba la recuperación de datos, el filtrado basado en la jurisdicción y el registro de auditoría.

Crea un archivo query.py con el siguiente contenido:

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

El script realiza tres operaciones principales:

1. Aplica el control de jurisdicción: El sistema comprueba tu rol antes de la recuperación. La jurisdicción se aplica a nivel de la base de datos mediante la inserción de un filtro en la solicitud de búsqueda vectorial. La base de datos solo devuelve documentos que se encuentran dentro del ámbito normativo permitido.
2. Recupera datos filtrados: La búsqueda de similitud vectorial se limita a tu ámbito normativo específico.
3. Registra el registro de auditoría obligatorio: Cada evento de consulta se registra localmente para cumplir con sus requisitos de cumplimiento normativo.

Ejecutar:

uv run query.py

Resultado esperado:

Resultado de la consulta que muestra los fragmentos recuperados con sus puntuaciones y metadatos

Paso 4: Generar respuestas con un modelo de lenguaje grande (LLM) local

Las respuestas se generan utilizando únicamente los documentos recuperados.

Iniciar Ollama:

ollama run mistral:7b

Añade la lógica de generación a tu archivo query.py:

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

Este paso garantiza:

• No se realizan llamadas a API externas.
• Respuestas basadas en tus documentos.
• Se incluyen las referencias para garantizar la verificabilidad.

Vuelve a ejecutarlo:

<>uv run query.py

Resultado esperado:

Registros de respuestas generados por Ollama

Nota: La ejecución de mistral:7b de forma local sin aceleración por GPU tarda entre 2 y 3 minutos en la primera inferencia, pero las consultas posteriores se benefician del almacenamiento en caché del modelo.

Paso 5: Configurar el registro de auditoría

Almacena cada consulta localmente utilizando la asignación de volúmenes definida durante la implementación.

La configuración de Docker monta ./audit_logs desde tu host en el contenedor. Cada interacción con el sistema crea una entrada en compliance_queries.jsonl.

Ejemplo de entrada en el registro:

{"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"}

Cada entrada recoge:

• ¿Quién presentó la solicitud?
• Lo que preguntaron.
• A qué documentos se ha accedido.

Este archivo forma parte íntegramente de tu infraestructura y cumple con los requisitos de auditoría.

Gestión de las actualizaciones normativas

Actualiza los documentos normativos de forma incremental sin necesidad de volver a generar el índice. Cuando un organismo regulador publique una nueva versión, solo tendrás que sustituir los vectores de los documentos afectados. De este modo, se preserva la integridad de la auditoría y se mantiene el sistema actualizado.

Cuando se modifique un documento de orientación, elimina todos los vectores vinculados al document_id existente, e inserta los fragmentos actualizados con la nueva fecha de vigencia.

Actualiza tu canalización de ingestión

Añade esta función a ingest.py:

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

Ejemplo de uso

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

Ejecuta la actualización

uv run ingest.py

Resultado esperado:

Registros de importación y actualización

Esto:

• Importar los documentos iniciales
• Actualización MAS_637_2025 con nuevo contenido y fecha (16 de abril de 2026)

Comprueba la actualización

uv run query.py

Resultado esperado:

Resultado de la consulta con el documento actualizado

El sistema:

• Devuelve el documento actualizado (16 de abril de 2026) para las consultas SG.
• Muestra la nueva sección en los resultados obtenidos.
• Genera respuestas basadas en el contenido actualizado.

Conclusión

Las restricciones normativas prohíben enviar datos de cumplimiento a sistemas RAG en la nube. Has resuelto este problema creando un flujo de trabajo totalmente local en el que la ingesta, la recuperación, la generación y el registro de auditoría se mantienen dentro de tu infraestructura.
Ahora dispones de:

• Un corpus de cumplimiento normativo con función de búsqueda.
• Filtrado que tiene en cuenta la jurisdicción.
• El LLM local responde con referencias.
• Un registro de auditoría completo para cada consulta.

Aplica esta arquitectura a otras cargas de trabajo reguladas, como el análisis de fraudes o los sistemas de riesgo internos. Consulta la documentación de VectorAI DB y el repositorio de GitHub para conocer las actualizaciones y los detalles de implementación.

Únete a la comunidad y descubre más Acerca de Actian.

Acerca del autor

Acerca de Tahiya Chowdhury

Tahiya Chowdhury es la directora de producto de Actian Zen y VectorAI DB, donde dirige la estrategia de la plataforma de datos en el borde más sólida del sector. Gracias a su experiencia en MongoDB y Goldman Sachs, Tahiya se especializa en crear productos que se sitúan en la intersección entre las necesidades empresariales a gran escala y la agilidad de los desarrolladores modernos. Le apasiona eliminar la complejidad de la infraestructura de datos, lo que permite a los equipos de ingeniería pasar más rápidamente del prototipo a la producción.