Korrektiver RAG

Wenn die Abrufung nicht ausreicht

Das RAG-System, das Sie aufgebaut haben, beruht auf einer impliziten Annahme: Die Abschnitte, die der Frage am ähnlichsten sind, sind auch die richtigen. Diese Annahme gilt oft, aber nicht immer. Wenn Sie den Index zu einem Thema abfragen, das er nicht abdeckt, gibt FAISS trotzdem die fünf nächsten Vektoren zurück, egal wie wenig zum Thema sie passen – ein Ähnlichkeitsmotor weiß nicht, wie er sagen soll: „Ich habe nichts.” Das Modell erhält dann ungeeigneten Kontext und da der Prompt es auffordert, sich darauf zu stützen, produziert es eine Antwort, die autoritär wirkt, aber keine Grundlage hat. Dies ist die heimtückischste Halluzinationssituation, da die technische Kette ohne Fehler funktioniert hat.

Korrektiver RAG (CRAG) fügt einen Bewertungsschritt zwischen Abruf und Generierung ein. Eine LLM untersucht jedes abgerufene Dokument und entscheidet: relevant oder nicht. Wenn das Urteil negativ ausfällt, generiert das System nicht einfach mit dem, was es hat – es schaltet auf eine Web-Suche um, um die Informationen zu finden, die im Korpus fehlen. Der Ablauf hat daher vier Stufen: Standardabruf aus dem Vektorspeicher, Bewertung jedes Dokuments gegenüber der Frage, Entscheidung, ob eine Web-Suche ausgelöst werden soll, falls mindestens ein Dokument als irrelevant eingestuft wird, und schließlich Generierung mit den besten verfügbaren Dokumenten.

Der Dokumenten-Bewerter

Alles hängt vom Bewerter ab. Der Trick besteht darin, die LLM nicht nach einer freien Einschätzung zu fragen, sondern nach einem binären Urteil im JSON-Format: {"score": "yes"} oder {"score": "no"}, nichts anderes. Der Parameter response_format={"type": "json_object"} garantiert eine parsierbare Ausgabe, und temperature=0 eliminiert Variabilität – das gleiche Frage/Dokumenten-Paar muss immer dasselbe Urteil erhalten, sonst wird das Verhalten Ihrer Pipeline nicht reproduzierbar.

from mistralai.client import Mistral
import json

client = Mistral(api_key=api_key)

def grade_document(question, document):
    """Evaluate whether a document is relevant to a question."""
    prompt = f"""You are a relevance evaluator. Given a question
and a document, determine if the document contains relevant information
to answer the question.

Respond ONLY with a JSON object: {{"score": "yes"}} or {{"score": "no"}}

Question : {question}
Document : {document}

JSON Evaluation:"""

    response = client.chat.complete(
        model="mistral-large-latest",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        response_format={"type": "json_object"},
    )

    result = json.loads(response.choices[0].message.content)
    return result.get("score", "no") == "yes"

# Test
question = "What are the steps of RAG?"
doc_pertinent = "RAG breaks down into retrieval, augmentation, and generation."
doc_non_pertinent = "The tiramisu recipe requires mascarpone."

print(grade_document(question, doc_pertinent))      # True
print(grade_document(question, doc_non_pertinent))   # False

Der Test am Ende des Blocks verdient Aufmerksamkeit. Das Tiramisu-Rezept illustriert genau, was die Vektorähnlichkeit durchschlüpfen lässt: Wenn Ihr Korpus nur über Kochen spricht und Sie ihn nach RAG abfragen, wird dieser Abschnitt aufgrund des Mangels an Konkurrenz an erster Stelle rangieren. Der Bewerter vergleicht jedoch keine Vektoren, sondern liest den Inhalt und lehnt ihn ab.

Angewendet auf alle abgerufenen Abschnitte liefert dieses Urteil zwei Informationen: die gefilterte Liste der zu behaltenden Dokumente und ein Boolean, das angibt, ob durch das Internet ergänzt werden muss. Die hier gewählte Schwelle ist absichtlich streng – ein einziges verworfenes Dokument reicht aus, um die Fallback-Lösung auszulösen, unter der Annahme, dass der Kontext dann unvollständig ist.

