Filtrage par métadonnées : Affiner la recherche RAG

Le filtrage par métadonnées combine la puissance de la recherche vectorielle avec la précision des filtres structurés. Au lieu de chercher seulement par similarité sémantique, vous pouvez contraindre les résultats par catégorie, date, auteur, prix, ou toute autre propriété. Ce guide explore les stratégies de filtrage et leur implémentation dans les systèmes RAG.

Pourquoi le filtrage par métadonnées ?

La recherche vectorielle pure a des limites :

Requête : "Derniers articles sur le machine learning"

Sans filtrage :
→ Trouve d'anciens articles très pertinents mais datés (2018, 2020)
→ Manque les articles récents moins optimisés sémantiquement

Avec filtrage (year >= 2024) :
→ Trouve uniquement les articles de 2024
→ Pertinence sémantique + fraîcheur garantie

Cas d'usage typiques

Types de métadonnées et filtres

1. Filtres scalaires (égalité, comparaison)

DEVELOPERpython
from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range

client = QdrantClient("localhost", port=6333)

# Égalité stricte
category_filter = Filter(
    must=[
        FieldCondition(
            key="category",
            match=MatchValue(value="electronics")
        )
    ]
)

# Comparaison numérique
price_filter = Filter(
    must=[
        FieldCondition(
            key="price",
            range=Range(gte=100, lte=500)  # 100 <= price <= 500
        )
    ]
)

# Booléen
in_stock_filter = Filter(
    must=[
        FieldCondition(
            key="in_stock",
            match=MatchValue(value=True)
        )
    ]
)

2. Filtres textuels (correspondance partielle)

DEVELOPERpython
from qdrant_client.models import MatchText

# Correspondance exacte dans un texte
title_filter = Filter(
    must=[
        FieldCondition(
            key="title",
            match=MatchText(text="guide")  # Contient "guide"
        )
    ]
)

# Préfixe
prefix_filter = Filter(
    must=[
        FieldCondition(
            key="product_code",
            match=MatchText(text="SKU-2024")  # Commence par "SKU-2024"
        )
    ]
)

3. Filtres sur tableaux

DEVELOPERpython
from qdrant_client.models import MatchAny

# Document avec au moins un des tags
tags_filter = Filter(
    must=[
        FieldCondition(
            key="tags",
            match=MatchAny(any=["rag", "llm", "embeddings"])
        )
    ]
)

# Document avec TOUS les tags (must pour chaque)
all_tags_filter = Filter(
    must=[
        FieldCondition(key="tags", match=MatchValue(value="rag")),
        FieldCondition(key="tags", match=MatchValue(value="production"))
    ]
)

4. Filtres temporels

DEVELOPERpython
from datetime import datetime, timedelta

# Documents des 7 derniers jours
now = datetime.now()
week_ago = now - timedelta(days=7)

recent_filter = Filter(
    must=[
        FieldCondition(
            key="created_at",
            range=Range(
                gte=week_ago.isoformat(),
                lte=now.isoformat()
            )
        )
    ]
)

# Documents d'une année spécifique
year_filter = Filter(
    must=[
        FieldCondition(
            key="published_date",
            range=Range(
                gte="2024-01-01T00:00:00Z",
                lt="2025-01-01T00:00:00Z"
            )
        )
    ]
)

5. Filtres géographiques

DEVELOPERpython
from qdrant_client.models import GeoRadius, GeoPoint

# Documents dans un rayon de 10km autour de Paris
geo_filter = Filter(
    must=[
        FieldCondition(
            key="location",
            geo_radius=GeoRadius(
                center=GeoPoint(lat=48.8566, lon=2.3522),
                radius=10000  # mètres
            )
        )
    ]
)

Opérateurs logiques

Combinaison AND (must)

DEVELOPERpython
# Tous les critères doivent être satisfaits
combined_filter = Filter(
    must=[
        FieldCondition(key="category", match=MatchValue(value="electronics")),
        FieldCondition(key="price", range=Range(lte=500)),
        FieldCondition(key="in_stock", match=MatchValue(value=True)),
        FieldCondition(key="rating", range=Range(gte=4.0))
    ]
)

