So richten Sie ein mandantenfähiges RAG für den Kundensupport ein

Support-Tickets enthalten Kundendaten, darunter Namen, E-Mail-Adressen, Kontodaten und Nutzungsmuster. Gemäß der Datenschutz-Grundverordnung (DSGVO) und dem California Consumer Privacy Act (CCPA) ergeben sich aus diesen Daten Verpflichtungen für Auftragsverarbeiter.

Das Einbinden dieser Tickets in einen Cloud stellt eine Verarbeitungsmaßnahme dar. Dies erfordert eine gültige Rechtsgrundlage, die Einhaltung der Anforderungen an den Datenaufbewahrungsort sowie eine ausdrückliche vertragliche Genehmigung.

Die meisten Plattform-Ingenieure, die interne KI-Tools entwickeln, stoßen auf dasselbe Hindernis. Ein Cloud RAG-Prototyp (Retrieval Augmented Generation) funktioniert in der Staging-Umgebung zwar gut, doch Compliance-Anforderungen verzögern Deployment. Ein Kunde verlangt möglicherweise, dass Daten im Rahmen einer Datenverarbeitungsvereinbarung (DPA) im EWR verbleiben; ein Prüfer stellt möglicherweise die Frage, wohin Kundeninformationen während der KI-Verarbeitung fließen; oder Rechtsabteilungen melden möglicherweise Tickets, die regulierte Kontonummern enthalten. In jedem Fall entsteht dasselbe Hindernis: Kundendaten verlassen die interne Infrastruktur.

Unternehmen, die in ihrer Support-Wissensdatenbank eine semantische Suche einsetzen, verzeichnen um 40 bis 60 %schnellere Lösungszeiten, wenn die Support-Mitarbeiter darauf zugreifen können. RAG-Systeme kombinieren die Retrieval-Funktion (Suche nach relevantem Kontext) mit der Generierungsfunktion (KI-gestützte Antworten) und eignen sich daher ideal für den Kundensupport. Die Leistungsvorteile liegen auf der Hand. Die Architektur muss lediglich dafür sorgen, dass die Daten lokal und sicher gespeichert werden.

Dieses Tutorial zeigt, wie Sie mit Actian VectorAI DB ein mandantenfähiges RAG-System aufbauen können, bei dem Kundendaten innerhalb der lokalen Infrastruktur verbleiben. VectorAI DB unterstützt Metadaten über seine Filter-API und gewährleistet dabei isolation mandantenbezogene isolation Anwendungsebene. Das System bettet Tickets lokal ein, speichert sie zusammen mit den erforderlichen Kunden-ID-Feldern und führt Abfragen durch, die die Mandantengrenzen einhalten. Es verlassen keine personenbezogenen Daten (PII) den Netzwerkperimeter.

Drei Wege, wie Cloud Haftungsrisiken mit sich bringt

Cloud verursacht bei der Verarbeitung von Kundensupportdaten drei kritische Fehlerarten:

1. Support-Tickets enthalten personenbezogene Daten, die Cloud nicht angemessen schützen können. Gemäß der DSGVO und dem CCPA kann die Einbindung dieser Tickets in einen Cloud als Datenverarbeitung gelten, was rechtliche Anforderungen und Anforderungen an den Datenaufbewahrungsort nach sich zieht, die über den typischen Geltungsbereich bestehender BAA- oder SOC-2-Vereinbarungen hinausgehen.
2. Der Aufruf der Einbettungs-API übermittelt Daten und erfordert dieselben Datenschutzmaßnahmen wie jede andere externe API-Anfrage. Wenn Ticket-Text zur Einbettung an die Server von OpenAI gesendet wird, werden personenbezogene Daten an einen Dritten übermittelt.
3. Bei Multi-Tenant-Architekturen besteht die Gefahr, dass Kundendaten durch falsch konfigurierte Filter oder API-Fehler offengelegt werden. Konnektor AI Konnektor von Asana Konnektor aufgrund eines unzulänglichen Zugriffsumfangs auf das Tool sensible Daten organisationsübergreifend Konnektor , was zu vertraglichen Haftungsrisiken geführt hat, die über SLA des Anbieters hinausgingen.
   Die Lösung ist architektonischer Natur: Embeddings sollten On-Premises verbleiben, Kunden sollten auf Metadaten isoliert werden und sensible Daten dürfen niemals an externe Dienste weitergeleitet werden. Dieses Tutorial zeigt, wie sich diese Architektur mithilfe von VectorAI DB aufbauen lässt.

