Self-Query Retrieval : Lassen Sie das LLM die Suche strukturieren

Self-Query Retrieval nutzt ein LLM, um eine Anfrage in natürlicher Sprache in eine Kombination aus semantischer Suche und strukturierten Filtern umzuwandeln. Anstatt nach "Samsung-Smartphones unter 500 Euro" zu suchen, versteht das System automatisch: Suche nach "Samsung-Smartphones" + Filter Preis < 500. Dieser Leitfaden untersucht diese leistungsstarke Technik und ihre Implementierung.

Das Problem komplexer Anfragen

Benutzeranfragen vermischen häufig semantische Absicht und faktische Einschränkungen:

"Articles sur le machine learning publiés en 2024 par des auteurs français"

Décomposition :
├── Recherche sémantique : "machine learning"
├── Filtre date : year = 2024
└── Filtre auteur : country = "France"

Eine reine Vektorsuche kann diese Einschränkungen nicht effektiv verarbeiten. Self-Query Retrieval löst dieses Problem.

Wie Self-Query funktioniert

┌─────────────────────────────────────────────────────────────┐
│                   Pipeline Self-Query                        │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  Requête utilisateur                                         │
│  "Produits Apple < 1000€ sortis cette année"                │
│                        │                                     │
│                        ▼                                     │
│                 ┌─────────────┐                             │
│                 │     LLM     │                             │
│                 │  Extracteur │                             │
│                 └─────────────┘                             │
│                        │                                     │
│          ┌─────────────┴─────────────┐                      │
│          ▼                           ▼                      │
│   ┌─────────────┐           ┌─────────────┐                │
│   │  Semantic   │           │  Structured │                │
│   │   Query     │           │   Filters   │                │
│   │ "Produits   │           │ brand=Apple │                │
│   │  Apple"     │           │ price<1000  │                │
│   └─────────────┘           │ year=2024   │                │
│          │                  └─────────────┘                │
│          │                         │                        │
│          └───────────┬─────────────┘                        │
│                      ▼                                      │
│              ┌─────────────┐                                │
│              │  Combined   │                                │
│              │   Search    │                                │
│              └─────────────┘                                │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Grundlegende Implementierung

1. Metadaten-Schema definieren

DEVELOPERpython
from pydantic import BaseModel, Field
from typing import Optional, Literal
from enum import Enum

class ProductCategory(str, Enum):
    ELECTRONICS = "electronics"
    CLOTHING = "clothing"
    HOME = "home"
    SPORTS = "sports"

class ProductMetadata(BaseModel):
    """Schéma des métadonnées disponibles pour le filtrage"""

    brand: Optional[str] = Field(
        None,
        description="Marque du produit (ex: Apple, Samsung, Nike)"
    )
    category: Optional[ProductCategory] = Field(
        None,
        description="Catégorie du produit"
    )
    price_min: Optional[float] = Field(
        None,
        description="Prix minimum en euros"
    )
    price_max: Optional[float] = Field(
        None,
        description="Prix maximum en euros"
    )
    year: Optional[int] = Field(
        None,
        description="Année de sortie du produit"
    )
    in_stock: Optional[bool] = Field(
        None,
        description="Disponibilité en stock"
    )
    rating_min: Optional[float] = Field(
        None,
        description="Note minimale (1-5)"
    )


class SelfQueryOutput(BaseModel):
    """Sortie du LLM self-query"""

    semantic_query: str = Field(
        description="La partie de la requête pour la recherche sémantique"
    )
    filters: ProductMetadata = Field(
        default_factory=ProductMetadata,
        description="Filtres structurés extraits de la requête"
    )

2. LLM-Extraktor erstellen

DEVELOPERpython
from openai import OpenAI
import json

class SelfQueryExtractor:
    def __init__(self):
        self.client = OpenAI()
        self.schema = ProductMetadata.model_json_schema()

    def extract(self, query: str) -> SelfQueryOutput:
        system_prompt = f"""Tu es un extracteur de requêtes. Analyse la question utilisateur et extrais :
1. La partie sémantique (ce qu'on recherche conceptuellement)
2. Les filtres structurés (contraintes factuelles)

Schéma des filtres disponibles :
{json.dumps(self.schema, indent=2)}

Règles :
- N'invente pas de filtres non présents dans la requête
- Pour les prix, utilise price_min et/ou price_max
- Pour "cette année", utilise year: 2024
- Pour "récent", utilise year: 2023 ou 2024

Réponds au format JSON :
{{
  "semantic_query": "description de ce qu'on cherche",
  "filters": {{
    "brand": "...",
    "category": "...",
    ...
  }}
}}"""

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": query}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )

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


# Test
extractor = SelfQueryExtractor()
result = extractor.extract("Smartphones Samsung à moins de 500€ avec bonne note")

print(f"Recherche sémantique: {result.semantic_query}")
# "Smartphones Samsung"

print(f"Filtres: {result.filters}")
# brand="Samsung", price_max=500, rating_min=4.0, category="electronics"

