Metadaten-Filter: Die RAG-Suche verfeinern

Das Filtern nach Metadaten kombiniert die Leistungsfähigkeit der vector-Suche mit der Präzision strukturierter Filter. Anstatt nur nach semantischer Ähnlichkeit zu suchen, können Sie die Ergebnisse nach Kategorie, Datum, Autor, Preis oder jeder anderen Eigenschaft einschränken. Dieser Leitfaden erläutert Filterstrategien und deren Implementierung in RAG-Systemen.

Warum Metadaten-Filter?

Reine vector-Suche hat Grenzen :

Anfrage: "Neueste Artikel über maschinelles Lernen"

Ohne Filterung:
→ Findet alte, sehr relevante, aber veraltete Artikel (2018, 2020)
→ Verpasst aktuelle Artikel, die semantisch weniger optimiert sind

Mit Filterung (year >= 2024):
→ Findet nur Artikel aus dem Jahr 2024
→ Semantische Relevanz + garantierte Aktualität

Typische Anwendungsfälle

Arten von Metadaten und Filtern

1. Skalare Filter (Gleichheit, Vergleich)

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

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

# Strikte Gleichheit
category_filter = Filter(
    must=[
        FieldCondition(
            key="category",
            match=MatchValue(value="electronics")
        )
    ]
)

# Numerischer Vergleich
price_filter = Filter(
    must=[
        FieldCondition(
            key="price",
            range=Range(gte=100, lte=500)  # 100 <= price <= 500
        )
    ]
)

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

2. Textfilter (partielle Übereinstimmung)

DEVELOPERpython
from qdrant_client.models import MatchText

# Exakte Übereinstimmung im Text
title_filter = Filter(
    must=[
        FieldCondition(
            key="title",
            match=MatchText(text="guide")  # Enthält "guide"
        )
    ]
)

# Präfix
prefix_filter = Filter(
    must=[
        FieldCondition(
            key="product_code",
            match=MatchText(text="SKU-2024")  # Beginnt mit "SKU-2024"
        )
    ]
)

3. Filter auf Arrays

DEVELOPERpython
from qdrant_client.models import MatchAny

# Dokument mit mindestens einem der Tags
tags_filter = Filter(
    must=[
        FieldCondition(
            key="tags",
            match=MatchAny(any=["rag", "llm", "embeddings"])
        )
    ]
)

# Dokument mit ALLEN Tags (must für jedes)
all_tags_filter = Filter(
    must=[
        FieldCondition(key="tags", match=MatchValue(value="rag")),
        FieldCondition(key="tags", match=MatchValue(value="production"))
    ]
)

4. Zeitliche Filter

DEVELOPERpython
from datetime import datetime, timedelta

# Dokumente der letzten 7 Tage
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()
            )
        )
    ]
)

# Dokumente eines bestimmten Jahres
year_filter = Filter(
    must=[
        FieldCondition(
            key="published_date",
            range=Range(
                gte="2024-01-01T00:00:00Z",
                lt="2025-01-01T00:00:00Z"
            )
        )
    ]
)

5. Geografische Filter

DEVELOPERpython
from qdrant_client.models import GeoRadius, GeoPoint

# Dokumente in einem Radius von 10 km um Paris
geo_filter = Filter(
    must=[
        FieldCondition(
            key="location",
            geo_radius=GeoRadius(
                center=GeoPoint(lat=48.8566, lon=2.3522),
                radius=10000  # Meter
            )
        )
    ]
)

Logische Operatoren

Kombination AND (must)

DEVELOPERpython
# Alle Kriterien müssen erfüllt sein
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))
    ]
)

Kombination OR (should)

DEVELOPERpython
# Mindestens ein Kriterium muss erfüllt sein
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"))
    ]
)

Ausschluss (must_not)

DEVELOPERpython
# Bestimmte Ergebnisse ausschließen
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"))
    ]
)

Komplexe Kombinationen

DEVELOPERpython
# (category = phones UND price < 1000) UND (brand = Apple ODER brand = Samsung) UND NICHT 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"))
    ]
)