Datenflusspfeile, die die Infrastrukturgrenze überschreiten (Cloud) vs. innerhalb der Infrastruktur verbleiben (On-Prem)

Da Sie nun wissen, warum Cloud für den Kundensupport ungeeignet ist, wollen wir eine konforme Alternative entwickeln.

Was Sie entwickeln

In diesem Tutorial wird ein dreistufiges System aufgebaut, das vollständig auf Ihrer Infrastruktur läuft. Support-Mitarbeiter stellen Fragen; das System durchsucht ausschließlich die Ticket-Historie des jeweiligen Kunden sowie die freigegebenen Artikel der Wissensdatenbank (KB); und ein lokales LLM generiert zitierte Antworten mit Ticketnummern und KB-Verweisen. Es verlassen keine Kundendaten das interne Netzwerk.

Ebene 1: Erfassung

Das System nimmt Support-Tickets und Knowledge-Base-Artikel separat auf. Das System unterteilt die Tickets in Segmente (512 Token, 50-Token-Überlappung) und bettet sie lokal ein mithilfe von all-MiniLM-L6-v2und speichert sie unter der customer_id als obligatorisches Metadaten Metadaten .

Das Erfassungsschema umfasst:

• source_type: „Ticket“ oder „knowledge_base“
• customer_id: Obligatorisch für Tickets, null für KB-Artikel
• Produktlinie: Produktbereich, auf den sich das Ticket bezieht
• ticket_status: Offen, geschlossen, eskaliert
• created_date: Zeitstempel für die chronologische Filterung

Jeder Ticket-Ausschnitt enthält diese Felder sowie den Textinhalt und dessen Vektor-Einbettung. Artikel aus der Wissensdatenbank durchlaufen denselben Einbettungsprozess, wobei der Quelltyp auf „knowledge_base““ festgelegt ist und keine Kunden-ID erforderlich ist. Diese Blöcke sind für alle Kunden bei der Suche zugänglich.

Ebene 2: Abfrage

Wenn ein Support-Mitarbeiter eine Frage zur Historie von Kunde A stellt, wendet die abfragen vor der semantischen Suche eine Filterung nach Kunden-ID an. Der Anwendungscode erstellt Filter, die die Ergebnisse auf den angegebenen Kunden beschränken. Die Filter-API von VectorAI DB wendet diese Filter an, doch isolation davon isolation , dass die Anwendung sie bei jeder abfragen korrekt übergibt. 

Die Hybrid-Suche kombiniert die customer_id mit semantischem Abgleich. Das System liefert relevante Ausschnitte aus den Tickets dieses Kunden sowie aus Artikeln der gemeinsamen Wissensdatenbank. Die Ergebnisse enthalten Ähnlichkeitswerte und Quellenangaben.

Ein lokales LLM (Ollama mit llama3.2:3b oder mistral:7b) generiert Antworten ausschließlich anhand des abgerufenen Kontexts. Die RAG-Eingabeaufforderung weist das Modell an, Quellen anzugeben und zu sagen, dass es die Antwort nicht weiß, wenn kein Kontext vorhanden ist. Die Generierung durch ein lokales LLM dauert in der Regel 2–5 Sekunden, abhängig von der Kontextgröße und der Hardware.

Ebene 3: isolation zwischen verschiedenen Mandanten

  Verhindert den Verlust von kundenübergreifenden Daten, wenn der Anwendungscode Filter korrekt implementiert. Alle Kunden nutzen eine gemeinsame Sammlung (support_data), wobei die Trennung der Kunden durch customer_id , die zum abfragen übergeben werden, sichergestellt.

VectorAI DB führt diese Filter über seine Filter-API aus. Allerdings wenn der Anwendungscode einen Fehler enthält und die Filter „customer_id“ auslässt, gibt die Datenbank alle Kundendaten zurück. Schema-Regeln verhindern keine uneingeschränkten Abfragen. isolation wird durch den Anwendungscode erzwungen, nicht durch die Datenbank.

Die Artikel der Wissensdatenbank stehen allen Kunden gemeinsam zur Verfügung. Die Ticketdaten sind logisch voneinander getrennt. Eine abfragen Kunde A sollte Filter erstellen, um folgende Daten abzurufen:

• Alle Chunks mit customer_id = „A“ UND source_type = „ticket“.
• Alle Chunks mit source_type = „knowledge_base“ (ohne Kundenfilter).

Wenn Filter korrekt angewendet werden, abfragen eine abfragen Ticket-Teile von Kunde B zurückgeben. Die Anwendungsschicht ist für Beständigkeit alle abfragen Beständigkeit verantwortlich.

