Cómo crear un RAG multicliente para la atención al cliente

Los tickets de asistencia contienen datos de los clientes, como nombres, direcciones de correo electrónico, datos de la cuenta y patrones de uso. En virtud del Reglamento General de Protección de Datos (RGPD) y de la Ley de Privacidad del Consumidor de California (CCPA), estos datos imponen obligaciones a los encargados del tratamiento.

La incorporación de estos tickets en un índice vectorial en la nube constituye una actividad de tratamiento. Esto requiere una base jurídica válida, el cumplimiento de los requisitos de residencia de datos y una autorización contractual explícita.

La mayoría de los ingenieros de plataformas que desarrollan herramientas internas de IA se topan con el mismo obstáculo. Un prototipo de «Generación Aumentada por Recuperación» (RAG) basado en la nube funciona bien en el entorno de pruebas, pero los requisitos de cumplimiento normativo frenan su implementación. Un cliente puede exigir que los datos permanezcan en el EEE en virtud de un acuerdo de tratamiento de datos (DPA); un auditor puede cuestionar hacia dónde fluye la información del cliente durante el procesamiento de la IA; o los equipos jurídicos pueden señalar los tickets que contengan números de cuenta sujetos a regulación. Cada caso plantea el mismo obstáculo: los datos de los clientes salen de la infraestructura interna.

Las empresas que implementan la búsqueda semántica en su base de conocimientos de atención al cliente registrantiempos de resolución entre un 40 % y un 60 %más rápidos cuando los agentes tienen acceso a ella. Los sistemas RAG combinan la recuperación (búsqueda de contexto relevante) con la generación (respuestas basadas en IA), lo que los hace ideales para la atención al cliente. Los argumentos a favor de su rendimiento son sólidos. La arquitectura solo tiene que garantizar que los datos se mantengan locales y seguros.

Este tutorial muestra cómo utilizar Actian VectorAI DB para crear un sistema RAG multitenant que mantenga los datos de los clientes dentro de la infraestructura local. VectorAI DB admite el filtrado de metadatos a través de su API de filtrado, con un aislamiento impuesto por cada tenant a nivel de aplicación. Incorpora los tickets localmente, los almacena con los campos de identificación de cliente necesarios y ejecuta consultas que respetan los límites de cada tenant. Ninguna información de identificación personal (PII) de los clientes traspasa el perímetro de la red.

Tres formas en las que el RAG en la nube genera responsabilidad civil

Cloud RAG genera tres modos de fallo críticos al gestionar los datos de atención al cliente:

1. Los tickets de asistencia contienen información de carácter personal (PII) que las bases de datos en la nube no pueden proteger adecuadamente. Según el RGPD y la CCPA, la incorporación de dichos tickets a un índice vectorial en la nube podría considerarse un tratamiento de datos, lo que daría lugar a requisitos legales y de residencia de datos que van más allá de lo que suelen cubrir los acuerdos de cuentaable (BAA) o la certificación SOC 2 existentes.
2. La llamada a la API de integración transmite datos y requiere los mismos controles de privacidad que cualquier otra solicitud a una API externa. Cuando se envía el texto del ticket a los servidores de OpenAI para su integración, se transfieren datos personales a un tercero.
3. Las arquitecturas multitenant entrañan el riesgo de que se expongan los datos de los clientes debido a filtros mal configurados o a errores en las API. El conector de IA de Asana expuso datos confidenciales de distintas organizaciones debido a un acceso a la herramienta con un ámbito de aplicación inadecuado, lo que generó responsabilidades contractuales que iban más allá de la cobertura del SLA del proveedor.
   La solución es de carácter arquitectónico: mantener las representaciones en las instalaciones propias, aislar a los clientes a nivel de metadatos y no enviar nunca datos confidenciales a servicios externos. Este tutorial muestra cómo crear esa arquitectura utilizando VectorAI DB.

Flechas de flujo de datos que cruzan el límite de la infraestructura (nube) frente a las que permanecen dentro (en las propias instalaciones)

Ahora que ya entiendes por qué el RAG en la nube no funciona para la atención al cliente, vamos a crear una alternativa que cumpla con los requisitos.

Lo que estás construyendo

En este tutorial se crea un sistema de tres capas que se ejecuta íntegramente en tu infraestructura. Los agentes de atención al cliente formulan preguntas; el sistema busca únicamente en el historial de tickets de ese cliente y en los artículos compartidos de la base de conocimientos (KB); y un modelo de lenguaje grande (LLM) local genera las respuestas citadas con los números de ticket y las referencias a la KB. Ningún dato de los clientes sale de la red interna.

Capa 1: Ingestión

El sistema importa los tickets de asistencia y los artículos de la base de conocimientos por separado. El sistema divide los tickets en fragmentos (512 tokens, con un solapamiento de 50 tokens) y los integra localmente utilizando all-MiniLM-L6-v2, y los almacena con el customer_id como campo de metadatos obligatorio.

El esquema de la colección incluye:

• tipo_de_fuente: «ticket» o «base_de_conocimientos»
• customer_id: Obligatorio para los tickets; nulo para los artículos de la base de conocimientos
• línea_de_productos: Área de productos a la que se refiere el ticket
• estado_del_ticket: Abierto, cerrado, escalado
• fecha_de_creación: Marca de tiempo para el filtrado cronológico

