Comment mettre en place une solution RAG multi-locataires pour Support client

Support contiennent des données relatives aux clients, notamment leurs noms, leurs adresses e-mail, les informations relatives à leur compte et leurs habitudes d'utilisation. En vertu du Règlement général sur la protection des données (RGPD) et de la Loi californienne sur la protection de la vie privée des consommateurs (CCPA), ces données imposent des obligations au sous-traitant.

L'intégration de ces tickets dans un index vectoriel dans le cloud constitue une opération de traitement. Cela nécessite une base juridique valable, le respect des exigences en matière de résidence des données et une autorisation contractuelle explicite.

La plupart des ingénieurs de plateforme qui développent des outils d’IA internes se heurtent au même obstacle. Un prototype de « Retrieval Augmented Generation » (RAG) basé sur le cloud fonctionne bien en environnement de test, mais les exigences de conformité bloquent déploiement. Un client peut exiger que les données restent dans l’EEE en vertu d’un accord de traitement des données (DPA) ; un auditeur peut s’interroger sur le parcours des informations client lors du traitement par l’IA ; ou les équipes juridiques peuvent signaler des tickets contenant des numéros de compte soumis à une réglementation. Chaque cas soulève le même obstacle : les données client quittent l’infrastructure interne.

Les entreprises qui déploient la recherche sémantique dans leur base support constatentune réduction de 40 à 60 %des délais de résolution lorsque les agents y ont accès. Les systèmes RAG combinent la recherche (recherche de contextes pertinents) et la génération (pilotée par l’IA ), ce qui les rend idéaux pour support client. Les arguments en faveur de leur efficacité sont solides. Il suffit simplement que l’architecture garantisse la conservation des données en local et leur sécurité.

Ce tutoriel explique comment utiliser Actian VectorAI DB pour créer un système RAG multi-locataires qui conserve les données clients au sein de l'infrastructure locale. VectorAI DB prend en charge métadonnées via son API de filtrage, avec isolement imposé par locataire isolement niveau de l'application. Il intègre les tickets localement, les stocke avec les champs d'identifiant client requis et exécute des requêtes qui respectent les limites de chaque locataire. Aucune information personnelle identifiable (PII) des clients ne franchit le périmètre du réseau.

Trois façons dont le RAG dans le cloud engendre une responsabilité civile

Cloud RAG génère trois modes de défaillance critiques lors du traitement support client :

1. Support contiennent des données à caractère personnel que les bases de données dans le cloud ne sont pas en mesure de protéger de manière adéquate. En vertu du RGPD et du CCPA, l'intégration de ces tickets dans un index vectoriel dans le cloud peut être considérée comme un traitement de données, ce qui entraîne des obligations juridiques et des exigences en matière de résidence des données allant au-delà de ce que couvrent généralement les accords BAA ou SOC 2 existants.
2. L'appel à l'API d'intégration transmet des données et nécessite les mêmes mesures de protection de la vie privée que toute autre requête adressée à une API externe. Lorsque le texte d'un ticket est envoyé aux serveurs d'OpenAI en vue de son intégration, des données à caractère personnel sont transférées à un tiers.
3. Les architectures multi-locataires exposent les données des clients à des risques de fuite en raison de filtres mal configurés ou de bogues dans les API. Le connecteur IA d’Asana a divulgué des données sensibles à d’autres organisations en raison d’un accès à l’outil mal défini, ce qui a entraîné des responsabilités contractuelles dépassant SLA du fournisseur.
   La solution est d’ordre architectural : conserver les embeddings sur site, isoler les clients au métadonnées et ne jamais acheminer de données sensibles vers des services externes. Ce tutoriel explique comment mettre en place cette architecture à l’aide de VectorAI DB.

Flèches de flux de données franchissant les limites de l'infrastructure (cloud) par opposition à celles qui restent à l'intérieur (sur site)

Maintenant que vous comprenez pourquoi le RAG dans le cloud ne convient pas au support client, mettons en place une alternative conforme.

