Query Routing : Orienter les requêtes vers la bonne source

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)
            # Centroïde des exemples
            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, Confiance: 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": "Question technique sur l'API et les limites"}

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:
        # Niveau 1: Mots-clés (rapide, gratuit)
        keyword_route = self.keyword_router.route(query)

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

        # Niveau 2: Embedding (plus lent, gratuit)
        embed_route, embed_confidence = self.embedding_router.route(query)

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

        # Niveau 3: LLM (lent, payant) - seulement si nécessaire
        if use_llm_fallback:
            llm_result = self.llm_router.route(query)
            return {
                "route": llm_result["route"],
                "method": "llm",
                "confidence": llm_result["confidence"]
            }

        # Fallback: route par défaut
        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:
        # Routing de base
        base_route, confidence = self.base_router.route(query)

        # Ajustements contextuels
        adjustments = self._compute_adjustments(user_context)

        # Appliquer les ajustements
        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 = {}

        # Utilisateur technique → boost docs
        if user_context.get("is_developer"):
            adjustments["docs"] = 1.3

        # Historique de tickets → boost tickets
        if user_context.get("has_open_tickets"):
            adjustments["tickets"] = 1.2

        # Page courante = produit → boost products
        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)}}
        )

        # Calculer la précision par méthode
        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())

        # Vérifier si on a une correction
        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

        # Optionnel : réentraîner le modèle périodiquement
        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.