def grade_documents(question, retrieved_chunks):
    """Evaluate and filter retrieved documents."""
    relevant_docs = []
    irrelevant_count = 0

    for chunk in retrieved_chunks:
        is_relevant = grade_document(question, chunk["text"])
        if is_relevant:
            relevant_docs.append(chunk)
        else:
            irrelevant_count += 1

    need_web_search = irrelevant_count > 0

    print(f"  Relevant: {len(relevant_docs)}/{len(retrieved_chunks)}")
    print(f"  Web search needed: {need_web_search}")

    return relevant_docs, need_web_search

Das Sicherheitsnetz für das Web

Wenn lokale Dokumente nicht ausreichen, sucht das System anderswo. Mistral stellt dafür ein integriertes Werkzeug zur Verfügung, web_search, das Sie über die Conversations-API aktivieren: Das Modell führt die Suche selbst durch und schreibt seine Antwort unter Angabe der Quellen. Dieser Detailaspekt ist wichtig, da ein einfacher client.chat.complete-Aufruf das Web nicht durchsuchen kann: Das Modell würde aus dem Gedächtnis antworten, mit dem Risiko, genau das zu erfinden, was CRAG vermeiden will. Und seit SDK-Version 3.0 lehnen Chat-Completions das web_search-Werkzeug tatsächlich ab.

Der entscheidende Designpunkt ist das Rückgabeformat: Die Funktion gibt eine Liste von Wörterbüchern zurück, die dieselbe Form wie Ihre lokalen Abschnitte haben, mit einem source-Feld, das auf web_search gesetzt ist, und fügt die zitierten URLs hinzu. Diese Einheitlichkeit ermöglicht es nachgelagerten Schritten, die Herkunft der Dokumente vollständig zu ignorieren, während der Pfad für den Endbenutzer erhalten bleibt.

def web_search_fallback(question):
    """Web search fallback when local documents are insufficient."""
    response = client.beta.conversations.start(
        model="mistral-medium-latest",
        inputs=question,
        instructions="Answer factually, relying on web search results.",
        tools=[{"type": "web_search"}],
        store=False,  # no need to keep the conversation on Mistral's side
    )

    texte, urls = "", []
    for entry in response.outputs:
        if entry.type != "message.output":
            continue  # ignore the entry describing tool execution
        if isinstance(entry.content, str):
            texte += entry.content
            continue
        for chunk in entry.content:
            if chunk.type == "text":
                texte += chunk.text
            elif chunk.type == "tool_reference" and chunk.url:
                urls.append(chunk.url)

    # Return as a "web" document, with cited sources
    return [{
        "text": texte,
        "source": "web_search",
        "urls": urls,
    }]

Die Antwort enthält zwei Arten von Einträgen: einen tool.execution-Eintrag, der die durchgeführte Suche beschreibt, und einen message.output-Eintrag, dessen Inhalt Textfragmente und Referenzen (tool_reference) abwechselnd enthält, die Titel und URL der konsultierten Seiten tragen. Ein Administrator einer Organisation kann abgefragte Domains über eine Whitelist oder Blacklist einschränken; gefilterte Ergebnisse verschwinden ohne Warnung, sodass eine zu enge Whitelist die Suche ohne Ergebnisse zurücklassen kann.

Offizielle Quelle: Websearch (Mistral-Dokumentation)

Die zusammengebaute Pipeline

