Ankererkennung für RAG: Parallele Detektoren, dann ein LLM-Aufruf am Ende

Ankererkennung für RAG: Parallele Detektoren, dann ein einziger LLM-Aufruf am Ende

Retrieval-Augmented Generation (RAG) hat sich zur De-facto-Architektur entwickelt, um die Ausgaben großer Sprachmodelle (LLMs) in externem Wissen zu verankern. Eine hartnäckige Herausforderung bleibt jedoch bestehen: Wie stellt man sicher, dass der abgerufene Kontext zuverlässige, relevante „Anker“ enthält, bevor das LLM eine Antwort generiert? Traditionelle Ansätze rufen das LLM mehrfach auf – einmal für den Abruf, einmal für die Verifikation und einmal für die finale Generierung –, was zu Latenz und Kosten führt. Ein effizienteres Paradigma zeichnet sich ab: Führen Sie parallele, leichtgewichtige Detektoren aus, um Anker in den abgerufenen Dokumenten zu identifizieren, und tätigen Sie dann am Ende einen einzigen LLM-Aufruf zur Generierung. Dieser Artikel untersucht den Ansatz der Ankererkennung, bietet eine konkrete Implementierung und zeigt, wie man mit Open-Source-Tools ein produktionsreifes System aufbaut.

Ankererkennung in RAG verstehen

Ein Anker in RAG ist ein abgerufener Textabschnitt, der als zuverlässige Grundlage für die Antwort des LLM dient. Es handelt sich um eine Tatsache, eine Statistik, ein Zitat oder eine logische Aussage, die überprüft und direkt bei der Generierung verwendet werden kann. Die Ankererkennung ist der Prozess, diese Teile zu identifizieren, bevor das LLM den Kontext sieht. Die entscheidende Erkenntnis ist, dass viele potenzielle Anker verrauscht sind – sie können irrelevant, widersprüchlich oder veraltet sein. Durch den Einsatz paralleler Detektoren können Sie Anker filtern und bewerten, ohne das LLM aufzurufen, und den LLM-Aufruf für die finale, ausgefeilte Ausgabe reservieren.

Dieser Ansatz passt zu aktuellen Trends im Design von KI-Systemen, bei denen spezialisierte, leichtgewichtige Modelle (z. B. Einbettungsmodelle, Klassifikatoren oder kleine Transformatoren) die Vorverarbeitung übernehmen, während große generative Modelle sparsam eingesetzt werden. Große Branchenakteure wie OpenAI, Google und Microsoft haben Forschungsergebnisse zur Reduzierung von LLM-Aufrufen durch effiziente Vorverarbeitungspipelines veröffentlicht, obwohl die spezifischen Implementierungsdetails variieren. Der Ansatz der parallelen Detektoren ist eine praktische Synthese dieser Ideen.

Voraussetzungen

Um dieser Anleitung zu folgen, benötigen Sie:

• Python 3.9 oder höher
• Einen Rechner mit mindestens 8 GB RAM (16 GB werden für den Betrieb mehrerer Detektoren empfohlen)
• Grundlegende Vertrautheit mit Python und Befehlszeilenwerkzeugen
• Die folgenden Python-Pakete: `transformers`, `sentence-transformers`, `torch`, `faiss-cpu`, `pandas`, `numpy`, `openai` (oder ähnliches für den LLM-Zugriff)

Sie benötigen außerdem einen LLM-API-Schlüssel (z. B. von OpenAI) für den finalen Generierungsschritt. Die Ankerdetektoren selbst laufen lokal.

Schritt-für-Schritt-Installation

1. Erstellen Sie eine virtuelle Umgebung

Erstellen Sie zunächst eine isolierte Python-Umgebung, um Abhängigkeitskonflikte zu vermeiden.

python3 -m venv anchor-rag-env
source anchor-rag-env/bin/activate

Dieser Befehl erstellt eine virtuelle Umgebung namens `anchor-rag-env` und aktiviert sie.

2. Installieren Sie die Kernabhängigkeiten

Installieren Sie die Bibliotheken, die für die Detektoren und die LLM-Integration erforderlich sind.

pip install transformers sentence-transformers torch faiss-cpu pandas numpy openai

