from abc import abstractmethod
from enum import Enum
from typing import Any, Dict, Optional
from pydantic import Field
from weaviate.collections.classes.config_base import (
_ConfigCreateModel,
_ConfigUpdateModel,
_QuantizerConfigCreate,
_QuantizerConfigUpdate,
)
from weaviate.collections.classes.config_vectorizers import VectorDistances
[docs]
class VectorFilterStrategy(str, Enum):
"""Set the strategy when doing a filtered HNSW search.
Attributes:
SWEEPING: Do normal ANN search and skip nodes.
ACORN: Multi-hop search to find new candidates matching the filter.
"""
SWEEPING = "sweeping"
ACORN = "acorn"
[docs]
class VectorIndexType(str, Enum):
"""The available vector index types in Weaviate.
Attributes:
HNSW: Hierarchical Navigable Small World (HNSW) index.
FLAT: Flat index.
"""
HNSW = "hnsw"
FLAT = "flat"
DYNAMIC = "dynamic"
[docs]
class _MultiVectorConfigCreateBase(_ConfigCreateModel):
enabled: bool = Field(default=True)
[docs]
class _MultiVectorConfigCreate(_MultiVectorConfigCreateBase):
aggregation: Optional[str]
[docs]
class _VectorIndexConfigCreate(_ConfigCreateModel):
distance: Optional[VectorDistances]
multivector: Optional[_MultiVectorConfigCreate]
quantizer: Optional[_QuantizerConfigCreate] = Field(exclude=True)
[docs]
@staticmethod
@abstractmethod
def vector_index_type() -> VectorIndexType: ...
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
if self.quantizer is not None:
ret_dict[self.quantizer.quantizer_name()] = self.quantizer._to_dict()
if self.distance is not None:
ret_dict["distance"] = str(self.distance.value)
return ret_dict
[docs]
class _VectorIndexConfigUpdate(_ConfigUpdateModel):
quantizer: Optional[_QuantizerConfigUpdate] = Field(exclude=True)
[docs]
@staticmethod
@abstractmethod
def vector_index_type() -> VectorIndexType: ...
[docs]
class _VectorIndexConfigSkipCreate(_VectorIndexConfigCreate):
skip: bool = True
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.HNSW
[docs]
class _VectorIndexConfigHNSWCreate(_VectorIndexConfigCreate):
cleanupIntervalSeconds: Optional[int]
dynamicEfMin: Optional[int]
dynamicEfMax: Optional[int]
dynamicEfFactor: Optional[int]
efConstruction: Optional[int]
ef: Optional[int]
filterStrategy: Optional[VectorFilterStrategy]
flatSearchCutoff: Optional[int]
maxConnections: Optional[int]
vectorCacheMaxObjects: Optional[int]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.HNSW
[docs]
class _VectorIndexConfigFlatCreate(_VectorIndexConfigCreate):
vectorCacheMaxObjects: Optional[int]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.FLAT
[docs]
class _VectorIndexConfigHNSWUpdate(_VectorIndexConfigUpdate):
dynamicEfMin: Optional[int]
dynamicEfMax: Optional[int]
dynamicEfFactor: Optional[int]
ef: Optional[int]
filterStrategy: Optional[VectorFilterStrategy]
flatSearchCutoff: Optional[int]
vectorCacheMaxObjects: Optional[int]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.HNSW
[docs]
class _VectorIndexConfigFlatUpdate(_VectorIndexConfigUpdate):
vectorCacheMaxObjects: Optional[int]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.FLAT
[docs]
class _VectorIndexConfigDynamicCreate(_VectorIndexConfigCreate):
threshold: Optional[int]
hnsw: Optional[_VectorIndexConfigHNSWCreate]
flat: Optional[_VectorIndexConfigFlatCreate]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.DYNAMIC
[docs]
def _to_dict(self) -> dict:
ret_dict = super()._to_dict()
if self.hnsw is not None:
ret_dict["hnsw"] = self.hnsw._to_dict()
if self.flat is not None:
ret_dict["flat"] = self.flat._to_dict()
if self.threshold is not None:
ret_dict["threshold"] = self.threshold
return ret_dict
[docs]
class _VectorIndexConfigDynamicUpdate(_VectorIndexConfigUpdate):
threshold: Optional[int]
hnsw: Optional[_VectorIndexConfigHNSWUpdate]
flat: Optional[_VectorIndexConfigFlatUpdate]
[docs]
@staticmethod
def vector_index_type() -> VectorIndexType:
return VectorIndexType.DYNAMIC