Cada fragmento de ticket incluye estos campos, junto con el contenido de texto y su representación vectorial. Los artículos de la base de conocimientos siguen el mismo proceso de representación, con el tipo de fuente establecido en «knowledge_base» y sin necesidad de un ID de cliente. Todos los clientes pueden acceder a estos fragmentos durante la búsqueda.

Capa 2: Consulta

Cuando un agente de atención al cliente formula una pregunta sobre el historial del cliente A, la capa de consultas aplica un filtrado por ID de cliente antes de la búsqueda semántica. El código de la aplicación crea filtros que limitan los resultados al cliente especificado. La API de filtrado de VectorAI DB aplica estos filtros, pero el aislamiento depende de que la aplicación los transmita correctamente en cada consulta. 

La búsqueda híbrida combina el filtro «customer_id» con la coincidencia semántica. El sistema devuelve fragmentos relevantes de los tickets de ese cliente, además de artículos de la base de conocimientos compartida. Los resultados incluyen puntuaciones de similitud y referencias a las fuentes.

Un modelo de lenguaje grande (LLM) local (Ollama con llama3.2:3b o mistral:7b) genera respuestas utilizando únicamente el contexto recuperado. La indicación RAG indica al modelo que cite las fuentes y que diga que no sabe la respuesta cuando falte contexto. La generación mediante un LLM local suele completarse en 2-5 segundos, dependiendo del tamaño del contexto y del hardware.

Capa 3: Aislamiento multitenant

  Evita las fugas de datos entre clientes cuando el código de la aplicación implementa los filtros correctamente. Todos los clientes comparten una única colección (support_data), y la separación entre clientes se garantiza mediante customer_id que se aplican en el momento de la consulta.

VectorAI DB aplica estos filtros a través de su API de filtrado. Sin embargo, si el código de la aplicación tiene un error y omite el filtro «customer_id», la base de datos devolverá todos los datos de los clientes. Las reglas del esquema no impiden las consultas sin restricciones. Es el código de la aplicación el que garantiza el aislamiento, no la base de datos.

Los artículos de la base de conocimientos se comparten entre todos los clientes. Los datos de los tickets están separados lógicamente. Una consulta sobre el cliente A debería crear filtros para recuperar:

• Todos los fragmentos con customer_id = «A» Y source_type = «ticket».
• Todos los fragmentos con source_type = «base_de_conocimientos» (sin filtro de clientes).

Cuando los filtros se aplican correctamente, una consulta no puede devolver fragmentos de tickets del Cliente B. La capa de aplicación es la encargada de garantizar la coherencia de los filtros en todas las rutas de consulta.

Alternativas arquitectónicas:

• Aislamiento real a nivel de colección: Crea una colección por cliente (p. ej., entradas_del_cliente_a, client_b_entradas). Esto establece un límite estricto que impide que se realicen consultas en la colección del cliente B al buscar datos del cliente A, incluso si el código contiene errores. La desventaja es la sobrecarga operativa añadida.
• Aislamiento del filtro de metadatos: Utiliza una única colección compartida con filtros aplicados por la aplicación. Es más sencillo de manejar, pero requiere una creación disciplinada de los filtros y una revisión del código para evitar fugas.

Este tutorial utiliza el aislamiento mediante filtros de metadatos, ya que ofrece un equilibrio entre la seguridad y la simplicidad operativa para los equipos que pueden aplicar normas de revisión de código y de pruebas.

Configuración básica de hardware: Este tutorial se ejecuta en una instancia con 16 GB de RAM. Las pruebas realizadas con cargas de trabajo similares mostraron una latencia de consulta inferior a 500 ms en hardware estándar. 

Arquitectura de tres capas (ingestión → almacenamiento → consulta) con una capa de control de identificadores de cliente

Para obtener más información sobre cómo medir la eficacia de este sistema, consulta «Cómo medir el rendimiento del sistema RAG».

Constrúyelo

Todos los bloques de código que aparecen a continuación están pensados para ejecutarse. El código se ha probado con la base de datos VectorAI y se ejecutará sin necesidad de modificaciones una vez que hayas instalado las dependencias.

Requisitos previos