Combinaison OR (should)

DEVELOPERpython
# Au moins un critère doit être satisfait
or_filter = Filter(
    should=[
        FieldCondition(key="brand", match=MatchValue(value="Apple")),
        FieldCondition(key="brand", match=MatchValue(value="Samsung")),
        FieldCondition(key="brand", match=MatchValue(value="Google"))
    ]
)

Exclusion (must_not)

DEVELOPERpython
# Exclure certains résultats
exclusion_filter = Filter(
    must=[
        FieldCondition(key="category", match=MatchValue(value="phones"))
    ],
    must_not=[
        FieldCondition(key="brand", match=MatchValue(value="Nokia")),
        FieldCondition(key="status", match=MatchValue(value="discontinued"))
    ]
)

Combinaisons complexes

DEVELOPERpython
# (category = phones AND price < 1000) AND (brand = Apple OR brand = Samsung) AND NOT refurbished
complex_filter = Filter(
    must=[
        FieldCondition(key="category", match=MatchValue(value="phones")),
        FieldCondition(key="price", range=Range(lt=1000))
    ],
    should=[
        FieldCondition(key="brand", match=MatchValue(value="Apple")),
        FieldCondition(key="brand", match=MatchValue(value="Samsung"))
    ],
    must_not=[
        FieldCondition(key="condition", match=MatchValue(value="refurbished"))
    ]
)

Implémentation dans un retriever

DEVELOPERpython
from sentence_transformers import SentenceTransformer

class MetadataFilteredRetriever:
    def __init__(self, collection: str):
        self.client = QdrantClient("localhost", port=6333)
        self.collection = collection
        self.embedder = SentenceTransformer("BAAI/bge-m3")

    def search(
        self,
        query: str,
        filters: dict = None,
        top_k: int = 5
    ) -> list[dict]:
        # Encoder la requête
        query_embedding = self.embedder.encode(query)

        # Construire le filtre
        qdrant_filter = self._build_filter(filters) if filters else None

        # Recherche vectorielle avec filtres
        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding.tolist(),
            query_filter=qdrant_filter,
            limit=top_k
        )

        return [
            {
                "id": hit.id,
                "content": hit.payload.get("content"),
                "metadata": {k: v for k, v in hit.payload.items() if k != "content"},
                "score": hit.score
            }
            for hit in results
        ]

    def _build_filter(self, filters: dict) -> Filter:
        """
        Convertit un dictionnaire simple en filtre Qdrant

        Syntaxe supportée :
        - {"category": "electronics"} → égalité
        - {"price__lt": 500} → moins que
        - {"price__gte": 100} → plus ou égal
        - {"tags__contains": "rag"} → contient
        - {"brand__in": ["Apple", "Samsung"]} → dans la liste
        - {"status__not": "draft"} → différent de
        """
        must_conditions = []
        must_not_conditions = []

        for key, value in filters.items():
            # Parser les opérateurs
            if "__" in key:
                field, operator = key.rsplit("__", 1)
            else:
                field, operator = key, "eq"

            condition = self._create_condition(field, operator, value)

            if operator == "not":
                must_not_conditions.append(condition)
            else:
                must_conditions.append(condition)

        return Filter(
            must=must_conditions if must_conditions else None,
            must_not=must_not_conditions if must_not_conditions else None
        )

    def _create_condition(self, field: str, operator: str, value) -> FieldCondition:
        if operator == "eq":
            return FieldCondition(key=field, match=MatchValue(value=value))

        elif operator == "lt":
            return FieldCondition(key=field, range=Range(lt=value))

        elif operator == "lte":
            return FieldCondition(key=field, range=Range(lte=value))

        elif operator == "gt":
            return FieldCondition(key=field, range=Range(gt=value))

        elif operator == "gte":
            return FieldCondition(key=field, range=Range(gte=value))

        elif operator == "in":
            return FieldCondition(key=field, match=MatchAny(any=value))

        elif operator == "contains":
            return FieldCondition(key=field, match=MatchValue(value=value))

        elif operator == "not":
            return FieldCondition(key=field, match=MatchValue(value=value))

        else:
            raise ValueError(f"Opérateur inconnu: {operator}")