• `transformers` und `sentence-transformers` bieten vortrainierte Modelle für Textkodierung und -klassifikation.
• `torch` ist das PyTorch-Backend.
• `faiss-cpu` ermöglicht schnelle Ähnlichkeitssuche für die Ankerbewertung.
• `pandas` und `numpy` dienen der Datenmanipulation.
• `openai` wird für den finalen LLM-Aufruf verwendet (ersetzen Sie es durch Ihren gewählten LLM-Anbieter).

3. Überprüfen Sie die Installation

Testen Sie, ob die Umgebung korrekt eingerichtet ist, indem Sie eine kleine Prüfung durchführen.

python -c "import sentence_transformers; print('Installation OK')"

Wenn „Installation OK“ angezeigt wird, fahren Sie fort. Andernfalls überprüfen Sie die Fehler (z. B. fehlende CUDA-Treiber für torch – wenn Sie keine GPU haben, reichen die CPU-Only-Versionen aus).

Architekturüberblick

Das Ankererkennungssystem arbeitet in drei Phasen:

1. **Abruf** – Beschaffen Sie einen Satz von Kandidatendokumenten (z. B. aus einer Vektordatenbank oder einer Websuche). 2. **Parallele Detektoren** – Führen Sie mehrere leichtgewichtige Modelle gleichzeitig aus, um Anker zu bewerten und zu filtern:

• Ein Relevanzdetektor (z. B. ein BERT-basierter Klassifikator)
• Ein Faktizitätsdetektor (z. B. ein NLI-Modell)
• Ein Redundanzdetektor (z. B. Kosinus-Ähnlichkeit)
• Ein Entitätskohärenzdetektor (z. B. Named-Entity-Überlappung)

3. **Ein einziger LLM-Aufruf** – Aggregieren Sie die am besten bewerteten Anker und übergeben Sie sie als verdichteten Kontext an das LLM zur Generierung.

Dieses Design minimiert LLM-Aufrufe und maximiert gleichzeitig die Qualität des Eingabekontexts.

Implementierung: Aufbau der parallelen Detektoren

Detektor 1: Relevanzbewertung

Der Relevanzdetektor verwendet ein Sentence-Transformer-Modell, um die Kosinus-Ähnlichkeit zwischen der Abfrage und jedem abgerufenen Textabschnitt zu berechnen. Dies ist ein schneller, unüberwachter Schritt.

from sentence_transformers import SentenceTransformer, util
import numpy as np

# Laden eines leichtgewichtigen Einbettungsmodells
model = SentenceTransformer('all-MiniLM-L6-v2')

def relevance_detector(query, chunks):
    query_emb = model.encode(query, convert_to_tensor=True)
    chunk_embs = model.encode(chunks, convert_to_tensor=True)
    scores = util.cos_sim(query_emb, chunk_embs)[0].cpu().numpy()
    return scores

Diese Funktion gibt ein numpy-Array mit Relevanzwerten (0 bis 1) für jeden Abschnitt zurück. Hohe Werte zeigen eine starke semantische Übereinstimmung mit der Abfrage an.

Detektor 2: Faktizitätsprüfung

Ein Faktizitätsdetektor verwendet ein Natural Language Inference (NLI)-Modell, um zu überprüfen, ob jeder Abschnitt logisch konsistente Aussagen enthält. Wir verwenden ein vortrainiertes NLI-Modell von Hugging Face.

from transformers import pipeline

nli_pipeline = pipeline("text-classification", model="roberta-large-mnli")

def factuality_detector(chunks):
    # Für jeden Abschnitt prüfen, ob er durch sich selbst "impliziert" wird (eine Heuristik für interne Konsistenz)
    scores = []
    for chunk in chunks:
        result = nli_pipeline(chunk[:512])  # Auf Modell-Maximallänge kürzen
        # "ENTAILMENT"-Score als Proxy für Faktizität verwenden
        if result[0]['label'] == 'ENTAILMENT':
            scores.append(result[0]['score'])
        else:
            scores.append(0.0)
    return np.array(scores)