• Docker y Docker Compose
• Python 3.10 o superior
• Gestor de paquetes UV instalado (curl -LsSf https://astral.sh/uv/install.sh | sh)
  

Dependencias de Python (instalar en el servidor):

uv add sentence-transformers pandas transformers actian-vectorai
# Or with pip:
pip install sentence-transformers pandas transformers actian-vectorai

Paso 1: Implementar VectorAI DB

Esto da como resultado: Una instancia de VectorAI DB en funcionamiento con almacenamiento persistente para los datos de los tickets y de la base de conocimientos.

# docker-compose.yml
version: '3.8'
services:
  vectorai-db:
    image: williamimoh/actian-vectorai-db:1.0b
    platform: linux/amd64
    container_name: vectorai-support-db
    ports:
      - "50052:50051"
    volumes:
      - ./data:/app/data
      - ./audit_logs:/app/audit_logs
    environment:
      - VECTORAI_LOG_LEVEL=info
    restart: unless-stopped

Explicación de los puntos de montaje de volúmenes:

• ./data:/app/data – VectorAI DB almacena sus archivos de base de datos en /app/data dentro del contenedor, que se asigna a ./data/ en tu servidor.
• ./audit_logs:/app/audit_logs – Creado para su uso futuro en caso de que se ejecuten scripts de auditoría dentro del contenedor.

Importante: En este tutorial, todos los scripts de Python se ejecutan en tu máquina host, no dentro del contenedor. Los scripts escriben los registros de auditoría en ./audit_logs/queries.jsonl en el host. El montaje del volumen de Docker es opcional para este tutorial, pero se incluye para aquellos equipos que más adelante quieran ejecutar scripts de Python dentro del contenedor.

Inicia la base de datos:

Ejecutar: 

docker-compose up -d

Inicio de Docker

Paso 2: Anonimizar los datos de carácter personal antes de su importación

Este paso genera datos de tickets limpios, en los que se han eliminado los datos de carácter personal, antes de que cualquier proceso se introduzca en tu base de datos vectorial.

Patrones habituales de datos de identificación personal (PII) que deben filtrarse:

# pii_filter.py
"""PII filtering for GDPR/CCPA compliance."""
import re

# GDPR/CCPA PII patterns
PII_PATTERNS = [
# Emails
    (r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]"),
# Phone numbers (US format)
    (r"\b\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b", "[PHONE]"),
# Credit card numbers
    (r"\b\d{4}[-\s]\d{4}[-\s]\d{4}[-\s]\d{4}\b", "[CARD]"),
# Social Security Numbers
    (r"\b\d{3}-\d{2}-\d{4}\b", "[SSN]"),
# Account numbers
    (r"\b[Aa]ccount\s*#?\s*:?\s*([A-Z]{2,4}[-]?)?\d{6,12}\b", "[ACCOUNT]"),
# Invoice numbers
    (r"\b[Ii]nvoice\s*#?\s*:?\s*\d{4,10}\b", "[INVOICE]"),
]

def sanitize_pii(text: str) -> str:
"""Remove GDPR/CCPA PII from text before embedding.
    Note: Date and IP patterns are excluded to avoid matching version numbers
    like "Python 3.10.4.1" and pagination like "page 3/10". For production,
    use Microsoft Presidio: https://microsoft.github.io/presidio/
    """
for pattern, replacement in PII_PATTERNS:
        text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
return text.strip()This filtering happens BEFORE chunking or embedding, ensuring no PII enters your pipeline.

Anonimización de datos personales

Paso 3: Crear el proceso de captación de tickets

Resultado: tickets de atención al cliente depurados que se dividen en fragmentos, se incrustan y se almacenan en la base de datos de VectorAI junto con los metadatos necesarios del ID del cliente. 

# ticket_ingestion.py
"""Ingest customer support tickets into VectorAI DB with tokenizer-based chunking."""
import hashlib
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer
from actian_vectorai import VectorAIClient, PointStruct, VectorParams, Distance
import pandas as pd
from pii_filter import sanitize_pii  # Import from Step 2

# Config
VECTORAI_HOST = "localhost:50052"
COLLECTION = "support_data"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
VECTOR_DIM = 384

# Initialize tokenizer for accurate token-based chunking
print(f"Loading tokenizer for: {EMBED_MODEL}")
tokenizer = AutoTokenizer.from_pretrained(EMBED_MODEL)

def chunk_text(text, chunk_size=512, overlap=50):
    """Split text into overlapping chunks by TOKEN count (not words).
    Args:
        text: Text to chunk
        chunk_size: Number of tokens per chunk (default: 512)
        overlap: Number of tokens to overlap between chunks (default: 50)
        
    Returns:
        List of text chunks
    """
    # Tokenize the entire text
    tokens = tokenizer.encode(text, add_special_tokens=False)
    chunks = []
    
    # Create overlapping chunks
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        # Decode back to text
        chunk_text = tokenizer.decode(chunk_tokens, skip_special_tokens=True)
        chunks.append(chunk_text)
    
    return chunks

# Initialize embedding model
print(f"Loading embedding model: {EMBED_MODEL}")
model = SentenceTransformer(EMBED_MODEL)

def embed(texts: list[str]) -> list[list[float]]:
    """Generate embeddings for a list of texts."""
    return model.encode(texts, normalize_embeddings=True).tolist()

def _generate_stable_id(text: str, index: int) -> int:
    """Generate stable integer ID using SHA-256 hash."""
    hash_input = f"{text}:{index}".encode()
    hash_output = hashlib.sha256(hash_input).hexdigest()
    return int(hash_output[:15], 16)

def ingest_tickets(csv_path, customer_id):
    """Ingest tickets for a specific customer.
    
    Args:
        csv_path: Path to CSV file with columns: ticket_id, ticket_text, 
                  product_line, status, created_date
        customer_id: Customer identifier for isolation
    """
    print(f"\nProcessing {csv_path} for {customer_id}...")
    df = pd.read_csv(csv_path)
    
    with VectorAIClient(VECTORAI_HOST) as client:
        # Create collection if it doesn't exist
        if not client.collections.exists(COLLECTION):
            client.collections.create(
                COLLECTION,
                vectors_config=VectorParams(size=VECTOR_DIM, distance=Distance.Cosine)
            )
            print(f"✓ Collection '{COLLECTION}' created (dim={VECTOR_DIM})")
        else:
            print(f"✓ Collection '{COLLECTION}' already exists")
        
        points_batch = []
        total_chunks = 0
        
        for idx, row in df.iterrows():
            # Step 1: Sanitize PII FIRST (imported from pii_filter.py)
            clean_text = sanitize_pii(row['ticket_text'])
            
            # Step 2: Chunk by TOKENS (not words)
            chunks = chunk_text(clean_text)
            
            # Step 3: Embed
            vectors = embed(chunks)
            
            # Step 4: Create PointStruct objects with strict metadata
            for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
                point_id = _generate_stable_id(row['ticket_id'], i)
                
                point = PointStruct(
                    id=point_id,
                    vector=vector,
                    payload={
                        "customer_id": customer_id,
                        "source_type": "ticket",
                        "ticket_id": row['ticket_id'],
                        "product_line": row.get('product_line', 'general'),
                        "ticket_status": row.get('status', 'closed'),
                        "created_date": row['created_date'],
                        "text": chunk,
                        "chunk_index": i,
                    }
                )
                points_batch.append(point)
            
            total_chunks += len(chunks)
        
        # Upload all points in batches
        if points_batch:
            client.upload_points(COLLECTION, points_batch, batch_size=100)
        
        print(f"✓ {customer_id}: ingested {len(df)} tickets, {total_chunks} chunks")

# Usage
if __name__ == "__main__":
    print("=" * 70)
    print("TICKET INGESTION - Token-Based Chunking")
    print("=" * 70)
    
    ingest_tickets("sample_tickets/customer_a_tickets.csv", customer_id="customer_a")
    ingest_tickets("sample_tickets/customer_b_tickets.csv", customer_id="customer_b")
    ingest_tickets("sample_tickets/customer_c_tickets.csv", customer_id="customer_c")
    
    print("\n" + "=" * 70)
    print("INGESTION COMPLETE!")
    print("=" * 70)

Ejecutar: 

uv run python ticket_ingestion.py

Resultado de la importación de tickets que muestra los bloques por cliente

Paso 4: Crear el proceso de incorporación de contenidos a la base de conocimientos

Esto da como resultado: Artículos de la base de conocimientos integrados y almacenados con el tipo de fuente «knowledge_base», accesibles para todos los clientes.

# kb_ingestion.py
"""Ingest knowledge base articles into VectorAI DB with tokenizer-based chunking."""
import glob
import hashlib
from actian_vectorai import VectorAIClient, PointStruct
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer

VECTORAI_HOST = "localhost:50052"
COLLECTION = "support_data"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"

# Initialize tokenizer for accurate token-based chunking
print(f"Loading tokenizer for: {EMBED_MODEL}")
tokenizer = AutoTokenizer.from_pretrained(EMBED_MODEL)

# Initialize embedding model
print(f"Loading embedding model: {EMBED_MODEL}")
model = SentenceTransformer(EMBED_MODEL)

def chunk_text(text, chunk_size=512, overlap=50):
"""Split text into overlapping chunks by TOKEN count (not words).
    Args:
        text: Text to chunk
        chunk_size: Number of tokens per chunk (default: 512)
        overlap: Number of tokens to overlap between chunks (default: 50)
    Returns:
        List of text chunks
    """
# Tokenize the entire text
    tokens = tokenizer.encode(text, add_special_tokens=False)
    chunks = []
# Create overlapping chunks
for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
# Decode back to text
        chunk_text = tokenizer.decode(chunk_tokens, skip_special_tokens=True)
        chunks.append(chunk_text)
return chunks

def _generate_stable_id(text: str, index: int) -> int:
"""Generate stable integer ID using SHA-256 hash."""
    hash_input = f"{text}:{index}".encode()
    hash_output = hashlib.sha256(hash_input).hexdigest()
return int(hash_output[:15], 16)

def ingest_kb_articles(markdown_dir):
"""Ingest knowledge base articles from markdown files.
    Args:
        markdown_dir: Directory containing .md files
    """
    print("=" * 70)
    print("KNOWLEDGE BASE INGESTION - Token-Based Chunking")
    print("=" * 70)
with VectorAIClient(VECTORAI_HOST) as client:
        points_batch = []
        total_chunks = 0
for filepath in glob.glob(f"{markdown_dir}/*.md"):
with open(filepath, 'r', encoding='utf-8') as f:
                content = f.read()
            article_id = filepath.split('/')[-1].replace('.md', '')
# Chunk by TOKENS (not words)
            chunks = chunk_text(content)
            vectors = model.encode(chunks, normalize_embeddings=True).tolist()
for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
                point_id = _generate_stable_id(article_id, i)
                point = PointStruct(
                    id=point_id,
                    vector=vector,
                    payload={
"source_type": "knowledge_base",
"article_id": article_id,
"text": chunk,
"chunk_index": i,
# NO customer_id -- shared across all customers
                    }
                )
                points_batch.append(point)
            total_chunks += len(chunks)
            print(f"✓ {article_id}: {len(chunks)} chunks")
# Upload all points
if points_batch:
            client.upload_points(COLLECTION, points_batch, batch_size=100)
        print(f"\n✓ KB ingestion complete: {total_chunks} chunks total")
        print("=" * 70)

if __name__ == "__main__":
    ingest_kb_articles("./knowledge_base")

Ejecutar:  

uv run python kb_ingestion.py

Resultado de la importación de la base de conocimientos

Paso 5: Crear un módulo de búsqueda compartida:

Resultado: El siguiente paso consiste en crear un módulo de búsqueda compartido que proporcione una función de búsqueda reutilizable para todos los scripts posteriores. Este módulo elimina la duplicación de funciones. Los cuatro pasos siguientes (consulta, demostración de aislamiento, sistema RAG y registro de auditoría) importan todos search_customer_tickets() desde este módulo compartido, en lugar de duplicar su lógica.

#search.py

"""Shared search functionality for customer-scoped queries.
"""
from actian_vectorai import VectorAIClient, FilterBuilder, Field
from sentence_transformers import SentenceTransformer

VECTORAI_HOST = "localhost:50052"
COLLECTION = "support_data"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"

# Initialize embedding model once
model = SentenceTransformer(EMBED_MODEL)

def search_customer_tickets(query_text, customer_id, top_k=5):
"""Search tickets for a specific customer plus shared KB articles.
    Args:
        query_text: Natural language query
        customer_id: Customer identifier for filtering
        top_k: Maximum number of results to return
    Returns:
        List of result dictionaries with score, customer_id, source_type, 
        ticket_id, article_id, and text fields
    """
    query_vector = model.encode([query_text], normalize_embeddings=True).tolist()[0]
    results = []
with VectorAIClient(VECTORAI_HOST) as client:
# Search customer's tickets
        ticket_filter = (
            FilterBuilder()
            .must(Field("customer_id").eq(customer_id))
            .must(Field("source_type").eq("ticket"))
            .build()
        )
        ticket_hits = client.points.search(
            collection_name=COLLECTION,
            vector=query_vector,
            limit=top_k,
            filter=ticket_filter
        )
# Search shared KB articles
        kb_filter = FilterBuilder().must(Field("source_type").eq("knowledge_base")).build()
        kb_hits = client.points.search(
            collection_name=COLLECTION,
            vector=query_vector,
            limit=top_k,
            filter=kb_filter
        )
# Combine results
for hit in ticket_hits + kb_hits:
            results.append({
"score": round(hit.score, 4),
"customer_id": hit.payload.get("customer_id"),
"source_type": hit.payload.get("source_type"),
"ticket_id": hit.payload.get("ticket_id"),
"article_id": hit.payload.get("article_id"),
"text": hit.payload.get("text", ""),
            })
# Sort by score and return top_k
    results.sort(key=lambda r: r["score"], reverse=True)
return results[:top_k]

Paso 6: Ejecutar consultas relacionadas con los clientes

Esto da como resultado: Resultados de búsqueda filtrados según los tickets de un cliente concreto, además de los artículos compartidos de la base de conocimientos.

# query.py
"""Run customer-scoped queries using shared search module."""
from search import search_customer_tickets

# Usage
if __name__ == "__main__":
    query = "How do I reset a customer's password?"
    customer = "customer_a"
    print(f"Query: {query}")
    print(f"Customer: {customer}\n")
    results = search_customer_tickets(query, customer_id=customer)
    print("Results:")
    print("=" * 70)
for i, hit in enumerate(results):
        source = hit['ticket_id'] or hit['article_id']
        cust = hit.get('customer_id') or 'N/A (KB)'
        print(f"\n{i+1}. [{hit['source_type']}] {source}")
        print(f"   Customer: {cust}")
        print(f"   Score: {hit['score']}")
        print(f"   Text: {hit['text'][:100]}...")
    print("\n" + "=" * 70)

Ejecutar:  

uv run python query.py

Fíjate en cómo query.py es ahora más conciso. Importa la función de búsqueda desde search.py en lugar de duplicar la lógica de búsqueda. El mismo patrón se mantiene en los tres pasos siguientes.

Paso 7: Demostrar el aislamiento multitenant con tres casos de prueba

Esto da como resultado: Prueba fehaciente de que el filtrado por identificación de cliente evita la fuga de datos entre clientes en los tres escenarios.

# isolation_demo.py
"""Demonstrate multi-tenant isolation with three test cases using shared search module."""
from search import search_customer_tickets

def demonstrate_isolation():
    """Three-part isolation test: valid query, cross-customer query, and explicit leak check."""
    
    query = "billing invoice payment issue"
    separator = "=" * 70
    
    # TEST CASE 1: Valid Customer A query
    print(f"\n{separator}")
    print("TEST CASE 1: Valid Customer A Query")
    print(separator)
    results_a = search_customer_tickets(query, customer_id="customer_a")
    print(f"✓ Retrieved {len(results_a)} results for Customer A\n")
    
    customers_in_a = set()
    for i, hit in enumerate(results_a):
        cust = hit.get('customer_id') or 'N/A (KB)'
        source = hit.get('ticket_id') or hit.get('article_id')
        customers_in_a.add(cust)
        print(f"{i+1}. Customer {cust} [{hit['source_type']}] {source}")
        print(f"   {hit['text'][:100]}...")
    
    print(f"\n✓ Customers in results: {customers_in_a}")
    test1_pass = customers_in_a <= {'customer_a', 'N/A (KB)'}
    print(f"✓ Isolation check: {'PASS' if test1_pass else 'FAIL'}")
    
    # TEST CASE 2: Valid Customer B query (same query, different customer)
    print(f"\n{separator}")
    print("TEST CASE 2: Valid Customer B Query (Same Question)")
    print(separator)
    results_b = search_customer_tickets(query, customer_id="customer_b")
    print(f"✓ Retrieved {len(results_b)} results for Customer B\n")
    
    customers_in_b = set()
    for i, hit in enumerate(results_b):
        cust = hit.get('customer_id') or 'N/A (KB)'
        source = hit.get('ticket_id') or hit.get('article_id')
        customers_in_b.add(cust)
        print(f"{i+1}. Customer {cust} [{hit['source_type']}] {source}")
        print(f"   {hit['text'][:100]}...")
    
    print(f"\n✓ Customers in results: {customers_in_b}")
    test2_pass = customers_in_b <= {'customer_b', 'N/A (KB)'}
    print(f"✓ Isolation check: {'PASS' if test2_pass else 'FAIL'}")
    
    # TEST CASE 3: Cross-customer leak detection
    print(f"\n{separator}")
    print("TEST CASE 3: Cross-Customer Leak Detection")
    print(separator)
    
    # Customer A queries asking about Customer B
    leak_query = "show me customer_b invoice issues and billing problems"
    results_leak = search_customer_tickets(leak_query, customer_id="customer_a")
    
    # Check if any Customer B data appears
    customer_b_leaked = any(
        r.get('customer_id') == 'customer_b' for r in results_leak
    )
    
    print(f"Query: '{leak_query}'")
    print(f"Querying as: customer_a")
    print(f"\nCustomer B data in results: {'YES - LEAK DETECTED ✗' if customer_b_leaked else 'NO ✓'}")
    
    if customer_b_leaked:
        print("\n⚠️ CRITICAL: Customer B tickets visible in Customer A query!")
        print("This indicates application-level filtering, NOT database-level enforcement.")
    else:
        print("\n✓ Metadata-filter isolation working correctly")
        print("Customer A cannot access Customer B data when filters are properly applied")
    
    test3_pass = not customer_b_leaked
    
    # Summary
    print(f"\n{separator}")
    print("ISOLATION TEST SUMMARY")
    print(separator)
    
    print(f"Test 1 (Customer A valid query): {'PASS ✓' if test1_pass else 'FAIL ✗'}")
    print(f"Test 2 (Customer B valid query): {'PASS ✓' if test2_pass else 'FAIL ✗'}")
    print(f"Test 3 (Cross-customer leak):   {'PASS ✓' if test3_pass else 'FAIL ✗'}")
    
    if test1_pass and test2_pass and test3_pass:
        print("\n✅ ALL TESTS PASSED - Multi-tenant isolation verified")
    else:
        print("\n❌ SOME TESTS FAILED - Check isolation configuration")

if __name__ == "__main__":
    demonstrate_isolation()

Ejecutar:  

uv run python isolation_demo.py

Resultados de las pruebas de aislamiento

Paso 8: Conexión del LLM local

Esto da como resultado: Un sistema RAG integral que genera respuestas citadas utilizando Ollama.

# rag_system.py
"""End-to-end RAG system with local LLM using shared search module."""
import json
import urllib.request
import urllib.error
from search import search_customer_tickets

OLLAMA_URL = "http://localhost:11434/api/generate"
OLLAMA_MODEL = "llama3.2:3b"

def generate_answer(query_text, customer_id):
"""Generate a cited answer using local LLM and retrieved context.
    Args:
        query_text: User's question
        customer_id: Customer identifier for filtering
    Returns:
        Generated answer with citations or fallback context
    """
# Retrieve relevant chunks using shared search
    chunks = search_customer_tickets(query_text, customer_id, top_k=3)
# Build context with citations
    context_parts = []
for i, chunk in enumerate(chunks):
        source_id = chunk.get('ticket_id') or chunk.get('article_id')
        source_type = chunk['source_type']
        context_parts.append(f"[{source_type.upper()} {source_id}]: {chunk['text']}")
    context = "\n\n".join(context_parts)
# RAG prompt
    system_prompt = (
"You are a support assistant. Answer ONLY using the provided context. "
"Do NOT use external knowledge. Cite each fact as [TICKET T-####] or [KB article-name]. "
"If the context does not contain the answer, say 'I cannot answer from the available documents.'"
    )
    prompt = f"{system_prompt}\n\nContext:\n{context}\n\nQuestion: {query_text}\n\nAnswer:"
# Call Ollama
    payload = json.dumps({
"model": OLLAMA_MODEL,
"prompt": prompt,
"stream": False,
"options": {"temperature": 0.3, "num_predict": 400},
    }).encode()
try:
        req = urllib.request.Request(
            OLLAMA_URL,
            data=payload,
            headers={"Content-Type": "application/json"},
        )
with urllib.request.urlopen(req, timeout=30) as resp:
            data = json.loads(resp.read())
return data.get("response", "").strip()
except urllib.error.URLError as e:
return f"[Ollama unreachable: {e}]\n\nRetrieved context:\n{context}"

# Usage
if __name__ == "__main__":
    query = "How do I reset a customer password?"
    customer = "customer_a"
    print(f"Query: {query}")
    print(f"Customer: {customer}\n")
    print("Generating answer...\n")
    answer = generate_answer(query, customer_id=customer)
    print("=" * 70)
    print("ANSWER:")
    print("=" * 70)
    print(answer)
    print("=" * 70)

Ejecutar:  

uv run python rag_system.py

Paso 9: Configurar el registro de auditoría

Esto da como resultado: Registro de auditoría completo de cada consulta, registrado localmente con total trazabilidad.

# audit.py
"""Audit logging for compliance using shared search module."""
import json
from pathlib import Path
from datetime import datetime, timezone
from search import search_customer_tickets

AUDIT_LOG = Path("./audit_logs/queries.jsonl")

def log_query(user_id, customer_id, query_text, results, access_denied=False):
"""Log every query for compliance audit trail.
    Args:
        user_id: Agent/user identifier
        customer_id: Customer identifier
        query_text: Search query
        results: List of search results
        access_denied: Whether access was denied (default: False)
    """
    AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
    record = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"user_id": user_id,
"customer_id": customer_id,
"query": query_text,
"results_count": len(results),
"sources": [
            {
"type": r.get('source_type'),
"id": r.get('ticket_id') or r.get('article_id'),
"score": r.get('score')
            }
for r in results
        ],
"access_denied": access_denied,
    }