# Utilisation
retriever = MetadataFilteredRetriever("products")

results = retriever.search(
    query="smartphone haut de gamme",
    filters={
        "category": "phones",
        "price__lte": 1000,
        "rating__gte": 4.5,
        "brand__in": ["Apple", "Samsung", "Google"],
        "status__not": "discontinued"
    },
    top_k=5
)

Indexation des métadonnées

Création d'une collection avec indices

DEVELOPERpython
from qdrant_client.models import (
    VectorParams,
    PayloadSchemaType,
    PayloadIndexParams,
    KeywordIndexParams,
    IntegerIndexParams,
    FloatIndexParams,
    TextIndexParams
)

# Créer la collection avec configuration des indices
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(size=1024, distance="Cosine")
)

# Ajouter des indices sur les champs fréquemment filtrés
client.create_payload_index(
    collection_name="products",
    field_name="category",
    field_schema=KeywordIndexParams(type="keyword")
)

client.create_payload_index(
    collection_name="products",
    field_name="price",
    field_schema=FloatIndexParams(type="float")
)

client.create_payload_index(
    collection_name="products",
    field_name="brand",
    field_schema=KeywordIndexParams(type="keyword")
)

client.create_payload_index(
    collection_name="products",
    field_name="created_at",
    field_schema=PayloadSchemaType.DATETIME
)

# Index full-text pour recherche dans le titre
client.create_payload_index(
    collection_name="products",
    field_name="title",
    field_schema=TextIndexParams(
        type="text",
        tokenizer="word",
        min_token_len=2,
        max_token_len=20
    )
)

Bonnes pratiques d'indexation

Optimisation des performances

Préfiltrage vs Postfiltrage

DEVELOPERpython
class OptimizedFilteredRetriever:
    def __init__(self, collection: str):
        self.client = QdrantClient("localhost", port=6333)
        self.collection = collection

    def search(
        self,
        query: str,
        filters: dict,
        top_k: int = 5,
        prefetch_multiplier: int = 3
    ) -> list[dict]:
        """
        Stratégie optimisée :
        1. Préfiltrage si les filtres sont très sélectifs
        2. Postfiltrage si les filtres sont permissifs
        """
        # Estimer la sélectivité des filtres
        selectivity = self._estimate_selectivity(filters)

        if selectivity < 0.1:  # < 10% des documents
            # Préfiltrage : filtrer puis chercher
            return self._prefetch_search(query, filters, top_k)
        else:
            # Postfiltrage : chercher plus puis filtrer
            return self._postfilter_search(query, filters, top_k, prefetch_multiplier)

    def _prefetch_search(self, query: str, filters: dict, top_k: int):
        """Applique les filtres avant la recherche vectorielle"""
        query_embedding = self.embedder.encode(query)
        qdrant_filter = self._build_filter(filters)

        return self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding.tolist(),
            query_filter=qdrant_filter,
            limit=top_k
        )

    def _postfilter_search(self, query: str, filters: dict, top_k: int, multiplier: int):
        """Récupère plus de résultats puis filtre localement"""
        query_embedding = self.embedder.encode(query)

        # Recherche large
        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding.tolist(),
            limit=top_k * multiplier
        )

        # Filtrage local
        filtered = [r for r in results if self._matches_filters(r.payload, filters)]
        return filtered[:top_k]

    def _estimate_selectivity(self, filters: dict) -> float:
        """Estime le pourcentage de documents qui passent les filtres"""
        # Requête de comptage
        total = self.client.count(collection_name=self.collection).count

        qdrant_filter = self._build_filter(filters)
        matching = self.client.count(
            collection_name=self.collection,
            count_filter=qdrant_filter
        ).count

        return matching / total if total > 0 else 0

Cache des filtres fréquents

DEVELOPERpython
from functools import lru_cache
import hashlib
import json