Architektonische Alternativen:

• Echte isolation auf Sammlungsebene: Erstellen Sie pro Kunde eine Sammlung (z. B. customer_a_tickets, customer_b_tickets). Dadurch entsteht eine klare Abgrenzung, die verhindert, dass bei der Suche nach Daten von Kunde A Abfragen an die Sammlung von Kunde B gesendet werden – selbst bei fehlerhaftem Code. Der Nachteil ist ein zusätzlicher betrieblicher Aufwand.
• isolation vonMetadaten: Verwenden Sie eine einzige gemeinsam genutzte Sammlung mit anwendungsseitig erzwungenen Filtern. Dies ist einfacher zu handhaben, erfordert jedoch eine disziplinierte Filtererstellung und Codeüberprüfung, um Lecks zu vermeiden.

In diesem Tutorial wird isolation Metadaten verwendet, isolation sie für Teams, die Standards für Code-Reviews und Tests durchsetzen können, ein ausgewogenes Verhältnis zwischen Sicherheit und einfacher Handhabung bietet.

Hardware-Basisvoraussetzungen: Dieses Tutorial läuft auf einer Instanz mit 16 GB RAM. Tests mit ähnlichen Workloads ergaben abfragen von unter 500 ms auf Standardhardware. 

Dreischichtige Architektur (Erfassung → Speicherung → abfragen) mit einer Ebene zur Durchsetzung der Kunden-ID

Weitere Informationen zur Messung der Wirksamkeit dieses Systems finden Sie unter „So messen Sie die Leistung des RAG-Systems“.

Bau es!

Jeder der folgenden Codeblöcke ist so konzipiert, dass er ausgeführt werden kann. Der Code wurde mit der VectorAI-Datenbank getestet und lässt sich nach der Installation der Abhängigkeiten ohne Änderungen ausführen.

Voraussetzungen

