Source code for langchain.retrievers.self_query.qdrant

from __future__ import annotations

from typing import TYPE_CHECKING, Tuple

from langchain_core.structured_query import (
    Comparator,
    Comparison,
    Operation,
    Operator,
    StructuredQuery,
    Visitor,
)

if TYPE_CHECKING:
    from qdrant_client.http import models as rest


[docs]class QdrantTranslator(Visitor): """Translate `Qdrant` internal query language elements to valid filters.""" allowed_operators = ( Operator.AND, Operator.OR, Operator.NOT, ) """Subset of allowed logical operators.""" allowed_comparators = ( Comparator.EQ, Comparator.LT, Comparator.LTE, Comparator.GT, Comparator.GTE, Comparator.LIKE, ) """Subset of allowed logical comparators."""
[docs] def __init__(self, metadata_key: str): self.metadata_key = metadata_key
[docs] def visit_operation(self, operation: Operation) -> rest.Filter: try: from qdrant_client.http import models as rest except ImportError as e: raise ImportError( "Cannot import qdrant_client. Please install with `pip install " "qdrant-client`." ) from e args = [arg.accept(self) for arg in operation.arguments] operator = { Operator.AND: "must", Operator.OR: "should", Operator.NOT: "must_not", }[operation.operator] return rest.Filter(**{operator: args})
[docs] def visit_comparison(self, comparison: Comparison) -> rest.FieldCondition: try: from qdrant_client.http import models as rest except ImportError as e: raise ImportError( "Cannot import qdrant_client. Please install with `pip install " "qdrant-client`." ) from e self._validate_func(comparison.comparator) attribute = self.metadata_key + "." + comparison.attribute if comparison.comparator == Comparator.EQ: return rest.FieldCondition( key=attribute, match=rest.MatchValue(value=comparison.value) ) if comparison.comparator == Comparator.LIKE: return rest.FieldCondition( key=attribute, match=rest.MatchText(text=comparison.value) ) kwargs = {comparison.comparator.value: comparison.value} return rest.FieldCondition(key=attribute, range=rest.Range(**kwargs))
[docs] def visit_structured_query( self, structured_query: StructuredQuery ) -> Tuple[str, dict]: try: from qdrant_client.http import models as rest except ImportError as e: raise ImportError( "Cannot import qdrant_client. Please install with `pip install " "qdrant-client`." ) from e if structured_query.filter is None: kwargs = {} else: filter = structured_query.filter.accept(self) if isinstance(filter, rest.FieldCondition): filter = rest.Filter(must=[filter]) kwargs = {"filter": filter} return structured_query.query, kwargs