Dies ist eine vereinfachte Heuristik. In der Produktion würden Sie Abschnitte mit einer vertrauenswürdigen Wissensdatenbank vergleichen oder ein spezielles Faktizitätsmodell verwenden. Der entscheidende Punkt ist, dass es parallel zu anderen Detektoren läuft.

Detektor 3: Redundanzfilter

Redundanz wird durch die Berechnung der paarweisen Kosinus-Ähnlichkeit zwischen den Abschnitten erkannt. Abschnitte, die einander zu ähnlich sind, werden herabgestuft, um doppelte Informationen zu vermeiden.

def redundancy_detector(chunks, threshold=0.85):
    chunk_embs = model.encode(chunks, convert_to_tensor=True)
    sim_matrix = util.cos_sim(chunk_embs, chunk_embs).cpu().numpy()
    scores = np.ones(len(chunks))
    for i in range(len(chunks)):
        # Zählen, wie viele andere Abschnitte diesem sehr ähnlich sind
        redundant_count = np.sum(sim_matrix[i] > threshold) - 1
        scores[i] = 1.0 / (1.0 + redundant_count)  # Redundante Abschnitte bestrafen
    return scores

Ein eindeutiger Abschnitt erhält einen Wert von 1,0; ein Abschnitt, der mit drei anderen identisch ist, erhält einen Wert von 0,25.

Detektor 4: Entitätskohärenz

Dieser Detektor stellt sicher, dass die Abschnitte Entitäten (Personen, Orte, Daten) enthalten, die für die Abfrage relevant sind. Er verwendet ein einfaches Named-Entity-Recognition (NER)-Modell.

from transformers import pipeline

ner_pipeline = pipeline("ner", model="dslim/bert-base-NER", aggregation_strategy="simple")

def entity_coherence_detector(query, chunks):
    query_entities = set([e['word'].lower() for e in ner_pipeline(query)])
    scores = []
    for chunk in chunks:
        chunk_entities = set([e['word'].lower() for e in ner_pipeline(chunk[:512])])
        overlap = len(query_entities & chunk_entities) / max(len(query_entities), 1)
        scores.append(overlap)
    return np.array(scores)

Ein Wert von 1,0 bedeutet, dass alle Abfrageentitäten im Abschnitt vorkommen; 0,0 bedeutet keine Überlappung.

Paralleles Ausführen der Detektoren

Um alle Detektoren gleichzeitig auszuführen, verwenden Sie das Modul `concurrent.futures` von Python. Dies ist der Kern des Konzepts der „parallelen Detektoren“.

from concurrent.futures import ThreadPoolExecutor

def run_all_detectors(query, chunks):
    with ThreadPoolExecutor(max_workers=4) as executor:
        future_relevance = executor.submit(relevance_detector, query, chunks)
        future_factuality = executor.submit(factuality_detector, chunks)
        future_redundancy = executor.submit(redundancy_detector, chunks)
        future_entity = executor.submit(entity_coherence_detector, query, chunks)
        
        relevance_scores = future_relevance.result()
        factuality_scores = future_factuality.result()
        redundancy_scores = future_redundancy.result()
        entity_scores = future_entity.result()
    
    # Kombinieren der Werte (einfache gewichtete Summe)
    combined = (0.4 * relevance_scores +
                0.3 * factuality_scores +
                0.2 * redundancy_scores +
                0.1 * entity_scores)
    return combined

Passen Sie die Gewichte an Ihren Anwendungsfall an. Die kombinierten Werte werden verwendet, um die Top-k-Anker auszuwählen.

Ein einziger LLM-Aufruf am Ende

Nachdem die parallelen Detektoren die Abschnitte bewertet und gefiltert haben, rufen Sie die Top-k-Anker ab und übergeben sie in einem einzigen Aufruf an das LLM. Hier findet die finale Generierung statt.

import openai

openai.api_key = "your-api-key-here"

def generate_with_anchors(query, top_anchors, llm_model="gpt-4"):
    # Anker zu einem verdichteten Kontext zusammenfügen
    anchor_text = "\n\n".join([f"[Anchor {i+1}] {chunk}" for i, chunk in enumerate(top_anchors)])
    
    system_prompt = "Sie sind ein hilfreicher Assistent. Verwenden Sie nur die bereitgestellten Anker, um die Abfrage zu beantworten. Wenn die Anker nicht ausreichen, sagen Sie es."
    user_prompt = f"Abfrage: {query}\n\nAnker:\n{anchor_text}\n\nAntwort:"
    
    response = openai.ChatCompletion.create(
        model=llm_model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        max_tokens=500,
        temperature=0.3
    )
    return response.choices[0].message['content']

