Adaptiver RAG
Auf dieser Seite
Fragen intelligent routen
Self-RAG wendet auf jede Frage dieselbe Behandlung an, und genau darin liegt sein Mangel. Fragen Sie ihn, was zwei plus zwei ergibt: Er kodiert die Anfrage, durchsucht den Index, bewertet vier irrelevante Dokumente, schaltet auf das Web um, generiert und verifiziert. Das sind ein Dutzend LLM-Aufrufe für eine Frage, die ein Modell sofort lösen kann. Nicht alle Fragen haben die gleiche Komplexität, und eine einfache Faktenfrage erfordert nicht dieselbe Behandlung wie eine, die eine Synthese über mehrere Dokumente hinweg verlangt. Adaptiver RAG fügt daher einen intelligenten Router upstream des Pipelines hinzu: Er analysiert die Frage und wählt die optimale Strategie aus, bevor er in kostspielige Verarbeitungsschritte einsteigt.
Der Router hat drei Ziele. Fragen zu Themen, die tatsächlich von Ihren Dokumenten abgedeckt werden, gehen zum Vektorspeicher; dies ist der vollständige RAG-Pfad. Fragen zu aktuellen Ereignissen oder außerhalb des indizierten Bereichs gehen zur Web-Suche, da eine Abfrage des Index hier nur Rauschen zurückgeben würde. Schließlich erhalten einfache oder allgemeine Wissensfragen eine direkte Antwort vom Modell, ohne jegliche Abrufoperation.
Implementierung des Routers
Der knifflige Teil ist, dass der Router den Umfang Ihres Indexes kennen muss, um eine Entscheidung treffen zu können. Die gewählte Lösung ist so einfach wie effektiv: Übermitteln Sie ihm die Liste der abgedeckten Themen, indexed_topics, die er mit der Frage vergleicht. Die drei Regeln sind im Prompt explizit formuliert, und das formatierte JSON-Garantiert einen verwertbaren Wert. Der Ausdruck result.get("route", "vectorstore") verdient Aufmerksamkeit: Im Falle einer unerwarteten Ausgabe fällt die Standardregel auf den Vektorspeicher zurück. Dies ist das sicherste Verhalten, da es auf Ihre Daten statt auf das Web oder das Gedächtnis des Modells vertraut.
import json
from mistralai.client import Mistral
client = Mistral(api_key=api_key)
def route_question(question, indexed_topics):
"""Route a question to the best strategy."""
topics_str = ", ".join(indexed_topics)
prompt = f"""You are a question router. Given a question,
determine the best source of information.
Your indexed documents cover the following topics: {topics_str}
Rules:
- If the question concerns indexed topics -> "vectorstore"
- If the question concerns current events or an uncovered topic -> "websearch"
- If the question is very simple or general knowledge -> "direct"
Respond ONLY with a JSON object: {{"route": "vectorstore"}} or {{"route": "websearch"}} or {{"route": "direct"}}
Question: {question}
Route JSON:"""
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("route", "vectorstore")
# Test the router
topics = ["embeddings", "RAG", "Mistral AI", "vector search", "FAISS"]
tests = [
"How does mistral-embed work?", # -> vectorstore
"What is Apple's stock price today?", # -> websearch
"How much is 2 + 2?", # -> direct
"What are the steps of RAG?", # -> vectorstore
"Who won the 2026 World Cup?", # -> websearch
]
for q in tests:
route = route_question(q, topics)
print(f" [{route:12s}] {q}")Der Testsatz illustriert jedes Szenario und dient als Ihre Absicherung. Eine Frage zu mistral-embed fällt eindeutig in die deklarierten Themen, daher geht sie zum Vektorspeicher. Aktienkurse oder Wettbewerbsergebnisse ändern sich täglich: Ein statischer Index kann sie nicht enthalten, daher gehen sie ins Web. Zwei plus zwei erfordert weder einen Korpus noch eine Suche. Nutzen Sie diesen Block mit Ihren eigenen Themen und etwa fünfzehn Fragen, die Ihren tatsächlichen Datenverkehr repräsentieren: Dies ist die wirtschaftlichste Methode, um zu überprüfen, ob die topics-Liste genau genug ist. Ein Router, der alles an den Vektorspeicher sendet, bringt keinen Mehrwert, und ein Router, der auf Ihre Geschäftsthemen direkt antwortet, ist geradezu gefährlich.
Das vollständige Pipeline
AdaptiveRAG vereint alles, was Sie in den letzten drei Lektionen aufgebaut haben. Jede Route hat ihre eigene Ausführungsart, und die ask-Methode liest sich wie ein Switch-Statement. Die direct-Route gibt sofort zurück, mit einer Temperatur von 0,3, da es keinen Kontext gibt, dem man folgen muss, und einer leeren Quellenliste, die ehrlich signalisiert, dass diese Antwort nicht auf etwas Unterliegendem basiert. Die websearch-Route generiert allein aus Web-Ergebnissen. Die vectorstore-Route hingegen setzt die volle Artillerie ein: Dokumentbewertung im Stil von Corrective RAG, Web-Reservefall, falls keine überleben, und anschließend ein Generierungszyklus mit Verankerungs- und Nutzenprüfungen, die aus Self-RAG übernommen wurden und hier in grade_and_correct gruppiert sind.
Ein Schwellenwertunterschied ist im Vergleich zur Corrective RAG-Lektion erwähnenswert. Der Web-Reservefall löst nicht mehr aus, sobald ein Dokument verworfen wird, sondern nur dann, wenn relevant_docs leer ist. Diese Wahl ist toleranter: Wenn zwei von vier Segmenten relevant bleiben, ist es vernünftig, mit ihnen zu generieren, anstatt ins Web zu gehen, was einen Aufruf spart und die Latenz reduziert. Sie müssen basierend auf den Anforderungen Ihres Fachgebiets abstimmen.
import numpy as np
class AdaptiveRAG:
"""Adaptive RAG pipeline with intelligent routing."""
def __init__(self, chunks, embeddings, index, topics):
self.chunks = chunks
self.embeddings = embeddings
self.index = index
self.topics = topics
def route(self, question):
"""Determine the optimal strategy."""
return route_question(question, self.topics)
def retrieve_from_vectorstore(self, question, k=4):
"""Retrieval from the local vector store."""
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": "vectorstore"}
for dist, idx in zip(distances[0], indices[0]) if idx >= 0
]
def search_web(self, question):
"""Web search for out-of-scope questions."""
return web_search_fallback(question)
def direct_answer(self, question):
"""Direct answer without retrieval."""
response = client.chat.complete(
model="mistral-large-latest",
messages=[{"role": "user", "content": question}],
temperature=0.3,
)
return response.choices[0].message.content
def generate_with_context(self, question, documents):
"""Generation with context (classic RAG)."""
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 provided context."},
{"role": "user", "content": prompt},
],
temperature=0.1,
)
return response.choices[0].message.content
def grade_and_correct(self, question, documents, generation):
"""Self-RAG checks (hallucinations + usefulness)."""
is_grounded = check_hallucination(documents, generation)
if not is_grounded:
return None, "hallucination"
is_useful = check_answer_usefulness(question, generation)
if not is_useful:
return None, "not_useful"
return generation, "ok"
def ask(self, question, max_retries=2):
"""Complete Adaptive RAG pipeline."""
print(f"\n{'='*60}")
print(f"Question: {question}")
# 1. Routing
route = self.route(question)
print(f"Chosen route: {route}")
# 2. Execute based on route
if route == "direct":
answer = self.direct_answer(question)
return {"answer": answer, "route": "direct", "sources": []}
elif route == "websearch":
docs = self.search_web(question)
answer = self.generate_with_context(question, docs)
return {"answer": answer, "route": "websearch", "sources": ["web"]}
else: # vectorstore
docs = self.retrieve_from_vectorstore(question)
# Document grading
relevant_docs = []
for doc in docs:
if grade_document(question, doc["text"]):
relevant_docs.append(doc)
# Web fallback if not enough relevant documents
if not relevant_docs:
print(" No relevant documents, web fallback...")
relevant_docs = self.search_web(question)
# Generation loop with verification
for attempt in range(max_retries + 1):
generation = self.generate_with_context(question, relevant_docs)
answer, status = self.grade_and_correct(
question, relevant_docs, generation
)
if status == "ok":
return {
"answer": answer,
"route": "vectorstore",
"attempts": attempt + 1,
"sources": [d.get("source", "") for d in relevant_docs],
}
elif status == "not_useful":
web_docs = self.search_web(question)
relevant_docs.extend(web_docs)
# Return the last generation even if imperfect
return {
"answer": generation,
"route": "vectorstore",
"attempts": max_retries + 1,
"sources": [d.get("source", "") for d in relevant_docs],
}
# Usage
adaptive_rag = AdaptiveRAG(chunks, embeddings, index, topics=topics)
questions = [
"How do I create a FAISS index with mistral-embed?",
"What is the weather in Paris tomorrow?",
"How many dimensions does codestral-embed have?",
]
for q in questions:
result = adaptive_rag.ask(q)
print(f"\nRoute: {result['route']}")
print(f"Answer: {result['answer'][:200]}...")Die drei Beispielfragen durchlaufen in einer einzigen Ausführung alle Pfade, und das Feld route im Ergebnis zeigt Ihnen, welcher Pfad gewählt wurde. Dies ist der Wert, den Sie in der Produktion protokollieren sollten: Die Verteilung der Routen über Ihren tatsächlichen Datenverkehr ist der beste Gesundheitsindikator für Ihr System und offenbart sowohl einen Index, der zu eng ist, als auch einen schlecht kalibrierten Router.
Kernaussagen
- Adaptiver RAG leitet jede Frage zur optimalen Strategie (Vektorspeicher, Web, direkt) weiter
- Der Router verwendet ein LLM mit strukturierten Ausgaben zur Klassifizierung von Fragen
- Fragen zu indizierten Themen gehen zum Vektorspeicher; der Rest geht ins Web
- Routing reduziert Latenz und Kosten, indem es für einfache Fragen den vollständigen RAG-Prozess vermeidet
- Adaptiver RAG kombiniert Ideen aus Corrective RAG und Self-RAG mit Routing