Query Routing : Anfragen zur richtigen Quelle leiten

Query Routing ist eine Technik, die jede Anfrage zur relevantesten Datenquelle leitet. Anstatt gleichzeitig in allen Datenbanken zu suchen, analysiert ein intelligenter Router die Anfrage und entscheidet, wo gesucht werden soll. Dieser Leitfaden untersucht Routing-Strategien, von einfachen Heuristiken bis hin zu ausgefeilten LLM-Klassifizierern.

Warum Query Routing?

In einem RAG-System für Unternehmen stammen die Daten aus mehreren Quellen:

┌─────────────────────────────────────────────────────────────┐
│                    Datenquellen                              │
├──────────────┬──────────────┬──────────────┬────────────────┤
│     FAQ      │     Docs     │   Produkte   │   Tickets      │
│  Support     │ Technisch    │  Katalog     │   Gelöst       │
├──────────────┴──────────────┴──────────────┴────────────────┤
│                                                              │
│  "Wie kann ich zurückgeben?" → FAQ Support                  │
│  "API rate limits?"          → Technische Docs               │
│  "Preis iPhone 15?"          → Produktkatalog                │
│  "WLAN-Verbindungsfehler?"   → Gelöste Tickets                │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Vorteile des Routings

Routing-Strategien

1. Routing nach Stichwörtern

Die einfachste Methode: Muster in der Anfrage erkennen.

DEVELOPERpython
import re

class KeywordRouter:
    def __init__(self):
        self.routes = {
            "faq": [
                r"comment\s+(faire|puis-je)",
                r"est-ce\s+possible",
                r"retour|remboursement|annuler",
                r"délai|livraison"
            ],
            "docs": [
                r"api|endpoint|webhook",
                r"intégr|configur|install",
                r"authentification|token|oauth",
                r"erreur\s+\d{3}"
            ],
            "products": [
                r"prix|tarif|coût",
                r"disponible|stock",
                r"caractéristiques|spécifications",
                r"comparer|versus|vs"
            ],
            "tickets": [
                r"bug|problème|erreur",
                r"ne\s+(fonctionne|marche)\s+pas",
                r"bloqué|coincé",
                r"résolu|solution"
            ]
        }

    def route(self, query: str) -> str:
        query_lower = query.lower()

        scores = {}
        for route, patterns in self.routes.items():
            scores[route] = sum(
                1 for p in patterns
                if re.search(p, query_lower)
            )

        if max(scores.values()) == 0:
            return "default"

        return max(scores, key=scores.get)


router = KeywordRouter()
print(router.route("Wie gebe ich ein Produkt zurück?"))  # "faq"
print(router.route("API rate limit exceeded"))           # "docs"

Vorteile: Schnell, vorhersagbar, keine LLM-Kosten
Nachteile: Starr, Pattern-Wartung nötig

2. Routing nach Embedding

Die Anfrage durch Ähnlichkeit mit repräsentativen Beispielen klassifizieren.

DEVELOPERpython
from sentence_transformers import SentenceTransformer
import numpy as np

class EmbeddingRouter:
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        self.route_embeddings = {}
        self.route_examples = {
            "faq": [
                "Wie mache ich eine Rückgabe?",
                "Wie lange dauert die Lieferung?",
                "Kann ich meine Bestellung stornieren?",
                "Wie kontaktiere ich den Support?"
            ],
            "docs": [
                "Wie integriere ich die API?",
                "Wie hoch ist das Anfragelimit?",
                "Wie konfiguriere ich OAuth?",
                "Dokumentation der Webhooks"
            ],
            "products": [
                "Was kostet dieses Produkt?",
                "Ist es auf Lager verfügbar?",
                "Merkmale vergleichen",
                "Aktuelle Bestseller"
            ]
        }
        self._build_route_embeddings()

    def _build_route_embeddings(self):
        for route, examples in self.route_examples.items():
            embeddings = self.model.encode(examples)
            # Zentroid der Beispiele
            self.route_embeddings[route] = np.mean(embeddings, axis=0)

    def route(self, query: str) -> tuple[str, float]:
        query_embedding = self.model.encode(query)

        best_route = None
        best_similarity = -1

        for route, centroid in self.route_embeddings.items():
            similarity = np.dot(query_embedding, centroid) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(centroid)
            )
            if similarity > best_similarity:
                best_similarity = similarity
                best_route = route

        return best_route, best_similarity


router = EmbeddingRouter()
route, confidence = router.route("Wie hoch ist der monatliche Preis?")
print(f"Route: {route}, Vertrauen: {confidence:.3f}")
# Route: products, Vertrauen: 0.847

3. Routing per LLM

Das LLM analysiert die Anfrage und entscheidet über die optimale Route.

DEVELOPERpython
from openai import OpenAI

