Query Routing : Anfragen zur richtigen Quelle leiten

Le query routing est une technique qui dirige chaque requête vers la source de données la plus pertinente. Plutôt que de chercher dans toutes les bases simultanément, un routeur intelligent analyse la requête et décide où chercher. Ce guide explore les stratégies de routing, des heuristiques simples aux classifieurs LLM sophistiqués.

Pourquoi le query routing ?

Dans un système RAG d'entreprise, les données proviennent de multiples sources :

┌─────────────────────────────────────────────────────────────┐
│                    Sources de données                        │
├──────────────┬──────────────┬──────────────┬────────────────┤
│     FAQ      │     Docs     │   Produits   │   Tickets      │
│  Support     │ Techniques   │  Catalogue   │   Résolus      │
├──────────────┴──────────────┴──────────────┴────────────────┤
│                                                              │
│  "Comment retourner ?"  →  FAQ Support                      │
│  "API rate limits ?"    →  Docs Techniques                  │
│  "Prix iPhone 15 ?"     →  Catalogue Produits               │
│  "Bug connexion WiFi ?" →  Tickets Résolus                  │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Bénéfices du routing

Stratégies de routing

1. Routing par mots-clés

La méthode la plus simple : détecter des patterns dans la requête.

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("Comment retourner un produit ?"))  # "faq"
print(router.route("API rate limit exceeded"))         # "docs"

Avantages : Rapide, prédictible, pas de coûts LLM
Inconvénients : Rigide, maintenance des patterns

2. Routing par embedding

Classifier la requête par similarité avec des exemples représentatifs.

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": [
                "Comment faire un retour ?",
                "Quels sont les délais de livraison ?",
                "Puis-je annuler ma commande ?",
                "Comment contacter le support ?"
            ],
            "docs": [
                "Comment intégrer l'API ?",
                "Quelle est la limite de requêtes ?",
                "Comment configurer OAuth ?",
                "Documentation des webhooks"
            ],
            "products": [
                "Quel est le prix de ce produit ?",
                "Est-ce disponible en stock ?",
                "Comparer les caractéristiques",
                "Meilleures ventes du moment"
            ]
        }
        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("Quel est le tarif mensuel ?")
print(f"Route: {route}, Confiance: {confidence:.3f}")
# Route: products, Vertrauen: 0.847

3. Routing par LLM

Le LLM analyse la requête et décide de la route optimale.

DEVELOPERpython
from openai import OpenAI

class LLMRouter:
    def __init__(self):
        self.client = OpenAI()
        self.routes = {
            "faq": "Questions générales sur le service, retours, livraisons, compte utilisateur",
            "docs": "Documentation technique, API, intégration, configuration, webhooks",
            "products": "Catalogue produits, prix, disponibilité, caractéristiques, comparaisons",
            "tickets": "Problèmes techniques, bugs, incidents, erreurs spécifiques"
        }

    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"""Tu es un routeur de requêtes. Analyse la question et détermine la meilleure source.

Sources disponibles:
{routes_description}

Réponds UNIQUEMENT au format JSON:
{{"route": "nom_route", "confidence": 0.0-1.0, "reasoning": "explication courte"}}"""
                },
                {"role": "user", "content": query}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )

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


router = LLMRouter()
result = router.route("L'API renvoie une erreur 429, comment augmenter ma limite ?")
# {"route": "docs", "confidence": 0.95, "reasoning": "Technische Frage zur API und zu den Limits"}

4. Routing hiérarchique

Combiner plusieurs niveaux de routing pour des décisions plus fines.

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
        }

Routing multi-route

Parfois, une requête nécessite plusieurs sources.

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": """Analyse la requête et détermine TOUTES les sources pertinentes.
Une requête peut nécessiter plusieurs sources.

Sources: faq, docs, products, tickets

Réponds au format JSON:
{"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)

        # Filtrer les routes avec relevance > 0.5
        relevant_routes = [
            r for r in result["routes"]
            if r["relevance"] > 0.5
        ]

        return relevant_routes


# Exemple
router = MultiRouteRouter()
routes = router.route("Pourquoi l'API produits renvoie une erreur 500 sur certains articles ?")
# [
#   {"name": "docs", "relevance": 0.9},
#   {"name": "tickets", "relevance": 0.8},
#   {"name": "products", "relevance": 0.6}
# ]

Routing avec métadonnées

Enrichir le routing avec le contexte utilisateur.

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


# Exemple
router = ContextualRouter()
result = router.route(
    query="Comment résoudre cette erreur ?",
    user_context={
        "is_developer": True,
        "has_open_tickets": True,
        "current_page": "/docs/api"
    }
)

Implémentation du pipeline complet

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. Retrieval ciblé
        retriever = self.retrievers[selected_route]
        documents = retriever.search(user_query, top_k=5)

        # 3. Génération
        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"Réponds à la question en te basant sur le contexte suivant:\n\n{context}"
                },
                {"role": "user", "content": query}
            ]
        )
        return response.choices[0].message.content

Monitoring et amélioration

Logging des décisions de routing

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 pour amélioration

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):
        """Enregistrer une correction utilisateur"""
        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()

Prochaines étapes

Le query routing optimise votre système en ciblant les bonnes sources. Pour aller plus loin :

• Self-Query Retrieval - Laisser le LLM structurer la recherche
• Metadata Filtering - Affiner avec les métadonnées
• Ensemble Retrieval - Combiner plusieurs retrievers

Query routing intelligent avec Ailog

Ailog implémente le query routing de manière transparente :

• Routing automatique basé sur vos sources de données
• Adaptation en temps réel selon les retours utilisateurs
• Multi-route intelligent quand plusieurs sources sont pertinentes
• Monitoring intégré pour optimiser continuellement

Testez gratuitement et bénéficiez d'un routing optimisé automatiquement.