Self-RAG
Auf dieser Seite
Selbstreflexion in RAG
Korrektives RAG steuert, was in den Prompt geht, nicht aber, was daraus kommt. Selbst perfekter Kontext garantiert keine gute Antwort: Das Modell könnte Details hinzufügen, die in den Dokumenten nicht vorkommen, oder Text produzieren, der zwar korrekt ist, aber die eigentliche Frage ausweicht. Self-RAG (Self-Reflektives RAG) schließt diese Lücke, indem es nach der Generierung zwei Prüfungen hinzufügt. Die erste prüft die faktische Fundierung – beruht die Antwort tatsächlich auf den bereitgestellten Dokumenten? Die zweite prüft die Nützlichkeit – beantwortet sie wirklich, was gefragt wurde? Scheitert eine der Prüfungen, schleift das System und versucht erneut, wodurch es sich zu einem Selbstkorrekturmechanismus statt zu einem einfachen Filter entwickelt.
Der Ablauf erweitert somit das korrektive RAG. Sie haben weiterhin die Abrufphase gefolgt von der Dokumentenbewertung und der Generierung. Zwei neue Knoten kommen danach: die Halluzinationsprüfung, die die Antwort mit den Dokumenten vergleicht, und die Antwortprüfung, die sie mit der Frage vergleicht. Scheitert eine Prüfung, kehrt das System entweder zur Generierung zurück oder holt frisches Material aus dem Web, je nachdem, welche Verifikation gescheitert ist.
Prüfung der faktischen Fundierung
Der erste Prüfer erhält zwei Eingaben: die zusammengefügten Dokumente und die generierte Antwort. Seine Aufgabe besteht nicht darin, zu beurteilen, ob die Antwort absolut wahr ist, sondern ob sie vollständig fundiert ist in dem, was ihm vorgelegt wurde. Diese Unterscheidung ist entscheidend: Eine zutreffende Aussage, die nicht in den Dokumenten steht, muss verworfen werden, da sie aus dem Trainingsgedächtnis des Modells stammt und nicht aus Ihrem Korpus – und es besteht keine Garantie, dass die nächste Aussage ebenfalls zutreffend ist. Das Ausgabeformat entspricht dem des Bewerters: ein binäres JSON-Objekt bei Temperatur 0.
import json
from mistralai.client import Mistral
client = Mistral(api_key=api_key)
def check_hallucination(documents, generation):
"""Check whether the generation is grounded in the documents."""
docs_text = "\n\n".join([d["text"] for d in documents])
prompt = f"""You are a factual grounding evaluator. Given a set of
documents and a generated answer, determine whether the answer is entirely
grounded in the information contained in the documents.
Respond ONLY with a JSON object: {{"grounded": "yes"}} or {{"grounded": "no"}}
Documents:
{docs_text}
Generated answer:
{generation}
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("grounded", "no") == "yes"Prüfung der Nützlichkeit
Der zweite Prüfer sieht weder die Dokumente noch die Abrufergebnisse – nur die Frage und die Antwort. Dieser eingeschränkte Blickwinkel ist absichtlich gewählt, da er einen Fehler isoliert, der der ersten Prüfung entgangen ist. Stellen Sie sich vor, jemand fragt, wie viele Dimensionen ein Embedding-Modell erzeugt, und die Antwort beschreibt ausführlich, wofür Embeddings verwendet werden, ohne die Zahl zu nennen. Alles ist in den Dokumenten fundiert, sodass der erste Prüfer „ja” sagt, und der Nutzer geht ohne seine Information davon. Der zweite Prüfer sagt „nein” und löst einen weiteren Versuch aus.
def check_answer_usefulness(question, generation):
"""Check whether the answer actually addresses the question."""
prompt = f"""You are an answer quality evaluator. Given a question
and an answer, determine whether the answer usefully and completely addresses the question.
Respond ONLY with a JSON object: {{"useful": "yes"}} or {{"useful": "no"}}
Question: {question}
Answer: {generation}
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("useful", "no") == "yes"Der Selbstkorrektur-Loop
Die SelfRAG-Klasse nutzt die vertrauten Schritte Abruf und Bewertung und wickelt diese in einen Wiederholungsloop ein, der durch max_retries begrenzt ist. Die Logik des Loops ist subtiler, als sie auf den ersten Blick erscheint, da die beiden Fehlertypen unterschiedlich behandelt werden. Eine erkannte Halluzination führt einfach zur Neugenerierung mit demselben Kontext: Das Problem liegt beim Modell, das zu weit gegriffen hat, und ein neuer Versuch reicht oft aus. Eine als unnütz bewertete Antwort löst jedoch eine zusätzliche Websuche vor dem erneuten Versuch aus: Hier liegt das Problem am Kontext, der zu dünn ist, um eine vollständige Antwort zu stützen, und eine Neugenerierung ohne Hinzufügung würde das gleiche Ergebnis liefern.
Beachten Sie den Fallback-Ausgang nach dem Loop. Wenn alle Versuche erschöpft sind, wirft das System keine Ausnahme auf: Es gibt die letzte Generierung zurück zusammen mit den Flags grounded und useful. Dies ist eine bewusste Designentscheidung – es ist besser, eine unvollkommene Antwort zurückzugeben, die so gekennzeichnet ist, damit Ihre Schnittstelle sie dem Nutzer anzeigen kann, als einen rohen Fehler zu werfen.
import numpy as np
class SelfRAG:
"""Self-RAG pipeline with self-correction."""
def __init__(self, chunks, embeddings, index, max_retries=2):
self.chunks = chunks
self.embeddings = embeddings
self.index = index
self.max_retries = max_retries
def retrieve(self, question, k=4):
"""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)
return [
{"text": self.chunks[idx]["text"], "source": self.chunks[idx].get("source", "")}
for dist, idx in zip(distances[0], indices[0]) if idx >= 0
]
def grade_documents(self, question, documents):
"""Evaluate document relevance."""
relevant = []
for doc in documents:
if grade_document(question, doc["text"]):
relevant.append(doc)
return relevant, len(relevant) < len(documents)
def generate(self, question, documents):
"""Generate an 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": "user", "content": prompt}],
temperature=0.2,
)
return response.choices[0].message.content
def ask(self, question, k=4):
"""Complete Self-RAG pipeline with self-correction."""
print(f"\n=== Self-RAG: {question[:60]}... ===")
# 1. Retrieval
print("1. Retrieval...")
docs = self.retrieve(question, k=k)
# 2. Grading
print("2. Grading documents...")
relevant_docs, need_web = self.grade_documents(question, docs)
if need_web:
print(" -> Fallback web search")
web_docs = web_search_fallback(question)
relevant_docs.extend(web_docs)
# 3. Generation loop with self-correction
for attempt in range(self.max_retries + 1):
print(f"3. Generation (attempt {attempt + 1})...")
generation = self.generate(question, relevant_docs)
# 4. Hallucination check
print("4. Checking hallucinations...")
is_grounded = check_hallucination(relevant_docs, generation)
if not is_grounded:
print(" -> HALLUCINATION detected, retrying...")
continue
# 5. Usefulness check
print("5. Checking usefulness...")
is_useful = check_answer_usefulness(question, generation)
if not is_useful:
print(" -> Answer NOT USEFUL, fallback web search...")
web_docs = web_search_fallback(question)
relevant_docs.extend(web_docs)
continue
# Success
print(" -> Answer validated!")
return {
"answer": generation,
"attempts": attempt + 1,
"grounded": True,
"useful": True,
}
# If all attempts fail
print(" -> Maximum attempts reached")
return {
"answer": generation,
"attempts": self.max_retries + 1,
"grounded": is_grounded,
"useful": is_useful,
}
# Usage
self_rag = SelfRAG(chunks, embeddings, index, max_retries=2)
result = self_rag.ask("What are the advantages of RAG over fine-tuning?")
print(f"\nAnswer: {result['answer']}")
print(f"Attempts: {result['attempts']}")
print(f"Grounded: {result['grounded']}, Useful: {result['useful']}")Die Kosten der Zuverlässigkeit
Jede Sicherheitsmaßnahme fügt einen weiteren LLM-Aufruf hinzu, und die Summe steigt schnell an. Die folgende Funktion macht dies explizit, indem sie die Aufrufe pro Schritt auflistet: eine Bewertungsanfrage pro Dokument, dann die Generierung, die Halluzinationsprüfung und die Nützlichkeitsprüfung, die jeweils bei jedem Wiederholungsversuch wiederholt werden. Bei vier Dokumenten und einem einzigen Wiederholungsversuch erreichen Sie zehn Aufrufe für eine einzelne Frage – wo grundlegendes RAG nur zwei benötigte. Führen Sie diese Schätzung mit Ihren eigenen Werten durch, bevor Sie k und max_retries festlegen: Diese Berechnung zeigt Ihnen, ob Ihr Budget und Ihre Latenzanforderungen Self-RAG über den gesamten Datenverkehr hinweg unterstützen oder nur für die sensibelsten Fragen.
def estimate_self_rag_cost(n_docs=4, n_retries=1):
"""Estimate the number of LLM calls for a Self-RAG question."""
calls = {
"grading": n_docs, # 1 call per document
"generation": 1 + n_retries, # 1 initial + retries
"hallucination_check": 1 + n_retries,
"answer_check": 1 + n_retries,
}
total = sum(calls.values())
print(f"Estimated LLM calls: {total}")
for step, count in calls.items():
print(f" {step}: {count}")
return total
estimate_self_rag_cost(n_docs=4, n_retries=1)
# Estimated LLM calls: 10Kernaussagen
- Self-RAG fügt zwei Prüfungen nach der Generierung hinzu: Halluzinationen und Nützlichkeit
- Wenn die Antwort nicht in den Dokumenten fundiert ist, generiert das System erneut
- Wenn die Antwort nicht nützlich ist, führt das System eine Websuche durch und versucht erneut
- Die Anzahl der Wiederholungsversuche muss begrenzt werden, um die Latenz zu kontrollieren
- Self-RAG ist teurer (mehr LLM-Aufrufe), aber zuverlässiger