class LLMRouter:
    def __init__(self):
        self.client = OpenAI()
        self.routes = {
            "faq": "Allgemeine Fragen zum Service, Rückgaben, Lieferungen, Nutzerkonto",
            "docs": "Technische Dokumentation, API, Integration, Konfiguration, Webhooks",
            "products": "Produktkatalog, Preise, Verfügbarkeit, Merkmale, Vergleiche",
            "tickets": "Technische Probleme, Bugs, Vorfälle, spezifische Fehler"
        }

    def route(self, query: str) -> dict:
        routes_description = "\n".join([
            f"- {name}: {desc}"
            for name, desc in self.routes.items()
        ])

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": f"""Du bist ein Query-Router. Analysiere die Frage und bestimme die beste Quelle.

Verfügbare Quellen:
{routes_description}

Antworte AUSSCHLIESSLICH im JSON-Format:
{{"route": "routen_name", "confidence": 0.0-1.0, "reasoning": "kurze Erklärung"}}"""
                },
                {"role": "user", "content": query}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )

        return json.loads(response.choices[0].message.content)


router = LLMRouter()
result = router.route("Die API gibt einen Fehler 429 zurück, wie erhöhe ich mein Limit?")
# {"route": "docs", "confidence": 0.95, "reasoning": "Technische Frage zur API und zu den Limits"}

4. Hierarchisches Routing

Mehrere Routing-Ebenen für präzisere Entscheidungen kombinieren.

DEVELOPERpython
class HierarchicalRouter:
    def __init__(self):
        self.keyword_router = KeywordRouter()
        self.embedding_router = EmbeddingRouter()
        self.llm_router = LLMRouter()

    def route(self, query: str, use_llm_fallback: bool = True) -> dict:
        # Ebene 1: Schlüsselwörter (schnell, kostenlos)
        keyword_route = self.keyword_router.route(query)

        if keyword_route != "default":
            return {
                "route": keyword_route,
                "method": "keyword",
                "confidence": 0.9
            }

        # Ebene 2: Embedding (langsamer, kostenlos)
        embed_route, embed_confidence = self.embedding_router.route(query)

        if embed_confidence > 0.85:
            return {
                "route": embed_route,
                "method": "embedding",
                "confidence": embed_confidence
            }

        # Ebene 3: LLM (langsam, kostenpflichtig) - nur wenn nötig
        if use_llm_fallback:
            llm_result = self.llm_router.route(query)
            return {
                "route": llm_result["route"],
                "method": "llm",
                "confidence": llm_result["confidence"]
            }

        # Fallback: Standard-Route
        return {
            "route": embed_route,
            "method": "embedding_fallback",
            "confidence": embed_confidence
        }

Multi-Route-Routing

Manchmal erfordert eine Anfrage mehrere Quellen.

DEVELOPERpython
class MultiRouteRouter:
    def __init__(self):
        self.client = OpenAI()

    def route(self, query: str) -> list[dict]:
        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": """Analysiere die Anfrage und bestimme ALLE relevanten Quellen.
Eine Anfrage kann mehrere Quellen erfordern.

Quellen: faq, docs, products, tickets