class CachedFilterRetriever:
    def __init__(self, collection: str, cache_size: int = 100):
        self.base_retriever = MetadataFilteredRetriever(collection)
        self._filter_cache = {}

    def search(self, query: str, filters: dict, top_k: int = 5) -> list[dict]:
        # Créer une clé de cache basée sur les filtres
        filter_key = self._hash_filters(filters)

        # Vérifier si on a des IDs pré-filtrés en cache
        if filter_key in self._filter_cache:
            cached_ids = self._filter_cache[filter_key]
            # Recherche vectorielle seulement parmi les IDs cachés
            return self._search_in_ids(query, cached_ids, top_k)

        # Recherche normale
        results = self.base_retriever.search(query, filters, top_k * 3)

        # Mettre en cache les IDs pour ce filtre
        self._filter_cache[filter_key] = [r["id"] for r in results]

        return results[:top_k]

    def _hash_filters(self, filters: dict) -> str:
        return hashlib.md5(json.dumps(filters, sort_keys=True).encode()).hexdigest()

Filtres dynamiques

Construction de filtres depuis l'interface utilisateur

DEVELOPERpython
class DynamicFilterBuilder:
    def __init__(self, schema: dict):
        """
        schema = {
            "category": {"type": "keyword", "options": ["phones", "laptops", ...]},
            "price": {"type": "range", "min": 0, "max": 5000},
            "brand": {"type": "multi_select", "options": [...]},
            "in_stock": {"type": "boolean"}
        }
        """
        self.schema = schema

    def build_from_ui(self, ui_params: dict) -> dict:
        """Convertit les paramètres UI en filtres"""
        filters = {}

        for field, value in ui_params.items():
            if field not in self.schema:
                continue

            field_type = self.schema[field]["type"]

            if field_type == "keyword" and value:
                filters[field] = value

            elif field_type == "range":
                if value.get("min") is not None:
                    filters[f"{field}__gte"] = value["min"]
                if value.get("max") is not None:
                    filters[f"{field}__lte"] = value["max"]

            elif field_type == "multi_select" and value:
                filters[f"{field}__in"] = value

            elif field_type == "boolean":
                if value is not None:
                    filters[field] = value

        return filters


# Utilisation depuis une API REST
@app.get("/search")
def search(
    q: str,
    category: str = None,
    price_min: float = None,
    price_max: float = None,
    brands: list[str] = Query(default=[]),
    in_stock: bool = None
):
    filter_builder = DynamicFilterBuilder(product_schema)

    ui_params = {
        "category": category,
        "price": {"min": price_min, "max": price_max},
        "brand": brands,
        "in_stock": in_stock
    }

    filters = filter_builder.build_from_ui(ui_params)
    return retriever.search(q, filters=filters)

Monitoring des filtres

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

    def log_filter_usage(
        self,
        filters: dict,
        results_count: int,
        latency_ms: float
    ):
        self.analytics.track("filter_usage", {
            "filters": filters,
            "filter_count": len(filters),
            "results_count": results_count,
            "latency_ms": latency_ms,
            "timestamp": datetime.now().isoformat()
        })

    def get_popular_filters(self, days: int = 7) -> dict:
        """Identifie les filtres les plus utilisés"""
        usages = self.analytics.query("filter_usage", days=days)

        filter_counts = {}
        for usage in usages:
            for field in usage["filters"].keys():
                filter_counts[field] = filter_counts.get(field, 0) + 1

        return sorted(filter_counts.items(), key=lambda x: x[1], reverse=True)

    def get_empty_result_filters(self, days: int = 7) -> list[dict]:
        """Identifie les filtres qui ne retournent aucun résultat"""
        usages = self.analytics.query("filter_usage", days=days)
        return [u for u in usages if u["results_count"] == 0]

Prochaines étapes

Le filtrage par métadonnées affine considérablement vos recherches RAG. Pour aller plus loin :

• Self-Query Retrieval - Laisser le LLM extraire les filtres
• Query Routing - Router selon les métadonnées
• Fondamentaux du Retrieval - Vue d'ensemble

Filtrage intelligent avec Ailog

Ailog implémente le filtrage par métadonnées de manière transparente :

• Indexation automatique des champs pertinents
• Extraction de filtres depuis les requêtes naturelles
• Optimisation dynamique préfiltrage/postfiltrage
• Interface de filtres intégrée pour vos utilisateurs

Testez gratuitement et affinez vos recherches avec des filtres puissants.