with open(AUDIT_LOG, 'a', encoding='utf-8') as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

def query_with_audit(user_id, customer_id, query_text):
"""Execute query and log to audit trail.
    Args:
        user_id: Agent/user identifier
        customer_id: Customer identifier
        query_text: Search query
    Returns:
        List of search results
    """
    results = search_customer_tickets(query_text, customer_id)
    log_query(user_id, customer_id, query_text, results)
return results

# Example
if __name__ == "__main__":
    print("=" * 70)
    print("AUDIT LOGGING TEST")
    print("=" * 70)
    print("\nRunning sample queries with audit logging...\n")
# Query 1
    query1 = "password reset"
    results1 = search_customer_tickets(query1, "customer_a")
    log_query(
        user_id="agent_007",
        customer_id="customer_a",
        query_text=query1,
        results=results1
    )
    print(f"✓ Query 1 logged: '{query1}' (customer_a, {len(results1)} results)")
# Query 2
    query2 = "billing invoice payment"
    results2 = search_customer_tickets(query2, "customer_b")
    log_query(
        user_id="agent_008",
        customer_id="customer_b",
        query_text=query2,
        results=results2
    )
    print(f"✓ Query 2 logged: '{query2}' (customer_b, {len(results2)} results)")
    print(f"\n✓ Audit log written to: {AUDIT_LOG}")
    print("\nSample log entries:")
    print("-" * 70)