Ce que vous construisez

Ce tutoriel vous guide dans la mise en place d'un système à trois niveaux fonctionnant entièrement sur votre infrastructure. Support posent des questions ; le système effectue des recherches uniquement dans l'historique des tickets de ce client et dans les articles de la base de connaissances (KB) partagée ; et un modèle de langage grand (LLM) local génère des réponses accompagnées des numéros de ticket et des références à la base de connaissances. Aucune donnée client ne quitte le réseau interne.

Couche 1 : Ingestion

Le système traite séparément support et les articles de la base de connaissances. Il découpe les tickets en segments (512 tokens, avec un chevauchement de 50 tokens), puis les intègre localement à l'aide de all-MiniLM-L6-v2, puis les stocke avec l’ customer_id comme métadonnées obligatoire.

Le schéma de collecte comprend :

• source_type: « ticket » ou «base_de_connaissances« »
• customer_id: Obligatoire pour les tickets, nul pour les articles de la base de connaissances
• gamme_de_produits: Gamme de produits à laquelle le ticket se rapporte
• statut_du_ticket: Ouvert, clôturé, transmis à un niveau supérieur
• created_date: Horodatage pour le filtrage chronologique

Chaque segment de ticket comprend ces champs, ainsi que le contenu textuel et sa représentation vectorielle. Les articles de la base de connaissances suivent le même processus de représentation, le type de source étant défini sur «knowledge_base» et aucun identifiant client n’est requis. Ces extraits sont accessibles à tous les clients lors d’une recherche.

Couche 2 : requête

Lorsqu’un support pose une question concernant l’historique du client A, la requête applique un filtrage par identifiant client avant la recherche sémantique. Le code de l’application crée des filtres qui limitent les résultats au client spécifié. L’API de filtrage de VectorAI DB applique ces filtres, mais isolement de la transmission correcte de ces derniers par l’application à chaque requête. 

La recherche hybride combine le filtre « customer_id » et la correspondance sémantique. Le système renvoie des extraits pertinents issus des tickets de ce client ainsi que des articles de la base de connaissances partagée. Les résultats incluent des scores de similarité et des références aux sources.

Un LLM local (Ollama avec llama3.2:3b ou mistral:7b) génère des réponses en utilisant uniquement le contexte récupéré. La consigne RAG demande au modèle de citer ses sources et d’indiquer qu’il ne sait pas répondre lorsque le contexte fait défaut. La génération par un LLM local s’effectue généralement en 2 à 5 secondes, selon la taille du contexte et le matériel utilisé.

Couche 3 : isolement multi-locataires

  empêche les fuites de données entre clients lorsque le code de l'application met correctement en œuvre les filtres. Tous les clients partagent une seule collection (support), la séparation entre les clients étant assurée par customer_id passés au requête .

VectorAI DB applique ces filtres via son API de filtrage. Cependant, si le code de l’application comporte un bug et omet le filtre « customer_id », la base de données renverra toutes les données client. Les règles de schéma n'empêchent pas les requêtes sans restriction. C'est le code de l'application qui assure isolement, et non la base de données.

Les articles de la base de connaissances sont partagés entre tous les clients. Les données des tickets sont séparées de manière logique. Une requête le client A doit créer des filtres permettant de récupérer :

• Tous les blocs dont customer_id = « A » ET source_type = « ticket ».
• Tous les blocs dont source_type = «knowledge_base» (sans filtre client).

Lorsque les filtres sont correctement appliqués, une requête renvoyer de segments de ticket provenant du client B. La couche applicative est chargée d'assurer cohérence des filtres cohérence l'ensemble requête .

Autres solutions architecturales :

• isolement véritable isolement au niveau de la collection: Créer une collection par client (par exemple, client_a_billets, client_b_billets). Cela crée une barrière stricte qui empêche toute requête sur la collection du client B lors d’une recherche dans les données du client A, même en cas de code défectueux. L’inconvénient réside dans la charge opérationnelle supplémentaire.
• métadonnées- isolement: Utilisez une seule collection partagée avec des filtres appliqués par l'application. Ce mode de fonctionnement est plus simple, mais nécessite une conception rigoureuse des filtres et une révision du code afin d'éviter les fuites.