Die CorrectiveRAG-Klasse wiederholt die Standard-Pipeline-Struktur, indem sie die Bewertung einfügt. Der Standardwert k=4 ist etwas großzügiger als die üblichen drei Abschnitte, was Sinn ergibt: Da einige Abschnitte durch die Filterung verworfen werden, ist es besser, mit einem Puffer zu beginnen. Die ask-Methode orchestriert die vier Stufen und zeigt jede davon an, sodass Sie die Schlussfolgerungen des Systems in Echtzeit verfolgen können – wie viele Dokumente die Bewertung überstanden haben, ob der Fallback ausgelöst wurde. Das zurückgegebene Wörterbuch stellt auch web_search_used bereit: Dies ist ein wertvoller Indikator, der in der Produktion überwacht werden sollte, da eine abnorm hohe Web-Nutzungsrate kein CRAG-Fehler signalisiert, sondern eine Lücke in Ihrem Korpus.

class CorrectiveRAG:
    """Corrective RAG pipeline with evaluation and web fallback."""

    def __init__(self, chunks, embeddings, index):
        self.chunks = chunks
        self.embeddings = embeddings
        self.index = index

    def retrieve(self, question, k=4):
        """Step 1: standard retrieval."""
        response = client.embeddings.create(
            model="mistral-embed",
            inputs=[question],
        )
        query_vec = np.array([response.data[0].embedding], dtype=np.float32)
        distances, indices = self.index.search(query_vec, k)

        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx >= 0:
                results.append({
                    "text": self.chunks[idx]["text"],
                    "source": self.chunks[idx].get("source", "local"),
                })
        return results

    def grade(self, question, documents):
        """Step 2: evaluate relevance."""
        relevant = []
        for doc in documents:
            if grade_document(question, doc["text"]):
                relevant.append(doc)
        need_web = len(relevant) < len(documents)
        return relevant, need_web

    def generate(self, question, documents):
        """Step 3: generate the answer."""
        context = "\n\n---\n\n".join([d["text"] for d in documents])
        prompt = f"""Context :\n{context}\n\nQuestion : {question}"""

        response = client.chat.complete(
            model="mistral-large-latest",
            messages=[
                {"role": "system", "content": "Answer based on the context."},
                {"role": "user", "content": prompt},
            ],
            temperature=0.1,
        )
        return response.choices[0].message.content

    def ask(self, question, k=4):
        """Full pipeline."""
        print(f"\n=== Question : {question} ===")

        # 1. Retrieval
        print("1. Retrieval...")
        docs = self.retrieve(question, k=k)

        # 2. Grading
        print("2. Grading...")
        relevant_docs, need_web = self.grade(question, docs)

        # 3. Web search fallback if needed
        if need_web:
            print("3. Fallback web search...")
            web_docs = web_search_fallback(question)
            relevant_docs.extend(web_docs)

        # 4. Generation
        print("4. Generation...")
        answer = self.generate(question, relevant_docs)

        return {
            "answer": answer,
            "sources_used": len(relevant_docs),
            "web_search_used": need_web,
        }

# Usage
crag = CorrectiveRAG(chunks, embeddings, index)
result = crag.ask("How does Mistral implement embeddings?")

print(f"\nAnswer : {result['answer']}")
print(f"Web search used : {result['web_search_used']}")

Diese Zuverlässigkeit hat einen Preis, den Sie bewusst akzeptieren müssen: Mit k=4 löst jede Frage vier Bewertungsaufrufe aus, noch bevor die Generierung beginnt. Die vom Nutzer wahrgenommene Latenz steigt entsprechend an, ebenso wie die Rechnung. Der Kompromiss variiert je nach Anwendungsfall – für einen Rechts- oder Medizin-Assistenten ist die Verifizierung unverzichtbar; für einen internen Suchmotor, bei dem der Nutzer die Quellen sieht und selbst urteilt, ist er weniger kritisch.

Kernaussagen

• Korrektiver RAG fügt einen Relevanzbewertungsschritt nach dem Abruf hinzu
• Eine LLM dient als „Bewerter”, um jedes Dokument zu beurteilen: relevant oder nicht
• Wenn Dokumente als irrelevant eingestuft werden, wird ein Fallback auf die Web-Suche ausgelöst
• Dieser Ansatz reduziert Halluzinationen aufgrund ungeeigneten Kontexts erheblich
• Die Bewertung erhöht die Latenz, verbessert aber die Zuverlässigkeit der Antworten stark