# Display last 2 entries
if AUDIT_LOG.exists():
with open(AUDIT_LOG, 'r', encoding='utf-8') as f:
            lines = f.readlines()
for line in lines[-2:]:
                entry = json.loads(line)
                print(json.dumps({
"timestamp": entry["timestamp"],
"user_id": entry["user_id"],
"customer_id": entry["customer_id"],
"query": entry["query"],
"results_count": entry["results_count"],
                }, indent=2))
                print()
    print("=" * 70)

Ejecutar:  

uv run python audit.py

Entradas del registro de auditoría con los identificadores de usuario y de cliente

Paso 10: Gestionar las actualizaciones de los billetes

Esto da como resultado: Incorporación gradual de nuevos tickets sin necesidad de reconstruir toda la colección.

# incremental_ingestion.py
"""Incremental ingestion - only ingest new tickets that don't exist in the database."""
from actian_vectorai import VectorAIClient, Field, FilterBuilder, PointStruct
from sentence_transformers import SentenceTransformer
from transformers import AutoTokenizer
import pandas as pd
import hashlib
import os
from pii_filter import sanitize_pii  # Import from Step 2

VECTORAI_HOST = "localhost:50052"
COLLECTION = "support_data"
EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
VECTOR_DIM = 384

# Initialize tokenizer for accurate token-based chunking
print(f"Loading tokenizer for: {EMBED_MODEL}")
tokenizer = AutoTokenizer.from_pretrained(EMBED_MODEL)