Ce tutoriel utilise isolement « métadonnées » isolement il offre un bon compromis entre sécurité et simplicité opérationnelle pour les équipes capables de faire respecter les normes en matière de révision du code et de tests.

Configuration matérielle de référence : ce tutoriel s'exécute sur une instance dotée de 16 Go de mémoire vive. Des tests réalisés avec des charges de travail similaires ont montré requête inférieure à 500 ms sur du matériel standard. 

Architecture à trois couches (ingestion → stockage → requête) avec une couche assurant le respect de l'identifiant client

Pour en savoir plus sur la manière de mesurer l'efficacité de ce système, consultez l'article « Comment mesurer les performances du système RAG ».

Construis-le

Chaque bloc de code ci-dessous est conçu pour s'exécuter. Le code a été testé avec la base de données VectorAI et s'exécutera sans modification une fois que vous aurez installé les dépendances.

Conditions préalables

• Docker et Docker Compose
• Python .10 ou version ultérieure
• Gestionnaire de paquets UV installé (curl -LsSf https://astral.sh/uv/install.sh | sh)
  

Python (à installer sur l'hôte) :

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

Étape 1 : Déployer VectorAI DB

Ce que cela donne: Une instance VectorAI DB opérationnelle avec un stockage persistant pour les données relatives aux tickets et à la base de connaissances.

# 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

Comprendre les montages de volumes :

• ./data:/app/data – VectorAI DB stocke ses fichiers de base de données dans /app/data à l’intérieur du conteneur, ce qui correspond à ./data/ sur votre hôte.
• ./audit_logs:/app/audit_logs – Créé pour une utilisation future si vous exécutez des scripts d'audit à l'intérieur du conteneur.

Important: Dans ce tutoriel, tous Python s'exécutent sur votre machine hôte, et non à l’intérieur du conteneur. Les scripts enregistrent les journaux d’audit dans le répertoire ./audit_logs/queries.jsonl sur l’hôte. Le montage du volume Docker est facultatif pour ce tutoriel, mais il est inclus pour les équipes qui souhaiteraient ultérieurement exécuter Python à l’intérieur du conteneur.

Démarrer la base de données :

Exécuter : 

docker-compose up -d

Démarrage de Docker

Étape 2 : Anonymiser les données à caractère personnel avant leur ingestion

Cette étape permet d'obtenir des données de tickets « propres », dont les informations à caractère personnel ont été supprimées, avant que le traitement ne soit intégré à votre base de données vectorielle.

Modèles courants d'informations personnelles identifiables (PII) à filtrer :

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

Anonymisation des données à caractère personnel

Étape 3 : Mettre en place le pipeline d'ingestion des tickets

Résultat : support client nettoyés, segmentés, intégrés à Embarqué et stockés dans la base de données VectorAI avec métadonnées d'identification client requises. 

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

Exécuter : 

uv run python ticket_ingestion.py

Sortie du traitement des tickets indiquant le nombre de blocs par client

Étape 4 : Mettre en place le pipeline d'ingestion de la base de connaissances

Ce que cela donne: Articles de la base de connaissances Embarqué stockés avec le type de source «knowledge_base», accessibles à tous les clients.

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

Exécuter :  

uv run python kb_ingestion.py

Résultats de l'intégration de la base de connaissances

Étape 5 : Créer un module de recherche partagé :

Résultat : L'étape suivante consiste à créer un module de recherche partagé qui fournit une fonction de recherche réutilisable pour tous les scripts suivants. Ce module élimine la duplication des fonctions. Les quatre étapes suivantes (requête, isolement , système RAG et journalisation d'audit) importent toutes search_customer_tickets() depuis ce module partagé plutôt que de dupliquer sa logique.

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

Étape 6 : Exécuter des requêtes portant sur un client spécifique

Ce que cela donne: Des résultats de recherche filtrés pour n'afficher que les tickets d'un client spécifique ainsi que les articles de la base de connaissances partagés.

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

Exécuter:  

uv run python query.py

Remarquez comment requête.py est désormais plus concis. Il importe la fonction de recherche depuis search.py au lieu de dupliquer la logique de recherche. Ce même principe s'applique aux trois étapes suivantes.

Étape 7 : Démontrer isolement multi-locataires isolement trois cas de test

Ce que cela donne: La preuve irréfutable que le filtrage par identifiant client empêche la fuite de données entre clients dans les trois scénarios.

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

Exécuter :  

uv run python isolation_demo.py

résultats isolement

Étape 8 : Raccordement du LLM local

Ce que cela donne: Un système RAG de bout en bout qui génère des réponses référencées à l'aide d'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)