• Docker und Docker Compose
• Python .10 oder höher
• UV-Paketmanager installiert (curl -LsSf https://astral.sh/uv/install.sh | sh)
  

Python (auf dem Host installieren):

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

Schritt 1: VectorAI DB bereitstellen

Was dabei herauskommt: Eine laufende VectorAI-DB-Instanz mit persistenter Speicherung für Ticket- und Wissensdatenbankdaten.

# 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

Die Volume-Mounts verstehen:

• ./data:/app/data – VectorAI DB speichert seine Datenbankdateien unter /app/data innerhalb des Containers, was auf ./data/ auf Ihrem Host entspricht.
• ./audit_logs:/app/audit_logs – Wurde für die zukünftige Verwendung angelegt, falls Sie Audit-Skripte innerhalb des Containers ausführen.

Wichtig: In diesem Tutorial werden alle Python auf Ihrem Host-Rechnerausgeführt, nicht innerhalb des Containers. Die Skripte schreiben Audit-Protokolle in den Ordner ./audit_logs/queries.jsonl auf dem Host. Die Einbindung des Docker-Volumes ist für dieses Tutorial optional, wird jedoch für Teams bereitgestellt, die später Python innerhalb des Containers ausführen möchten.

Starten Sie die Datenbank:

Ausführen: 

docker-compose up -d

Docker-Start

Schritt 2: Anonymisierung personenbezogener Daten vor der Erfassung

Dieser Schritt sorgt dafür, dass saubere Ticketdaten, aus denen personenbezogene Daten entfernt wurden, in Ihre Vektordatenbank gelangen, bevor die Verarbeitung beginnt.

Häufige Muster von personenbezogenen Daten, die gefiltert werden sollen:

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

Anonymisierung personenbezogener Daten

Schritt 3: Aufbau der Pipeline zur Ticket-Erfassung

Das Ergebnis: Aufbereitete Kundensupport-Tickets, die in Blöcke unterteilt, eingebettet und zusammen mit Metadaten erforderlichen Metadaten zur Kunden-ID in der VectorAI-Datenbank gespeichert werden. 

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

Ausführen: 

uv run python ticket_ingestion.py

Ausgabe der Ticket-Erfassung mit Angabe der Chunks pro Kunde

Schritt 4: Aufbau der KB-Erfassungspipeline

Was dabei herauskommt: Wissensdatenbank-Artikel, eingebettet gespeichert mit dem Quelltyp „knowledge_base“, die für alle Kunden zugänglich sind.“, die für alle Kunden zugänglich sind.

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

Ausführen:  

uv run python kb_ingestion.py

Ausgabe der Wissensdatenbank-Erfassung

Schritt 5: Gemeinsames Suchmodul erstellen:

Ergebnis: Der nächste Schritt besteht darin, ein gemeinsames Suchmodul zu erstellen, das eine wiederverwendbare Suchfunktion für alle nachfolgenden Skripte bereitstellt. Dieses Modul verhindert die doppelte Verwendung von Funktionen. Die nächsten vier Schritte (abfragen,Demo, RAG-System und Audit-Protokollierung) importieren alle search_customer_tickets() aus diesem gemeinsam genutzten Modul, anstatt dessen Logik zu duplizieren.

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

Schritt 6: Abfragen auf Kundenebene ausführen

Was dabei herauskommt: Suchergebnisse, die auf die Tickets eines bestimmten Kunden sowie auf gemeinsam genutzte KB-Artikel gefiltert sind.

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

Ausführen:  

uv run python query.py

Beachten Sie, wie abfragen.py jetzt prägnanter ist. Es importiert die Suchfunktion aus search.py , anstatt die Suchlogik zu duplizieren. Das gleiche Muster zieht sich durch die nächsten drei Schritte.

Schritt 7: Demonstrieren Sie isolation zwischen Mandanten isolation drei Testfällen

Was dabei herauskommt: Umfassender Nachweis, dass die Filterung anhand der Kunden-ID in allen drei Szenarien den datenübergreifenden Datenverlust verhindert.

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

Ausführen:  

uv run python isolation_demo.py

Ergebnisse Isolation

Schritt 8: Das lokale LLM anschließen

Was dabei herauskommt: Ein durchgängiges RAG-System, das mithilfe von Ollama zitierte Antworten generiert.

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

Ausführen:  

uv run python rag_system.py

Schritt 9: Protokollierung der Überwachung konfigurieren

Was dabei herauskommt:Protokoll vollständiges Protokoll abfragen, Protokoll lokal mit vollständiger Rückverfolgbarkeit protokolliert wird.

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

Ausführen:  

uv run python audit.py

Protokoll-Einträge mit Nutzer Kunden-IDs

Schritt 10: Ticket-Aktualisierungen bearbeiten

Was dabei herauskommt: Schrittweise Erfassung neuer Tickets, ohne die gesamte Sammlung neu aufbauen zu müssen.

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

Sie haben soeben ein privates, mandantenfähiges RAG-System eingerichtet

Sie haben ein produktionsbereit RAG-System auf Basis der VectorAI DB aufgebaut, bei dem die personenbezogenen Daten Ihrer Kunden innerhalb Ihrer Infrastruktur verbleiben. isolation der Mandanten isolation durch Metadaten . Es verlassen keine Kundendaten Ihre Netzwerkgrenzen. Wenn Aufsichtsbehörden fragen, wo sich die Embeddings befinden, lautet die Antwort: „In unserer Infrastruktur.“

Produktionsbetriebe, Gesundheitsdienstleister und Finanzdienstleister nutzen dieses Muster, um sensible Datensätze ohne Cloud zu durchsuchen. Die regulatorischen Framework je nach DSGVO, HIPAA oder PCI-DSS, doch das Deployment bleibt dasselbe: Die Einbettungen bleiben lokal, Daten werden niemals an externe Dienste weitergeleitet.

Erweitern Sie dies für den Produktiveinsatz um semantisches Routing, das Anfragen mit geringer Zuverlässigkeit eskaliert, kundenübergreifende Analysen zum Schutz personenbezogener Daten, Feedbackschleifen zur Verbesserung der Suchqualität sowie inkrementelles Lernen aus gelösten Tickets.

Testen Sie VectorAI DB anhand Ihrer Daten mithilfe des GitHub-Repositorys. Entscheiden Sie vor der Produktionsbereitstellung, ob Sie RAG oderfeinabstimmen verwenden möchten und wann Sie sich für On-Premises Cloud entscheiden sollten. Treten Sie der Actian-Community auf Discord bei, wo Plattform-Ingenieure Deployment austauschen.

Die Daten Ihrer Kunden verdienen Besseres als Cloud , bei denen die Datenschutzbestimmungen den Anbieter schützen, nicht aber den Kunden. SaaS-Anbieter verschlüsseln Daten während der Übertragung, doch in ihren eingebetteten Modellen werden die personenbezogenen Daten Ihrer Kunden weiterhin im Klartext angezeigt. Sie haben gerade bewiesen, dass Sie eine konforme Alternative entwickeln können.

Über den Autor

Über Tahiya Chowdhury

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