3. Mit dem Retriever integrieren

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

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

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        # 1. Extraire requête sémantique et filtres
        extracted = self.extractor.extract(query)

        # 2. Construire les filtres Qdrant
        qdrant_filter = self._build_filter(extracted.filters)

        # 3. Encoder la requête sémantique
        query_embedding = self.embedder.encode(extracted.semantic_query)

        # 4. Recherche combinée
        results = self.client.search(
            collection_name=self.collection,
            query_vector=query_embedding.tolist(),
            query_filter=qdrant_filter,
            limit=top_k
        )

        return [
            {
                "content": hit.payload["content"],
                "metadata": hit.payload,
                "score": hit.score,
                "extracted_filters": extracted.filters.model_dump(exclude_none=True)
            }
            for hit in results
        ]

    def _build_filter(self, filters: ProductMetadata) -> Filter:
        conditions = []

        if filters.brand:
            conditions.append(
                FieldCondition(key="brand", match=MatchValue(value=filters.brand))
            )

        if filters.category:
            conditions.append(
                FieldCondition(key="category", match=MatchValue(value=filters.category))
            )

        if filters.price_max:
            conditions.append(
                FieldCondition(key="price", range=Range(lte=filters.price_max))
            )

        if filters.price_min:
            conditions.append(
                FieldCondition(key="price", range=Range(gte=filters.price_min))
            )

        if filters.year:
            conditions.append(
                FieldCondition(key="year", match=MatchValue(value=filters.year))
            )

        if filters.in_stock is not None:
            conditions.append(
                FieldCondition(key="in_stock", match=MatchValue(value=filters.in_stock))
            )

        if filters.rating_min:
            conditions.append(
                FieldCondition(key="rating", range=Range(gte=filters.rating_min))
            )

        return Filter(must=conditions) if conditions else None

Umgang mit komplexen Anfragen

Logische Operatoren (OR, NOT)

DEVELOPERpython
class AdvancedSelfQueryOutput(BaseModel):
    semantic_query: str
    must_filters: list[dict] = Field(default_factory=list, description="Conditions AND")
    should_filters: list[dict] = Field(default_factory=list, description="Conditions OR")
    must_not_filters: list[dict] = Field(default_factory=list, description="Exclusions")


class AdvancedSelfQueryExtractor:
    def extract(self, query: str) -> AdvancedSelfQueryOutput:
        system_prompt = """Analyse la requête et extrais les filtres avec leur logique :
- must_filters : conditions obligatoires (ET)
- should_filters : conditions optionnelles (OU)
- must_not_filters : exclusions (SAUF)

Exemple pour "Laptops Apple ou Dell, pas gaming, moins de 1500€" :
{
  "semantic_query": "Laptops",
  "must_filters": [{"price_max": 1500}],
  "should_filters": [{"brand": "Apple"}, {"brand": "Dell"}],
  "must_not_filters": [{"category": "gaming"}]
}"""

        # ... appel LLM similaire


def build_advanced_filter(extracted: AdvancedSelfQueryOutput) -> Filter:
    """Construire un filtre Qdrant avec opérateurs logiques"""

    must_conditions = [
        _condition_to_qdrant(f) for f in extracted.must_filters
    ]

    should_conditions = [
        _condition_to_qdrant(f) for f in extracted.should_filters
    ]

    must_not_conditions = [
        _condition_to_qdrant(f) for f in extracted.must_not_filters
    ]

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

Relative zeitliche Filter

DEVELOPERpython
from datetime import datetime, timedelta

class TemporalFilters(BaseModel):
    """Gestion des expressions temporelles relatives"""

    after_date: Optional[datetime] = None
    before_date: Optional[datetime] = None
    relative_period: Optional[str] = Field(
        None,
        description="Période relative : 'last_week', 'last_month', 'last_year', 'this_year'"
    )


def resolve_temporal_filter(period: str) -> tuple[datetime, datetime]:
    """Convertir une période relative en dates absolues"""
    now = datetime.now()

    periods = {
        "today": (now.replace(hour=0, minute=0), now),
        "yesterday": (now - timedelta(days=1), now - timedelta(days=1)),
        "last_week": (now - timedelta(weeks=1), now),
        "last_month": (now - timedelta(days=30), now),
        "last_year": (now - timedelta(days=365), now),
        "this_year": (datetime(now.year, 1, 1), now),
        "this_month": (datetime(now.year, now.month, 1), now),
    }

    return periods.get(period, (None, None))

Fortgeschrittene Optimierungen

Cache der Extraktionen

DEVELOPERpython
import hashlib
from functools import lru_cache

class CachedSelfQueryExtractor:
    def __init__(self, cache_size: int = 1000):
        self.base_extractor = SelfQueryExtractor()
        self._cache = {}
        self.cache_size = cache_size

    def extract(self, query: str) -> SelfQueryOutput:
        # Normaliser la requête pour le cache
        normalized = self._normalize(query)
        cache_key = hashlib.md5(normalized.encode()).hexdigest()

        if cache_key in self._cache:
            return self._cache[cache_key]

        result = self.base_extractor.extract(query)

        # Gérer la taille du cache
        if len(self._cache) >= self.cache_size:
            # Supprimer les plus anciennes entrées
            oldest_key = next(iter(self._cache))
            del self._cache[oldest_key]

        self._cache[cache_key] = result
        return result

    def _normalize(self, query: str) -> str:
        return query.lower().strip()