# Initialize embedding model
print(f"Loading embedding model: {EMBED_MODEL}")
model = SentenceTransformer(EMBED_MODEL)

def chunk_text(text, chunk_size=512, overlap=50):
    """Split text into overlapping chunks by TOKEN count (not words).
    
    Args:
        text: Text to chunk
        chunk_size: Number of tokens per chunk (default: 512)
        overlap: Number of tokens to overlap between chunks (default: 50)
        
    Returns:
        List of text chunks
    """
    # Tokenize the entire text
    tokens = tokenizer.encode(text, add_special_tokens=False)
    chunks = []
    
    # Create overlapping chunks
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        # Decode back to text
        chunk_text = tokenizer.decode(chunk_tokens, skip_special_tokens=True)
        chunks.append(chunk_text)
    
    return chunks

def embed(texts):
    return model.encode(texts, normalize_embeddings=True).tolist()

def _generate_stable_id(text: str, index: int) -> int:
    """Generate stable integer ID using SHA-256 hash."""
    hash_input = f"{text}:{index}".encode()
    hash_output = hashlib.sha256(hash_input).hexdigest()
    return int(hash_output[:15], 16)

def get_existing_ticket_ids(customer_id):
    """Query VectorAI DB for existing ticket IDs for a customer."""
    existing = set()
    
    with VectorAIClient(VECTORAI_HOST) as client:
        # Build filter for this customer's tickets
        ticket_filter = (
            FilterBuilder()
            .must(Field("customer_id").eq(customer_id))
            .must(Field("source_type").eq("ticket"))
            .build()
        )
        
        # Dummy vector for metadata-only search
        dummy_vector = [0.0] * VECTOR_DIM
        
        # Search with high limit to get all tickets
        hits = client.points.search(
            collection_name=COLLECTION,
            vector=dummy_vector,
            limit=10000,
            filter=ticket_filter
        )
        
        for hit in hits:
            ticket_id = hit.payload.get('ticket_id')
            if ticket_id:
                existing.add(ticket_id)
    
    return existing