Implementierung in einem 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]:
        # Anfrage enkodieren
        query_embedding = self.embedder.encode(query)

        # Filter erstellen
        qdrant_filter = self._build_filter(filters) if filters else None

        # Vector-Suche mit Filtern
        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:
        """
        Wandelt ein einfaches Dictionary in einen Qdrant-Filter um

        Unterstützte Syntax:
        - {"category": "electronics"} → Gleichheit
        - {"price__lt": 500} → kleiner als
        - {"price__gte": 100} → größer oder gleich
        - {"tags__contains": "rag"} → enthält
        - {"brand__in": ["Apple", "Samsung"]} → in der Liste
        - {"status__not": "draft"} → ungleich
        """
        must_conditions = []
        must_not_conditions = []

        for key, value in filters.items():
            # Operatoren parsen
            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"Unbekannter Operator: {operator}")


# Verwendung
retriever = MetadataFilteredRetriever("products")

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

Indexierung der Metadaten

Erstellen einer Collection mit Indizes

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

# Collection mit Index-Konfiguration erstellen
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(size=1024, distance="Cosine")
)

# Indizes für häufig gefilterte Felder hinzufügen
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
)

# Volltextindex für Suche im Titel
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
    )
)

Best Practices für die Indexierung

Performance-Optimierung

Vorfilterung vs Nachfilterung

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]:
        """
        Optimierte Strategie:
        1. Vorfilterung, wenn die Filter sehr selektiv sind
        2. Nachfilterung, wenn die Filter permissiv sind
        """
        # Selektivität der Filter schätzen
        selectivity = self._estimate_selectivity(filters)

        if selectivity < 0.1:  # < 10% der Dokumente
            # Prefiltering: zuerst filtern, dann suchen
            return self._prefetch_search(query, filters, top_k)
        else:
            # Postfiltering: mehr suchen und dann filtern
            return self._postfilter_search(query, filters, top_k, prefetch_multiplier)

    def _prefetch_search(self, query: str, filters: dict, top_k: int):
        """Wendet die Filter vor der vector-Suche an"""
        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):
        """Holt mehr Ergebnisse und filtert dann lokal"""
        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
        )

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

    def _estimate_selectivity(self, filters: dict) -> float:
        """Schätzt den Prozentsatz der Dokumente, die die Filter erfüllen"""
        # Zählabfrage
        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 für häufige Filter

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]:
        # Erstelle einen Cache-Schlüssel basierend auf den Filtern
        filter_key = self._hash_filters(filters)

        # Prüfen, ob vorgefilterte IDs im Cache vorhanden sind
        if filter_key in self._filter_cache:
            cached_ids = self._filter_cache[filter_key]
            # Nur vector-Suche unter den gecachten IDs
            return self._search_in_ids(query, cached_ids, top_k)

        # Normale Suche
        results = self.base_retriever.search(query, filters, top_k * 3)

        # IDs für diesen Filter cachen
        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()

Dynamische Filter

Erstellen von Filtern aus der Benutzeroberfläche

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:
        """Wandelt UI-Parameter in Filter um"""
        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


# Verwendung aus einer REST-API
@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 der Filter

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:
        """Identifiziert die am häufigsten verwendeten Filter"""
        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]:
        """Identifiziert Filter, die kein Ergebnis liefern"""
        usages = self.analytics.query("filter_usage", days=days)
        return [u for u in usages if u["results_count"] == 0]

Nächste Schritte

Das Filtern nach Metadaten verfeinert Ihre RAG-Suchen erheblich. Um weiterzugehen:

• Self-Query Retrieval - Das LLM Filter extrahieren lassen
• Query Routing - Routing basierend auf Metadaten
• Retrieval-Grundlagen - Übersicht

Intelligentes Filtern mit Ailog

Ailog implementiert Metadaten-Filter nahtlos:

• Automatische Indexierung relevanter Felder
• Extraktion von Filtern aus natürlichen Anfragen
• Dynamische Optimierung Prefiltering/Postfiltering
• Integrierte Filteroberfläche für Ihre Nutzer

Kostenlos testen und verfeinern Sie Ihre Suchen mit leistungsstarken Filtern.