Exécuter :  

uv run python rag_system.py

Étape 9 : Configurer la journalisation d'audit

Ce que cela donne: Une piste d'audit complète de chaque requête, enregistrée localement avec une traçabilité totale.

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

Exécuter :  

uv run python audit.py

Entrées du journal d'audit avec les identifiants utilisateur client

Étape 10 : Gérer les mises à jour des tickets

Ce que cela donne: Une intégration progressive des nouveaux tickets sans avoir à reconstruire l'intégralité de la collection.

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

Vous venez de mettre en place un système RAG privé multi-locataires

Vous avez mis en place un système RAG prêt pour la production , basé sur VectorAI DB, qui conserve les données à caractère personnel (PII) de vos clients au sein de votre infrastructure. isolement multi-locataires isolement assurée par métadonnées . Aucune donnée client ne franchit les limites de votre réseau. Lorsque les autorités de régulation vous demandent où sont stockés les embeddings, la réponse est : « dans notre infrastructure ».

Les usines de fabrication, les prestataires de soins de santé et les équipes du secteur des services financiers utilisent ce modèle pour effectuer des recherches dans des dossiers sensibles sans dépendre du cloud. Le cadre réglementaire varie selon le RGPD, l'HIPAA ou la norme PCI-DSS, mais le déploiement reste le même : conserver les représentations en local et ne jamais acheminer les données vers des services externes.

Étendez cette approche à l'environnement de production grâce à un routage sémantique qui transfère les requêtes à faible niveau de confiance, à des analyses inter-clients qui protègent les données à caractère personnel, à des boucles de rétroaction qui améliorent la qualité des résultats, et à un apprentissage incrémental tiré des tickets résolus.

Testez VectorAI DB avec vos données à l'aide du dépôt GitHub. Avant la mise en production, déterminez s'il convient d'utiliser RAG ouoptimiser, et dans quels cas privilégier une solution sur site le cloud. Rejoignez la communauté Actian sur Discord, où les ingénieurs de la plateforme partagent déploiement en production.

Les données de vos clients méritent mieux que les systèmes RAG dans le cloud, où la politique de protection des données (DPA) protège le fournisseur, et non le client. Les fournisseurs SaaS chiffrent les données en transit, mais leurs modèles intégrés continuent d'accéder aux informations personnelles identifiables (PII) de vos clients en clair. Vous venez de prouver que vous êtes capable de développer une alternative conforme.

À propos de l'auteur

À propos de Tahiya Chowdhury

Tahiya Chowdhury est chef de produit pour Actian Zen et VectorAI DB, où elle dirige la stratégie de la plateforme de données en périphérie la plus robuste du secteur. Forte de son expérience chez MongoDB et Goldman Sachs, Tahiya est spécialisée dans la conception de produits qui concilient les besoins des grandes entreprises et la rapidité de développement moderne. Elle s'attache avec passion à simplifier les infrastructures de données, afin de permettre aux équipes d'ingénieurs de passer plus rapidement du prototype à la mise en production.