def ingest_new_tickets(csv_path, customer_id):
    """Ingest only new tickets that don't exist in the database."""
    print(f"\nChecking for new tickets in {csv_path}...")
    df = pd.read_csv(csv_path)
    
    # Get existing ticket IDs from database
    existing_ids = get_existing_ticket_ids(customer_id)
    print(f"Found {len(existing_ids)} existing tickets for {customer_id}")
    
    # Filter to only new tickets
    new_tickets = df[~df['ticket_id'].isin(existing_ids)]
    
    if new_tickets.empty:
        print(f"✓ No new tickets for {customer_id} - all up to date!")
        return
    
    print(f"Found {len(new_tickets)} new tickets to ingest")
    
    # Ingest new tickets using same logic as main ingestion
    with VectorAIClient(VECTORAI_HOST) as client:
        points_batch = []
        total_chunks = 0
        
        for idx, row in new_tickets.iterrows():
            # Sanitize PII (imported from pii_filter.py)
            clean_text = sanitize_pii(row['ticket_text'])
            
            # Chunk by TOKENS (not words)
            chunks = chunk_text(clean_text)
            
            # Embed
            vectors = embed(chunks)
            
            # Create PointStruct objects
            for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
                point_id = _generate_stable_id(row['ticket_id'], i)
                
                point = PointStruct(
                    id=point_id,
                    vector=vector,
                    payload={
                        "customer_id": customer_id,
                        "source_type": "ticket",
                        "ticket_id": row['ticket_id'],
                        "product_line": row.get('product_line', 'general'),
                        "ticket_status": row.get('status', 'closed'),
                        "created_date": row['created_date'],
                        "text": chunk,
                        "chunk_index": i,
                    }
                )
                points_batch.append(point)
            
            total_chunks += len(chunks)
        
        # Upload new points
        if points_batch:
            client.upload_points(COLLECTION, points_batch, batch_size=100)
        
        print(f"✓ Ingested {len(new_tickets)} new tickets, {total_chunks} chunks for {customer_id}")