Antworte im JSON-Format:
{"routes": [{"name": "...", "relevance": 0.0-1.0}], "reasoning": "..."}"""
                },
                {"role": "user", "content": query}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )

        result = json.loads(response.choices[0].message.content)

        # Routen mit relevance > 0.5 filtern
        relevant_routes = [
            r for r in result["routes"]
            if r["relevance"] > 0.5
        ]

        return relevant_routes


# Beispiel
router = MultiRouteRouter()
routes = router.route("Warum gibt die Produkt-API bei manchen Artikeln einen Fehler 500 zurück?")
# [
#   {"name": "docs", "relevance": 0.9},
#   {"name": "tickets", "relevance": 0.8},
#   {"name": "products", "relevance": 0.6}
# ]

Routing mit Metadaten

Das Routing mit dem Nutzerkontext anreichern.

DEVELOPERpython
class ContextualRouter:
    def __init__(self):
        self.base_router = EmbeddingRouter()

    def route(
        self,
        query: str,
        user_context: dict
    ) -> dict:
        # Basis-Routing
        base_route, confidence = self.base_router.route(query)

        # Kontextuelle Anpassungen
        adjustments = self._compute_adjustments(user_context)

        # Anpassungen anwenden
        route_scores = {base_route: confidence}

        for route, adjustment in adjustments.items():
            if route in route_scores:
                route_scores[route] *= adjustment
            else:
                route_scores[route] = confidence * 0.5 * adjustment

        final_route = max(route_scores, key=route_scores.get)

        return {
            "route": final_route,
            "confidence": route_scores[final_route],
            "adjustments_applied": adjustments
        }

    def _compute_adjustments(self, user_context: dict) -> dict:
        adjustments = {}

        # Technischer Nutzer → docs verstärken
        if user_context.get("is_developer"):
            adjustments["docs"] = 1.3

        # Verlauf von Tickets → tickets verstärken
        if user_context.get("has_open_tickets"):
            adjustments["tickets"] = 1.2

        # Aktuelle Seite = Produkt → products verstärken
        if "product" in user_context.get("current_page", ""):
            adjustments["products"] = 1.4

        return adjustments


# Beispiel
router = ContextualRouter()
result = router.route(
    query="Wie behebe ich diesen Fehler?",
    user_context={
        "is_developer": True,
        "has_open_tickets": True,
        "current_page": "/docs/api"
    }
)

Implementierung der vollständigen Pipeline

DEVELOPERpython
class RoutedRAGPipeline:
    def __init__(self):
        self.router = HierarchicalRouter()
        self.retrievers = {
            "faq": FAQRetriever(),
            "docs": DocsRetriever(),
            "products": ProductRetriever(),
            "tickets": TicketRetriever()
        }
        self.llm = OpenAI()

    def query(self, user_query: str, user_context: dict = None) -> dict:
        # 1. Routing
        route_result = self.router.route(user_query)
        selected_route = route_result["route"]

        # 2. Gezieltes Retrieval
        retriever = self.retrievers[selected_route]
        documents = retriever.search(user_query, top_k=5)

        # 3. Generierung
        context = "\n\n".join([d["content"] for d in documents])
        response = self._generate_response(user_query, context)

        return {
            "answer": response,
            "route": selected_route,
            "routing_method": route_result["method"],
            "routing_confidence": route_result["confidence"],
            "sources": documents
        }

    def _generate_response(self, query: str, context: str) -> str:
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {
                    "role": "system",
                    "content": f"Beantworte die Frage auf Basis des folgenden Kontexts:\n\n{context}"
                },
                {"role": "user", "content": query}
            ]
        )
        return response.choices[0].message.content

Monitoring und Verbesserung

Protokollierung der Routing-Entscheidungen

DEVELOPERpython
class RoutingLogger:
    def __init__(self, analytics_client):
        self.analytics = analytics_client

    def log_routing_decision(
        self,
        query: str,
        route: str,
        method: str,
        confidence: float,
        user_feedback: str = None
    ):
        self.analytics.track("routing_decision", {
            "query": query,
            "route": route,
            "method": method,
            "confidence": confidence,
            "timestamp": datetime.now().isoformat(),
            "feedback": user_feedback
        })

    def analyze_routing_accuracy(self, days: int = 7) -> dict:
        decisions = self.analytics.query(
            "routing_decision",
            filters={"timestamp": {"$gte": days_ago(days)}}
        )

        # Berechne die Genauigkeit pro Methode
        by_method = {}
        for d in decisions:
            method = d["method"]
            if method not in by_method:
                by_method[method] = {"correct": 0, "incorrect": 0, "unknown": 0}

            if d.get("feedback") == "correct":
                by_method[method]["correct"] += 1
            elif d.get("feedback") == "incorrect":
                by_method[method]["incorrect"] += 1
            else:
                by_method[method]["unknown"] += 1

        return by_method

Feedback-Loop zur Verbesserung

DEVELOPERpython
class AdaptiveRouter:
    def __init__(self):
        self.base_router = EmbeddingRouter()
        self.corrections = {}  # query_hash -> correct_route

    def route(self, query: str) -> dict:
        query_hash = hash(query.lower().strip())

        # Prüfen, ob eine Korrektur vorhanden ist
        if query_hash in self.corrections:
            return {
                "route": self.corrections[query_hash],
                "method": "correction",
                "confidence": 1.0
            }

        return self.base_router.route(query)

    def record_correction(self, query: str, correct_route: str):
        """Eine Nutzerkorrektur speichern"""
        query_hash = hash(query.lower().strip())
        self.corrections[query_hash] = correct_route

        # Optional: Modell periodisch neu trainieren
        if len(self.corrections) % 100 == 0:
            self._retrain_router()

Nächste Schritte

Query Routing optimiert Ihr System, indem es die richtigen Quellen ansteuert. Um tiefer einzusteigen:

• Self-Query Retrieval - Das LLM die Suche strukturieren lassen
• Metadata Filtering - Mit Metadaten verfeinern
• Ensemble Retrieval - Mehrere Retriever kombinieren

Intelligentes Query Routing mit Ailog

Ailog implementiert Query Routing transparent:

• Automatisches Routing basierend auf Ihren Datenquellen
• Echtzeit-Anpassung anhand von Nutzerfeedback
• Intelligentes Multi-Route-Routing wenn mehrere Quellen relevant sind
• Integriertes Monitoring zur kontinuierlichen Optimierung

Kostenlos testen und von automatisch optimiertem Routing profitieren.