Diese Funktion tätigt genau einen LLM-Aufruf. Das LLM erhält nur die am besten bewerteten Anker, was die Token-Nutzung und das Halluzinationsrisiko reduziert.

Anwendungsbeispiel

Schritt 1: Beispieldaten vorbereiten

Angenommen, Sie haben drei Abschnitte aus einer Wissensdatenbank abgerufen.

query = "Was ist die Hauptstadt von Frankreich?"
chunks = [
    "Frankreich ist ein Land in Europa. Seine Hauptstadt ist Paris, das für den Eiffelturm bekannt ist.",
    "Paris ist die Hauptstadt von Frankreich und wird oft als Stadt des Lichts bezeichnet.",
    "Das Louvre-Museum in Paris beherbergt die Mona Lisa."
]

Schritt 2: Parallele Detektoren ausführen

combined_scores = run_all_detectors(query, chunks)
top_indices = np.argsort(combined_scores)[-2:][::-1]  # Top 2 Anker auswählen
top_anchors = [chunks[i] for i in top_indices]
print("Top-Anker:", top_anchors)

Die Ausgabe könnte sein:

Top-Anker: ['Frankreich ist ein Land in Europa. Seine Hauptstadt ist Paris, das für den Eiffelturm bekannt ist.', 'Paris ist die Hauptstadt von Frankreich und wird oft als Stadt des Lichts bezeichnet.']

Schritt 3: Finale Antwort generieren

answer = generate_with_anchors(query, top_anchors)
print(answer)

Ausgabe:

Die Hauptstadt von Frankreich ist Paris.

Es wurde nur ein LLM-Aufruf getätigt, und es wurden nur die beiden relevantesten, faktisch korrektesten und nicht-redundanten Abschnitte verwendet.

Leistungsaspekte

• **Latenz:** Parallele Detektoren laufen in etwa 0,5–2 Sekunden auf einer CPU (abhängig von der Anzahl der Abschnitte und der Modellgröße). Der LLM-Aufruf fügt 1–3 Sekunden hinzu. Die Gesamtzeit liegt typischerweise unter 5 Sekunden.
• **Kosten:** Sie zahlen für einen LLM-Aufruf pro Abfrage statt für drei oder vier. Die Detektoren laufen lokal ohne API-Kosten.
• **Genauigkeit:** Das gewichtete Bewertungssystem kann abgestimmt werden. Für domänenspezifische Anwendungen ersetzen Sie das generische NLI-Modell durch ein feinabgestimmtes Faktizitätsmodell.
• **Skalierbarkeit:** Verwenden Sie FAISS-Indizierung für große Abschnittsmengen (z. B. Tausende von Abschnitten), um die Relevanz- und Redundanzerkennung zu beschleunigen.

Fazit

Die Ankererkennung mit parallelen Detektoren, gefolgt von einem einzigen LLM-Aufruf, ist ein pragmatisches Entwurfsmuster für produktive RAG-Systeme. Es reduziert die Latenz, senkt die API-Kosten und verbessert die Ausgabequalität, indem verrauschter Kontext gefiltert wird, bevor das LLM ihn sieht. Die hier gezeigte Implementierung verwendet leichtgewichtige Open-Source-Modelle für Relevanz, Faktizität, Redundanz und Entitätskohärenz – alle laufen parallel. Der finale LLM-Aufruf generiert dann eine fundierte Antwort aus einem kuratierten Satz von Ankern. Dieser Ansatz ist inspiriert von Branchentrends hin zu effizienten, mehrstufigen KI-Pipelines, wie sie in der Forschung von OpenAI, Google und Microsoft zu sehen sind. Durch die Übernahme dieses Musters können Sie RAG-Systeme bauen, die sowohl schnell als auch zuverlässig sind, ohne auf die generative Kraft großer Sprachmodelle verzichten zu müssen.