if __name__ == "__main__":
    print("=" * 70)
    print("INCREMENTAL INGESTION - New Tickets Only (Token-Based Chunking)")
    print("=" * 70)
    
    # Check if we have new ticket files
    if os.path.exists(os.path.join("sample_tickets", "customer_a_new_tickets.csv")):
        ingest_new_tickets(
            os.path.join("sample_tickets", "customer_a_new_tickets.csv"),
            customer_id="customer_a"
        )
    else:
        print("\nNo new ticket files found.")
        print("To test incremental ingestion:")
        print("1. Create customer_a_new_tickets.csv with new tickets")
        print("2. Run this script again")
    
    print("\n" + "=" * 70)

Acabas de crear un sistema RAG privado y multitenant

Has creado un sistema RAG multitenant listo para producción, basado en VectorAI DB, que mantiene la información de carácter personal (PII) de los clientes dentro de tu infraestructura. El aislamiento multitenant se garantiza mediante el filtrado de metadatos. Ningún dato de los clientes traspasa los límites de tu red. Cuando las autoridades reguladoras preguntan dónde se almacenan las representaciones, la respuesta es «en nuestra infraestructura».

Las plantas de fabricación, los proveedores de asistencia sanitaria y los equipos de servicios financieros utilizan este modelo para buscar registros confidenciales sin depender de la nube. El marco normativo varía según se trate del RGPD, la HIPAA o la norma PCI-DSS, pero el modelo de implementación sigue siendo el mismo: mantener las representaciones locales y no enviar nunca los datos a servicios externos.

Amplía esto para su uso en producción con un enrutamiento semántico que derive las consultas de baja fiabilidad, análisis que abarquen a varios clientes y protejan la información de carácter personal, ciclos de retroalimentación que mejoren la calidad de la recuperación de información y un aprendizaje incremental a partir de los tickets resueltos.

Prueba VectorAI DB con tus datos utilizando el repositorio de GitHub. Antes de pasar a producción, decide si vas a utilizar RAG o el ajuste fino, y cuándo optar por la implementación local frente a la nube. Únete a la comunidad de Actian en Discord, donde los ingenieros de la plataforma comparten patrones de implementación en producción.

Los datos de tus clientes merecen algo mejor que los sistemas RAG en la nube, en los que la ley de protección de datos protege al proveedor, pero no al cliente. Los proveedores de SaaS cifran los datos en tránsito, pero sus modelos integrados siguen viendo la información de carácter personal de tus clientes en texto sin cifrar. Acabas de demostrar que puedes crear una alternativa que cumpla con la normativa.

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.