Lokale Extraktion mit feinabgestimmtem Modell

Um Latenz und Kosten zu reduzieren, verwenden Sie ein lokales Modell:

DEVELOPERpython
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer

class LocalSelfQueryExtractor:
    def __init__(self, model_path: str = "your-org/self-query-extractor"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSeq2SeqLM.from_pretrained(model_path)

    def extract(self, query: str) -> SelfQueryOutput:
        inputs = self.tokenizer(
            f"extract filters: {query}",
            return_tensors="pt",
            max_length=256,
            truncation=True
        )

        outputs = self.model.generate(
            **inputs,
            max_length=128,
            num_beams=4
        )

        result_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return self._parse_output(result_text)

    def _parse_output(self, text: str) -> SelfQueryOutput:
        # Parser le format structuré généré par le modèle
        # Format attendu : "query: ... | brand: ... | price_max: ..."
        parts = dict(p.split(": ") for p in text.split(" | "))
        return SelfQueryOutput(
            semantic_query=parts.get("query", ""),
            filters=ProductMetadata(**{k: v for k, v in parts.items() if k != "query"})
        )

Validierung und Fallback

DEVELOPERpython
class RobustSelfQueryRetriever:
    def __init__(self, collection: str):
        self.extractor = SelfQueryExtractor()
        self.retriever = VectorRetriever(collection)

    def search(self, query: str, top_k: int = 5) -> list[dict]:
        try:
            # Tenter l'extraction self-query
            extracted = self.extractor.extract(query)

            # Valider les filtres extraits
            if not self._validate_filters(extracted.filters):
                raise ValueError("Filtres invalides")

            # Recherche avec filtres
            results = self.retriever.search(
                query=extracted.semantic_query,
                filters=extracted.filters,
                top_k=top_k
            )

            # Vérifier qu'on a des résultats
            if len(results) < 2:
                # Fallback : relâcher les filtres
                results = self._search_with_relaxed_filters(
                    extracted, top_k
                )

            return results

        except Exception as e:
            # Fallback : recherche vectorielle pure
            print(f"Self-query failed: {e}, falling back to vector search")
            return self.retriever.search(query=query, top_k=top_k)

    def _validate_filters(self, filters: ProductMetadata) -> bool:
        """Valider la cohérence des filtres"""
        if filters.price_min and filters.price_max:
            if filters.price_min > filters.price_max:
                return False

        if filters.rating_min and (filters.rating_min < 1 or filters.rating_min > 5):
            return False

        return True

    def _search_with_relaxed_filters(
        self,
        extracted: SelfQueryOutput,
        top_k: int
    ) -> list[dict]:
        """Relâcher progressivement les filtres si pas de résultats"""
        filters = extracted.filters.model_copy()

        # Ordre de relaxation : prix → date → note → marque
        relaxation_order = ["price_max", "price_min", "year", "rating_min", "brand"]

        for field in relaxation_order:
            setattr(filters, field, None)
            results = self.retriever.search(
                query=extracted.semantic_query,
                filters=filters,
                top_k=top_k
            )
            if len(results) >= 2:
                return results

        # Dernier recours : pas de filtres
        return self.retriever.search(query=extracted.semantic_query, top_k=top_k)

Monitoring und Verbesserung

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

    def log_extraction(
        self,
        original_query: str,
        extracted: SelfQueryOutput,
        results_count: int,
        latency_ms: float
    ):
        self.analytics.track("self_query_extraction", {
            "original_query": original_query,
            "semantic_query": extracted.semantic_query,
            "filters_count": len([f for f in extracted.filters.model_dump().values() if f]),
            "results_count": results_count,
            "latency_ms": latency_ms,
            "timestamp": datetime.now().isoformat()
        })

    def get_common_filters(self, days: int = 7) -> dict:
        """Identifier les filtres les plus utilisés"""
        extractions = self.analytics.query("self_query_extraction", days=days)

        filter_counts = {}
        for e in extractions:
            for field, value in e.get("filters", {}).items():
                if value:
                    filter_counts[field] = filter_counts.get(field, 0) + 1

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

Nächste Schritte

Self-Query Retrieval transformiert komplexe Anfragen in strukturierte Suchen. Um tiefer einzusteigen:

• Metadata Filtering - Fortgeschrittene Filter beherrschen
• Query Routing - Anfragen an die richtigen Quellen leiten
• Contextual Compression - Das Wesentliche extrahieren

Intelligentes Self-Query mit Ailog

Ailog implementiert Self-Query Retrieval automatisch:

• Intelligente Filterextraktion basierend auf Ihren Metadaten
• Automatischer Fallback wenn Filter zu restriktiv sind
• Optimierter Cache für häufige Anfragen
• Integriertes Monitoring zur Verbesserung der Extraktion

Kostenlos testen und komplexe Anfragen in präzise Suchen verwandeln.