import datetime
from dataclasses import dataclass
from dataclasses import fields as _dataclass_fields
from typing import (
Any,
ClassVar,
Dict,
List,
Literal,
Optional,
Sequence,
Type,
TypeVar,
Union,
cast,
)
from pydantic import (
AnyHttpUrl,
Field,
TypeAdapter,
ValidationInfo,
field_validator,
model_validator,
)
from typing_extensions import TypeAlias
from typing_extensions import deprecated as typing_deprecated
from weaviate.collections.classes.config_base import (
_ConfigBase,
_ConfigCreateModel,
_ConfigUpdateModel,
_EnumLikeStr,
_QuantizerConfigUpdate,
)
from weaviate.collections.classes.config_named_vectors import (
_NamedVectorConfigCreate,
_NamedVectorConfigUpdate,
_NamedVectors,
_NamedVectorsUpdate,
)
from weaviate.collections.classes.config_object_ttl import (
_ObjectTTL,
_ObjectTTLConfigCreate,
_ObjectTTLConfigUpdate,
_ObjectTTLUpdate,
)
from weaviate.collections.classes.config_vector_index import (
PQEncoderDistribution,
PQEncoderType,
VectorFilterStrategy,
_BQConfigUpdate,
_PQConfigUpdate,
_PQEncoderConfigUpdate,
_RQConfigUpdate,
_SQConfigUpdate,
_VectorIndex,
_VectorIndexConfigCreate,
_VectorIndexConfigDynamicUpdate,
_VectorIndexConfigFlatUpdate,
_VectorIndexConfigHFreshUpdate,
_VectorIndexConfigHNSWUpdate,
_VectorIndexConfigUpdate,
)
from weaviate.collections.classes.config_vector_index import (
VectorIndexType as VectorIndexTypeAlias,
)
from weaviate.collections.classes.config_vectorizers import (
VectorDistances as VectorDistancesAlias,
)
from weaviate.collections.classes.config_vectorizers import (
Vectorizers as VectorizersAlias,
)
from weaviate.collections.classes.config_vectorizers import (
_Vectorizer,
_VectorizerConfigCreate,
)
from weaviate.collections.classes.config_vectors import (
_MultiVectors,
_VectorConfigCreate,
_VectorConfigUpdate,
_Vectors,
_VectorsUpdate,
)
from weaviate.exceptions import WeaviateInsertInvalidPropertyError, WeaviateInvalidInputError
from weaviate.str_enum import BaseEnum
from weaviate.util import _capitalize_first_letter, docstring_deprecated
from weaviate.warnings import _Warnings
# BC for direct imports
Vectorizers: TypeAlias = VectorizersAlias
VectorIndexType: TypeAlias = VectorIndexTypeAlias
VectorDistances: TypeAlias = VectorDistancesAlias
AWSService: TypeAlias = Literal[
"bedrock",
"sagemaker",
]
OpenAiVerbosity: TypeAlias = Literal[
"low",
"medium",
"high",
]
OpenAiReasoningEffort: TypeAlias = Literal[
"minimal",
"low",
"medium",
"high",
]
IndexName: TypeAlias = Literal[
"searchable",
"filterable",
"rangeFilters",
]
[docs]
class ConsistencyLevel(str, BaseEnum):
"""The consistency levels when writing to Weaviate with replication enabled.
Attributes:
ALL: Wait for confirmation of write success from all, `N`, replicas.
ONE: Wait for confirmation of write success from only one replica.
QUORUM: Wait for confirmation of write success from a quorum: `N/2+1`, of replicas.
"""
ALL = "ALL"
ONE = "ONE"
QUORUM = "QUORUM"
[docs]
class DataType(str, BaseEnum):
"""The available primitive data types in Weaviate.
Attributes:
TEXT: Text data type.
TEXT_ARRAY: Text array data type.
INT: Integer data type.
INT_ARRAY: Integer array data type.
BOOL: Boolean data type.
BOOL_ARRAY: Boolean array data type.
NUMBER: Number data type.
NUMBER_ARRAY: Number array data type.
DATE: Date data type.
DATE_ARRAY: Date array data type.
UUID: UUID data type.
UUID_ARRAY: UUID array data type.
GEO_COORDINATES: Geo coordinates data type.
BLOB: Blob data type.
BLOB_HASH: Blob hash data type.
PHONE_NUMBER: Phone number data type.
OBJECT: Object data type.
OBJECT_ARRAY: Object array data type.
"""
TEXT = "text"
TEXT_ARRAY = "text[]"
INT = "int"
INT_ARRAY = "int[]"
BOOL = "boolean"
BOOL_ARRAY = "boolean[]"
NUMBER = "number"
NUMBER_ARRAY = "number[]"
DATE = "date"
DATE_ARRAY = "date[]"
UUID = "uuid"
UUID_ARRAY = "uuid[]"
GEO_COORDINATES = "geoCoordinates"
BLOB = "blob"
BLOB_HASH = "blobHash"
PHONE_NUMBER = "phoneNumber"
OBJECT = "object"
OBJECT_ARRAY = "object[]"
[docs]
class Tokenization(str, BaseEnum):
"""The available inverted index tokenization methods for text properties in Weaviate.
Attributes:
WORD: Tokenize by word.
WHITESPACE: Tokenize by whitespace.
LOWERCASE: Tokenize by lowercase.
FIELD: Tokenize by field.
GSE: Tokenize using GSE (for Chinese and Japanese).
TRIGRAM: Tokenize into trigrams.
KAGOME_JA: Tokenize using the 'Kagome' tokenizer (for Japanese).
KAGOME_KR: Tokenize using the 'Kagome' tokenizer and a Korean MeCab dictionary (for Korean).
GSE_CH: Tokenize using GSE (for Chinese).
"""
WORD = "word"
WHITESPACE = "whitespace"
LOWERCASE = "lowercase"
FIELD = "field"
GSE = "gse"
TRIGRAM = "trigram"
KAGOME_JA = "kagome_ja"
KAGOME_KR = "kagome_kr"
GSE_CH = "gse_ch"
[docs]
class GenerativeSearches(str, BaseEnum):
"""The available generative search modules in Weaviate.
These modules generate text from text-based inputs.
See the [docs](https://weaviate.io/developers/weaviate/modules/reader-generator-modules) for more details.
Attributes:
AWS: Weaviate module backed by AWS Bedrock generative models.
ANTHROPIC: Weaviate module backed by Anthropic generative models.
ANYSCALE: Weaviate module backed by Anyscale generative models.
COHERE: Weaviate module backed by Cohere generative models.
CONTEXTUALAI: Weaviate module backed by ContextualAI generative models.
DATABRICKS: Weaviate module backed by Databricks generative models.
FRIENDLIAI: Weaviate module backed by FriendliAI generative models.
MISTRAL: Weaviate module backed by Mistral generative models.
NVIDIA: Weaviate module backed by NVIDIA generative models.
OLLAMA: Weaviate module backed by generative models deployed on Ollama infrastructure.
OPENAI: Weaviate module backed by OpenAI and Azure-OpenAI generative models.
PALM: Weaviate module backed by PaLM generative models.
"""
AWS = "generative-aws"
ANTHROPIC = "generative-anthropic"
ANYSCALE = "generative-anyscale"
COHERE = "generative-cohere"
CONTEXTUALAI = "generative-contextualai"
DATABRICKS = "generative-databricks"
DUMMY = "generative-dummy"
FRIENDLIAI = "generative-friendliai"
MISTRAL = "generative-mistral"
NVIDIA = "generative-nvidia"
OLLAMA = "generative-ollama"
OPENAI = "generative-openai"
PALM = "generative-palm" # rename to google once all versions support it
XAI = "generative-xai"
[docs]
class Rerankers(str, BaseEnum):
"""The available reranker modules in Weaviate.
These modules rerank the results of a search query.
See the [docs](https://weaviate.io/developers/weaviate/modules/retriever-vectorizer-modules#re-ranking) for more details.
Attributes:
NONE: No reranker.
COHERE: Weaviate module backed by Cohere reranking models.
CONTEXTUALAI: Weaviate module backed by ContextualAI reranking models.
TRANSFORMERS: Weaviate module backed by Transformers reranking models.
VOYAGEAI: Weaviate module backed by VoyageAI reranking models.
JINAAI: Weaviate module backed by JinaAI reranking models.
NVIDIA: Weaviate module backed by NVIDIA reranking models.
"""
NONE = "none"
COHERE = "reranker-cohere"
CONTEXTUALAI = "reranker-contextualai"
TRANSFORMERS = "reranker-transformers"
VOYAGEAI = "reranker-voyageai"
JINAAI = "reranker-jinaai"
NVIDIA = "reranker-nvidia"
[docs]
class StopwordsPreset(str, BaseEnum):
"""Preset stopwords to use in the `Stopwords` class.
Attributes:
EN: English stopwords.
NONE: No stopwords.
"""
NONE = "none"
EN = "en"
[docs]
class ReplicationDeletionStrategy(str, BaseEnum):
"""How object deletions in multi node environments should be resolved.
Attributes:
PERMANENT_DELETION: Once an object has been deleted on one node it will be deleted on all nodes in case of conflicts.
NO_AUTOMATED_RESOLUTION: No deletion resolution.
"""
DELETE_ON_CONFLICT = "DeleteOnConflict"
NO_AUTOMATED_RESOLUTION = "NoAutomatedResolution"
TIME_BASED_RESOLUTION = "TimeBasedResolution"
[docs]
class _ShardingConfigCreate(_ConfigCreateModel):
virtualPerPhysical: Optional[int]
desiredCount: Optional[int]
desiredVirtualCount: Optional[int]
key: str = "_id"
strategy: str = "hash"
function: str = "murmur3"
[docs]
class _AsyncReplicationConfigCreate(_ConfigCreateModel):
maxWorkers: Optional[int]
hashtreeHeight: Optional[int]
frequency: Optional[int]
frequencyWhilePropagating: Optional[int]
aliveNodesCheckingFrequency: Optional[int]
loggingFrequency: Optional[int]
diffBatchSize: Optional[int]
diffPerNodeTimeout: Optional[int]
prePropagationTimeout: Optional[int]
propagationTimeout: Optional[int]
propagationLimit: Optional[int]
propagationDelay: Optional[int]
propagationConcurrency: Optional[int]
propagationBatchSize: Optional[int]
[docs]
class _AsyncReplicationConfigUpdate(_ConfigUpdateModel):
maxWorkers: Optional[int]
hashtreeHeight: Optional[int]
frequency: Optional[int]
frequencyWhilePropagating: Optional[int]
aliveNodesCheckingFrequency: Optional[int]
loggingFrequency: Optional[int]
diffBatchSize: Optional[int]
diffPerNodeTimeout: Optional[int]
prePropagationTimeout: Optional[int]
propagationTimeout: Optional[int]
propagationLimit: Optional[int]
propagationDelay: Optional[int]
propagationConcurrency: Optional[int]
propagationBatchSize: Optional[int]
[docs]
class _ReplicationConfigCreate(_ConfigCreateModel):
factor: Optional[int]
asyncEnabled: Optional[bool]
asyncConfig: Optional[_AsyncReplicationConfigCreate]
deletionStrategy: Optional[ReplicationDeletionStrategy]
[docs]
class _ReplicationConfigUpdate(_ConfigUpdateModel):
factor: Optional[int]
asyncEnabled: Optional[bool]
asyncConfig: Optional[_AsyncReplicationConfigUpdate]
deletionStrategy: Optional[ReplicationDeletionStrategy]
[docs]
def merge_with_existing(self, schema: Dict[str, Any]) -> Dict[str, Any]:
if self.factor is not None:
schema["factor"] = self.factor
if self.asyncEnabled is not None:
schema["asyncEnabled"] = self.asyncEnabled
if not self.asyncEnabled:
schema.pop("asyncConfig", None)
if self.deletionStrategy is not None:
schema["deletionStrategy"] = str(self.deletionStrategy.value)
if self.asyncConfig is not None:
# Replace entire asyncConfig (like generative/reranker pattern)
# rather than merging, so omitted fields revert to server defaults
schema["asyncConfig"] = self.asyncConfig.model_dump(exclude_none=True)
return schema
[docs]
class _BM25ConfigCreate(_ConfigCreateModel):
b: float
k1: float
[docs]
class _BM25ConfigUpdate(_ConfigUpdateModel):
b: Optional[float]
k1: Optional[float]
[docs]
class _StopwordsCreate(_ConfigCreateModel):
preset: Optional[StopwordsPreset]
additions: Optional[List[str]]
removals: Optional[List[str]]
[docs]
class _StopwordsUpdate(_ConfigUpdateModel):
preset: Optional[StopwordsPreset]
additions: Optional[List[str]]
removals: Optional[List[str]]
[docs]
class _InvertedIndexConfigCreate(_ConfigCreateModel):
bm25: Optional[_BM25ConfigCreate]
cleanupIntervalSeconds: Optional[int]
indexTimestamps: Optional[bool]
indexPropertyLength: Optional[bool]
indexNullState: Optional[bool]
stopwords: _StopwordsCreate
stopwordPresets: Optional[Dict[str, List[str]]] = None
[docs]
class _InvertedIndexConfigUpdate(_ConfigUpdateModel):
bm25: Optional[_BM25ConfigUpdate]
cleanupIntervalSeconds: Optional[int]
stopwords: Optional[_StopwordsUpdate]
stopwordPresets: Optional[Dict[str, List[str]]] = None
[docs]
class _MultiTenancyConfigCreate(_ConfigCreateModel):
enabled: bool
autoTenantCreation: Optional[bool]
autoTenantActivation: Optional[bool]
[docs]
class _MultiTenancyConfigUpdate(_ConfigUpdateModel):
autoTenantCreation: Optional[bool]
autoTenantActivation: Optional[bool]
[docs]
class _GenerativeProvider(_ConfigCreateModel):
generative: Union[GenerativeSearches, _EnumLikeStr]
[docs]
class _GenerativeAnyscale(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.ANYSCALE, frozen=True, exclude=True
)
baseURL: Optional[str]
temperature: Optional[float]
model: Optional[str]
[docs]
class _GenerativeCustom(_GenerativeProvider):
module_config: Optional[Dict[str, Any]]
[docs]
def _to_dict(self) -> Dict[str, Any]:
if self.module_config is None:
return {}
return self.module_config
[docs]
class _GenerativeDatabricks(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.DATABRICKS, frozen=True, exclude=True
)
endpoint: str
maxTokens: Optional[int]
temperature: Optional[float]
topK: Optional[int]
topP: Optional[float]
[docs]
class _GenerativeMistral(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.MISTRAL, frozen=True, exclude=True
)
temperature: Optional[float]
model: Optional[str]
maxTokens: Optional[int]
baseURL: Optional[str]
[docs]
class _GenerativeNvidia(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.NVIDIA, frozen=True, exclude=True
)
temperature: Optional[float]
model: Optional[str]
maxTokens: Optional[int]
baseURL: Optional[str]
[docs]
class _GenerativeXai(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.XAI, frozen=True, exclude=True
)
temperature: Optional[float]
model: Optional[str]
maxTokens: Optional[int]
baseURL: Optional[str]
topP: Optional[float]
[docs]
class _GenerativeFriendliai(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.FRIENDLIAI, frozen=True, exclude=True
)
temperature: Optional[float]
model: Optional[str]
maxTokens: Optional[int]
baseURL: Optional[str]
[docs]
class _GenerativeOllama(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.OLLAMA, frozen=True, exclude=True
)
model: Optional[str]
apiEndpoint: Optional[str]
[docs]
class _GenerativeOpenAIConfigBase(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.OPENAI, frozen=True, exclude=True
)
baseURL: Optional[AnyHttpUrl]
frequencyPenalty: Optional[float]
presencePenalty: Optional[float]
maxTokens: Optional[int]
temperature: Optional[float]
topP: Optional[float]
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
if self.baseURL is not None:
ret_dict["baseURL"] = self.baseURL.unicode_string()
return ret_dict
[docs]
class _GenerativeOpenAIConfig(_GenerativeOpenAIConfigBase):
model: Optional[str]
verbosity: Optional[str]
reasoningEffort: Optional[str]
[docs]
class _GenerativeAzureOpenAIConfig(_GenerativeOpenAIConfigBase):
resourceName: str
deploymentId: str
[docs]
class _GenerativeCohereConfig(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.COHERE, frozen=True, exclude=True
)
baseURL: Optional[AnyHttpUrl]
k: Optional[int]
model: Optional[str]
maxTokens: Optional[int]
stopSequences: Optional[List[str]]
temperature: Optional[float]
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
if self.baseURL is not None:
ret_dict["baseURL"] = self.baseURL.unicode_string()
return ret_dict
[docs]
class _GenerativeContextualAIConfig(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.CONTEXTUALAI, frozen=True, exclude=True
)
model: Optional[str]
temperature: Optional[float]
topP: Optional[float]
maxNewTokens: Optional[int]
systemPrompt: Optional[str]
avoidCommentary: Optional[bool]
knowledge: Optional[List[str]]
[docs]
class _GenerativeGoogleConfig(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.PALM, frozen=True, exclude=True
)
apiEndpoint: Optional[str]
endpointId: Optional[str]
region: Optional[str]
maxOutputTokens: Optional[int]
modelId: Optional[str]
projectId: str
temperature: Optional[float]
topK: Optional[int]
topP: Optional[float]
# TODO - implement passing `stopSequences`, `frequencyPenalty`, `presencePenalty` parameters from collection config once added server side
stopSequences: Optional[list[str]]
frequencyPenalty: Optional[float]
presencePenalty: Optional[float]
[docs]
class _GenerativeAWSConfig(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.AWS, frozen=True, exclude=True
)
region: str
service: str
model: Optional[str]
endpoint: Optional[str]
temperature: Optional[float]
targetModel: Optional[str]
targetVariant: Optional[str]
topK: Optional[int]
topP: Optional[float]
stopSequences: Optional[List[str]]
maxTokens: Optional[int]
[docs]
class _GenerativeAnthropicConfig(_GenerativeProvider):
generative: Union[GenerativeSearches, _EnumLikeStr] = Field(
default=GenerativeSearches.ANTHROPIC, frozen=True, exclude=True
)
model: Optional[str]
maxTokens: Optional[int]
stopSequences: Optional[List[str]]
temperature: Optional[float]
topK: Optional[int]
topP: Optional[float]
[docs]
class _RerankerProvider(_ConfigCreateModel):
reranker: Union[Rerankers, _EnumLikeStr]
RerankerCohereModel = Literal["rerank-english-v2.0", "rerank-multilingual-v2.0"]
[docs]
class _RerankerCohereConfig(_RerankerProvider):
reranker: Union[Rerankers, _EnumLikeStr] = Field(
default=Rerankers.COHERE, frozen=True, exclude=True
)
model: Optional[Union[RerankerCohereModel, str]] = Field(default=None)
baseURL: Optional[AnyHttpUrl]
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
if self.baseURL is not None:
ret_dict["baseURL"] = self.baseURL.unicode_string()
return ret_dict
[docs]
class _RerankerCustomConfig(_RerankerProvider):
module_config: Optional[Dict[str, Any]]
[docs]
def _to_dict(self) -> Dict[str, Any]:
if self.module_config is None:
return {}
return self.module_config
RerankerJinaAIModel = Literal[
"jina-reranker-v2-base-multilingual",
"jina-reranker-v1-base-en",
"jina-reranker-v1-turbo-en",
"jina-reranker-v1-tiny-en",
"jina-colbert-v1-en",
]
[docs]
class _RerankerJinaAIConfig(_RerankerProvider):
reranker: Union[Rerankers, _EnumLikeStr] = Field(
default=Rerankers.JINAAI, frozen=True, exclude=True
)
model: Optional[Union[RerankerJinaAIModel, str]] = Field(default=None)
RerankerVoyageAIModel = Literal["rerank-2", "rerank-2-lite", "rerank-lite-1", "rerank-1"]
[docs]
class _RerankerVoyageAIConfig(_RerankerProvider):
reranker: Union[Rerankers, _EnumLikeStr] = Field(
default=Rerankers.VOYAGEAI, frozen=True, exclude=True
)
model: Optional[Union[RerankerVoyageAIModel, str]] = Field(default=None)
[docs]
class _RerankerNvidiaConfig(_RerankerProvider):
reranker: Union[Rerankers, _EnumLikeStr] = Field(
default=Rerankers.NVIDIA, frozen=True, exclude=True
)
model: Optional[str] = Field(default=None)
baseURL: Optional[AnyHttpUrl]
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
if self.baseURL is not None:
ret_dict["baseURL"] = self.baseURL.unicode_string()
return ret_dict
RerankerContextualAIModel = Literal[
"ctxl-rerank-v2-instruct-multilingual",
"ctxl-rerank-v2-instruct-multilingual-mini",
"ctxl-rerank-v1-instruct",
]
[docs]
class _RerankerContextualAIConfig(_RerankerProvider):
reranker: Union[Rerankers, _EnumLikeStr] = Field(
default=Rerankers.CONTEXTUALAI, frozen=True, exclude=True
)
model: Optional[Union[RerankerContextualAIModel, str]] = Field(default=None)
instruction: Optional[str] = Field(default=None)
topN: Optional[int] = Field(default=None)
[docs]
class _Generative:
"""Use this factory class to create the correct object for the `generative_config` argument in the `collections.create()` method.
Each staticmethod provides options specific to the named generative search module in the function's name. Under-the-hood data validation steps
will ensure that any mis-specifications will be caught before the request is sent to Weaviate.
"""
[docs]
@staticmethod
def anyscale(
model: Optional[str] = None,
temperature: Optional[float] = None,
base_url: Optional[str] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAnyscale` object for use when generating using the `generative-anyscale` module.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeAnyscale(model=model, temperature=temperature, baseURL=base_url)
[docs]
@staticmethod
def custom(
module_name: str,
module_config: Optional[Dict[str, Any]] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeCustom` object for use when generating using a custom specification.
Args:
module_name: The name of the module to use, REQUIRED.
module_config: The configuration to use for the module. Defaults to `None`, which uses the server-defined default.
"""
return _GenerativeCustom(generative=_EnumLikeStr(module_name), module_config=module_config)
[docs]
@staticmethod
def databricks(
*,
endpoint: str,
max_tokens: Optional[int] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeDatabricks` object for use when performing AI generation using the `generative-databricks` module.
Args:
endpoint: The URL where the API request should go. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K value to use. Defaults to `None`, which uses the server-defined default
top_p: The top P value to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeDatabricks(
endpoint=endpoint,
maxTokens=max_tokens,
temperature=temperature,
topK=top_k,
topP=top_p,
)
[docs]
@staticmethod
def friendliai(
*,
base_url: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeFriendliai` object for use when performing AI generation using the `generative-friendliai` module.
Args:
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeFriendliai(
model=model, temperature=temperature, maxTokens=max_tokens, baseURL=base_url
)
[docs]
@staticmethod
def mistral(
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
base_url: Optional[str] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeMistral` object for use when performing AI generation using the `generative-mistral` module.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeMistral(
model=model, temperature=temperature, maxTokens=max_tokens, baseURL=base_url
)
[docs]
@staticmethod
def nvidia(
*,
base_url: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeNvidia` object for use when performing AI generation using the `generative-nvidia` module.
Args:
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeNvidia(
model=model, temperature=temperature, maxTokens=max_tokens, baseURL=base_url
)
[docs]
@staticmethod
def xai(
*,
base_url: Optional[str] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeXai` object for use when performing AI generation using the `generative-xai` module.
Args:
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
top_p: The top P value to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeXai(
model=model, temperature=temperature, maxTokens=max_tokens, topP=top_p, baseURL=base_url
)
[docs]
@staticmethod
def ollama(
*,
api_endpoint: Optional[str] = None,
model: Optional[str] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeOllama` object for use when performing AI generation using the `generative-ollama` module.
Args:
api_endpoint: The API endpoint to use. Defaults to `None`, which uses the server-defined default
Docker users may need to specify an alias, such as `http://host.docker.internal:11434` so that the container can access the host machine.
model: The model to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeOllama(model=model, apiEndpoint=api_endpoint)
[docs]
@staticmethod
def openai(
model: Optional[str] = None,
frequency_penalty: Optional[float] = None,
max_tokens: Optional[int] = None,
presence_penalty: Optional[float] = None,
temperature: Optional[float] = None,
top_p: Optional[float] = None,
base_url: Optional[AnyHttpUrl] = None,
*,
verbosity: Optional[Union[OpenAiVerbosity, str]] = None,
reasoning_effort: Optional[Union[OpenAiReasoningEffort, str]] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeOpenAIConfig` object for use when performing AI generation using the `generative-openai` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-openai)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
frequency_penalty: The frequency penalty to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
presence_penalty: The presence penalty to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
verbosity: The verbosity level to use. One of 'low', 'medium', 'high'. Defaults to `None`, which uses the server-defined default
reasoning_effort: The reasoning effort level to use. One of 'minimal', 'low
"""
return _GenerativeOpenAIConfig(
baseURL=base_url,
frequencyPenalty=frequency_penalty,
maxTokens=max_tokens,
model=model,
presencePenalty=presence_penalty,
temperature=temperature,
topP=top_p,
verbosity=verbosity,
reasoningEffort=reasoning_effort,
)
[docs]
@staticmethod
def azure_openai(
resource_name: str,
deployment_id: str,
frequency_penalty: Optional[float] = None,
max_tokens: Optional[int] = None,
presence_penalty: Optional[float] = None,
temperature: Optional[float] = None,
top_p: Optional[float] = None,
base_url: Optional[AnyHttpUrl] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAzureOpenAIConfig` object for use when performing AI generation using the `generative-openai` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-openai)
for detailed usage.
Args:
resource_name: The name of the Azure OpenAI resource to use.
deployment_id: The Azure OpenAI deployment ID to use.
frequency_penalty: The frequency penalty to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
presence_penalty: The presence penalty to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeAzureOpenAIConfig(
baseURL=base_url,
deploymentId=deployment_id,
frequencyPenalty=frequency_penalty,
maxTokens=max_tokens,
presencePenalty=presence_penalty,
resourceName=resource_name,
temperature=temperature,
topP=top_p,
)
[docs]
@staticmethod
def cohere(
model: Optional[str] = None,
k: Optional[int] = None,
max_tokens: Optional[int] = None,
return_likelihoods: Optional[str] = None,
stop_sequences: Optional[List[str]] = None,
temperature: Optional[float] = None,
base_url: Optional[AnyHttpUrl] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeCohereConfig` object for use when performing AI generation using the `generative-cohere` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-cohere)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
k: The number of sequences to generate. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
return_likelihoods: (Deprecated) The return likelihoods setting to use. No longer has any effect.
stop_sequences: The stop sequences to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL where the API request should go. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeCohereConfig(
baseURL=base_url,
k=k,
maxTokens=max_tokens,
model=model,
stopSequences=stop_sequences,
temperature=temperature,
)
[docs]
@staticmethod
def contextualai(
model: Optional[str] = None,
max_new_tokens: Optional[int] = None,
temperature: Optional[float] = None,
top_p: Optional[float] = None,
system_prompt: Optional[str] = None,
avoid_commentary: Optional[bool] = None,
knowledge: Optional[List[str]] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeContextualAIConfig` object for use when performing AI generation using the `generative-contextualai` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/contextualai/generative)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
max_new_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_p: Nucleus sampling parameter (0 < x <= 1). Defaults to `None`, which uses the server-defined default
system_prompt: System instructions the model follows. Defaults to `None`, which uses the server-defined default
avoid_commentary: If `True`, reduce conversational commentary in responses. Defaults to `None`, which uses the server-defined default
knowledge: Additional detailed knowledge sources with varied information. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeContextualAIConfig(
model=model,
temperature=temperature,
topP=top_p,
maxNewTokens=max_new_tokens,
systemPrompt=system_prompt,
avoidCommentary=avoid_commentary,
knowledge=knowledge,
)
[docs]
@staticmethod
@docstring_deprecated(
deprecated_in="4.9.0",
details="""
This method is deprecated and will be removed in Q2 '25. Please use :meth:`~weaviate.collections.classes.config._Generative.google` instead.
""",
)
@typing_deprecated(
"This method is deprecated and will be removed in Q2 '25. Please use `google` instead."
)
def palm(
project_id: str,
api_endpoint: Optional[str] = None,
max_output_tokens: Optional[int] = None,
model_id: Optional[str] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativePaLMConfig` object for use when performing AI generation using the `generative-palm` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-palm)
for detailed usage.
Args:
project_id: The PalM project ID to use.
api_endpoint: The API endpoint to use without a leading scheme such as `http://`. Defaults to `None`, which uses the server-defined default
max_output_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
model_id: The model ID to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
"""
_Warnings.palm_to_google_gen()
return _GenerativeGoogleConfig(
apiEndpoint=api_endpoint,
region=None,
maxOutputTokens=max_output_tokens,
modelId=model_id,
projectId=project_id,
endpointId=None,
temperature=temperature,
topK=top_k,
topP=top_p,
stopSequences=None,
frequencyPenalty=None,
presencePenalty=None,
)
[docs]
@staticmethod
@typing_deprecated(
"`google()` is deprecated and will be removed after Q3 '26. Use a service-specific method instead, such as `google_vertex` or `google_gemini`."
)
def google(
project_id: str,
api_endpoint: Optional[str] = None,
max_output_tokens: Optional[int] = None,
model_id: Optional[str] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeGoogleConfig` object for use when performing AI generation using the `generative-google` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/google/generative)
for detailed usage.
Args:
project_id: The PalM project ID to use.
api_endpoint: The API endpoint to use without a leading scheme such as `http://`. Defaults to `None`, which uses the server-defined default
max_output_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
model_id: The model ID to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeGoogleConfig(
apiEndpoint=api_endpoint,
region=None,
maxOutputTokens=max_output_tokens,
modelId=model_id,
projectId=project_id,
endpointId=None,
temperature=temperature,
topK=top_k,
topP=top_p,
stopSequences=None,
frequencyPenalty=None,
presencePenalty=None,
)
[docs]
@staticmethod
def google_vertex(
project_id: str,
api_endpoint: Optional[str] = None,
region: Optional[str] = None,
max_output_tokens: Optional[int] = None,
model_id: Optional[str] = None,
endpoint_id: Optional[str] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeGoogleConfig` object for use when performing AI generation using the `generative-google` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/google/generative)
for detailed usage.
Args:
project_id: The Google Vertex project ID to use.
api_endpoint: The API endpoint to use without a leading scheme such as `http://`. Defaults to `None`, which uses the server-defined default
region: The region to use. Defaults to `None`, which uses the server-defined default
max_output_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
model_id: The model ID to use. Defaults to `None`, which uses the server-defined default
endpoint_id: The endpoint ID to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeGoogleConfig(
apiEndpoint=api_endpoint,
region=region,
maxOutputTokens=max_output_tokens,
modelId=model_id,
projectId=project_id,
endpointId=endpoint_id,
temperature=temperature,
topK=top_k,
topP=top_p,
stopSequences=None,
frequencyPenalty=None,
presencePenalty=None,
)
[docs]
@staticmethod
def google_gemini(
max_output_tokens: Optional[int] = None,
model: Optional[str] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeGoogleConfig` object for use when performing AI generation using the `generative-google` module with the Gemini API.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/google/generative)
for detailed usage.
Args:
max_output_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
model: The model to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeGoogleConfig(
apiEndpoint=None,
region=None,
maxOutputTokens=max_output_tokens,
modelId=model,
projectId="",
endpointId=None,
temperature=temperature,
topK=top_k,
topP=top_p,
stopSequences=None,
frequencyPenalty=None,
presencePenalty=None,
)
[docs]
@staticmethod
@typing_deprecated(
"`aws` is deprecated and will be removed after Q3 '26. Use a service-specific method instead, such as `aws_bedrock`."
)
def aws(
model: Optional[str] = None,
region: str = "", # cant have a non-default value after a default value, but we cant change the order for BC
endpoint: Optional[str] = None,
service: Union[AWSService, str] = "bedrock",
max_tokens: Optional[int] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAWSConfig` object for use when performing AI generation using the `generative-aws` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-aws)
for detailed usage.
Args:
model: The model to use, REQUIRED for service "bedrock".
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default.
region: The AWS region to run the model from, REQUIRED.
endpoint: The model to use, REQUIRED for service "sagemaker".
service: The AWS service to use, options are "bedrock" and "sagemaker".
"""
return _GenerativeAWSConfig(
model=model,
region=region,
service=service,
endpoint=endpoint,
temperature=None,
maxTokens=max_tokens,
targetModel=None,
targetVariant=None,
topK=None,
topP=None,
stopSequences=None,
)
[docs]
@staticmethod
def aws_bedrock(
model: str,
region: str,
temperature: Optional[float] = None,
max_tokens: Optional[int] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
stop_sequences: Optional[List[str]] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAWSConfig` object for use when performing AI generation using the `generative-aws` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-aws)
for detailed usage.
Args:
model: The model to use, REQUIRED
region: The AWS region to run the model from, REQUIRED.
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default.
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
stop_sequences: The stop sequences to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeAWSConfig(
model=model,
region=region,
service="bedrock",
endpoint=None,
temperature=temperature,
maxTokens=max_tokens,
targetModel=None,
targetVariant=None,
topK=top_k,
topP=top_p,
stopSequences=stop_sequences,
)
[docs]
@staticmethod
def aws_sagemaker(
region: str,
endpoint: str,
max_tokens: Optional[int] = None,
target_model: Optional[str] = None,
target_variant: Optional[str] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
stop_sequences: Optional[List[str]] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAWSConfig` object for use when performing AI generation using the `generative-aws` module.
See the [documentation](https://weaviate.io/developers/weaviate/modules/reader-generator-modules/generative-aws)
for detailed usage.
Args:
region: The AWS region to run the model from, REQUIRED.
endpoint: The model to use, REQUIRED.
max_tokens: The maximum token count to generate. Defaults to `None`, which uses the server-defined default.
target_model: The target model to use. Defaults to `None`, which uses the server-defined default
target_variant: The target variant to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
stop_sequences: The stop sequences to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeAWSConfig(
model=None,
region=region,
service="sagemaker",
endpoint=endpoint,
temperature=temperature,
maxTokens=max_tokens,
targetModel=target_model,
targetVariant=target_variant,
topK=top_k,
topP=top_p,
stopSequences=stop_sequences,
)
[docs]
@staticmethod
def anthropic(
model: Optional[str] = None,
max_tokens: Optional[int] = None,
stop_sequences: Optional[List[str]] = None,
temperature: Optional[float] = None,
top_k: Optional[int] = None,
top_p: Optional[float] = None,
) -> _GenerativeProvider:
"""Create a `_GenerativeAnthropicConfig` object for use when performing AI generation using the `generative-anthropic` module.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
max_tokens: The maximum number of tokens to generate. Defaults to `None`, which uses the server-defined default
stop_sequences: The stop sequences to use. Defaults to `None`, which uses the server-defined default
temperature: The temperature to use. Defaults to `None`, which uses the server-defined default
top_k: The top K to use. Defaults to `None`, which uses the server-defined default
top_p: The top P to use. Defaults to `None`, which uses the server-defined default
"""
return _GenerativeAnthropicConfig(
model=model,
maxTokens=max_tokens,
stopSequences=stop_sequences,
temperature=temperature,
topK=top_k,
topP=top_p,
)
[docs]
class _Reranker:
"""Use this factory class to create the correct object for the `reranker_config` argument in the `collections.create()` method.
Each staticmethod provides options specific to the named reranker in the function's name. Under-the-hood data validation steps
will ensure that any mis-specifications will be caught before the request is sent to Weaviate.
"""
[docs]
@staticmethod
def custom(
module_name: str, module_config: Optional[Dict[str, Any]] = None
) -> _RerankerProvider:
"""Create a `_RerankerCustomConfig` object for use when reranking using a custom module.
Args:
module_name: The name of the module to use, REQUIRED.
module_config: The configuration to use for the module. Defaults to `None`, which uses the server-defined default.
"""
return _RerankerCustomConfig(
reranker=_EnumLikeStr(module_name), module_config=module_config
)
[docs]
@staticmethod
def cohere(
model: Optional[Union[RerankerCohereModel, str]] = None,
base_url: Optional[str] = None,
) -> _RerankerProvider:
"""Create a `_RerankerCohereConfig` object for use when reranking using the `reranker-cohere` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/cohere/reranker)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL to send the reranker requests to. Defaults to `None`, which uses the server-defined default.
"""
return _RerankerCohereConfig(
model=model,
baseURL=TypeAdapter(AnyHttpUrl).validate_python(base_url)
if base_url is not None
else None,
)
[docs]
@staticmethod
def jinaai(
model: Optional[Union[RerankerJinaAIModel, str]] = None,
) -> _RerankerProvider:
"""Create a `_RerankerJinaAIConfig` object for use when reranking using the `reranker-jinaai` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/jinaai/reranker)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
"""
return _RerankerJinaAIConfig(model=model)
[docs]
@staticmethod
def voyageai(
model: Optional[Union[RerankerVoyageAIModel, str]] = None,
) -> _RerankerProvider:
"""Create a `_RerankerVoyageAIConfig` object for use when reranking using the `reranker-voyageai` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/voyageai/reranker)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
"""
return _RerankerVoyageAIConfig(model=model)
[docs]
@staticmethod
def nvidia(
model: Optional[str] = None,
base_url: Optional[AnyHttpUrl] = None,
) -> _RerankerProvider:
"""Create a `_RerankerNvidiaConfig` object for use when reranking using the `reranker-nvidia` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/nvidia/reranker)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
base_url: The base URL to send the reranker requests to. Defaults to `None`, which uses the server-defined default.
"""
return _RerankerNvidiaConfig(model=model, baseURL=base_url)
[docs]
@staticmethod
def contextualai(
model: Optional[str] = None,
instruction: Optional[str] = None,
top_n: Optional[int] = None,
) -> _RerankerProvider:
"""Create a `_RerankerContextualAIConfig` object for use when reranking using the `reranker-contextualai` module.
See the [documentation](https://weaviate.io/developers/weaviate/model-providers/contextualai/reranker)
for detailed usage.
Args:
model: The model to use. Defaults to `None`, which uses the server-defined default
instruction: Custom instructions for reranking. Defaults to `None`.
top_n: Number of top results to return. Defaults to `None`, which uses the server-defined default.
"""
return _RerankerContextualAIConfig(model=model, instruction=instruction, topN=top_n)
[docs]
class _CollectionConfigUpdate(_ConfigUpdateModel):
description: Optional[str] = Field(default=None)
property_descriptions: Optional[Dict[str, str]] = Field(default=None)
invertedIndexConfig: Optional[_InvertedIndexConfigUpdate] = Field(
default=None, alias="inverted_index_config"
)
objectTTLConfig: Optional[_ObjectTTLConfigUpdate] = Field(
default=None, alias="object_ttl_config"
)
replicationConfig: Optional[_ReplicationConfigUpdate] = Field(
default=None, alias="replication_config"
)
vectorIndexConfig: Optional[_VectorIndexConfigUpdate] = Field(
default=None, alias="vector_index_config"
)
vectorizerConfig: Optional[Union[_VectorIndexConfigUpdate, List[_NamedVectorConfigUpdate]]] = (
Field(default=None, alias="vectorizer_config")
)
vectorConfig: Optional[Union[_VectorConfigUpdate, List[_VectorConfigUpdate]]] = Field(
default=None, alias="vector_config"
)
multiTenancyConfig: Optional[_MultiTenancyConfigUpdate] = Field(
default=None, alias="multi_tenancy_config"
)
generativeConfig: Optional[_GenerativeProvider] = Field(default=None, alias="generative_config")
rerankerConfig: Optional[_RerankerProvider] = Field(default=None, alias="reranker_config")
[docs]
@field_validator("vectorConfig", mode="before")
def mutual_exclusivity(
cls,
v: Optional[Union[_VectorConfigUpdate, List[_VectorConfigUpdate]]],
info: ValidationInfo,
):
if v is None:
return v
if info.data["vectorizerConfig"] is not None:
raise ValueError(
"Cannot specify vectorizerConfig when also specifying vectorConfig. Please use one or the other."
)
if info.data["vectorIndexConfig"] is not None:
raise ValueError(
"Cannot specify vectorIndexConfig when also specifying vectorConfig. Please use one or the other."
)
return v
def __check_quantizers(
self,
quantizer: Optional[_QuantizerConfigUpdate],
vector_index_config: dict,
) -> None:
if (
(
isinstance(quantizer, _PQConfigUpdate)
and (
vector_index_config.get("bq", {"enabled": False})["enabled"]
or vector_index_config.get("sq", {"enabled": False})["enabled"]
or vector_index_config.get("rq", {"enabled": False})["enabled"]
)
)
or (
isinstance(quantizer, _BQConfigUpdate)
and (
vector_index_config["pq"]["enabled"]
or vector_index_config.get("sq", {"enabled": False})["enabled"]
or vector_index_config.get("rq", {"enabled": False})["enabled"]
)
)
or (
isinstance(quantizer, _SQConfigUpdate)
and (
vector_index_config["pq"]["enabled"]
or vector_index_config.get("bq", {"enabled": False})["enabled"]
or vector_index_config.get("rq", {"enabled": False})["enabled"]
)
)
or (
isinstance(quantizer, _RQConfigUpdate)
and (
vector_index_config["pq"]["enabled"]
or vector_index_config.get("bq", {"enabled": False})["enabled"]
or vector_index_config.get("sq", {"enabled": False})["enabled"]
)
)
):
raise WeaviateInvalidInputError(
f"Cannot update vector index config {vector_index_config} to change its quantizer. To do this, you must recreate the collection."
)
return None
[docs]
def merge_with_existing(self, schema: Dict[str, Any]) -> Dict[str, Any]:
if self.description is not None:
schema["description"] = self.description
if self.property_descriptions is not None:
if (p := schema["properties"]) is None:
raise WeaviateInvalidInputError(
"Cannot update property descriptions without existing properties in the schema"
)
props = {prop["name"]: prop for prop in p}
for prop_name, prop_desc in self.property_descriptions.items():
if prop_name not in props:
raise WeaviateInvalidInputError(
f"Property {prop_name} does not exist in the existing properties"
)
props[prop_name]["description"] = prop_desc
if self.invertedIndexConfig is not None:
schema["invertedIndexConfig"] = self.invertedIndexConfig.merge_with_existing(
schema["invertedIndexConfig"]
)
if self.replicationConfig is not None:
schema["replicationConfig"] = self.replicationConfig.merge_with_existing(
schema["replicationConfig"]
)
if self.multiTenancyConfig is not None:
schema["multiTenancyConfig"] = self.multiTenancyConfig.merge_with_existing(
schema["multiTenancyConfig"]
)
if self.objectTTLConfig is not None:
schema["objectTTLConfig"] = self.objectTTLConfig.merge_with_existing(
schema.get("objectTTLConfig", {})
)
if self.vectorIndexConfig is not None:
self.__check_quantizers(self.vectorIndexConfig.quantizer, schema["vectorIndexConfig"])
schema["vectorIndexConfig"] = self.vectorIndexConfig.merge_with_existing(
schema["vectorIndexConfig"]
)
if self.generativeConfig is not None:
# clear any existing generative config
if "moduleConfig" in schema:
schema["moduleConfig"] = {
k: v for k, v in schema["moduleConfig"].items() if "generative" not in k
}
self.__add_to_module_config(
schema,
self.generativeConfig.generative.value,
self.generativeConfig._to_dict(),
)
if self.rerankerConfig is not None:
# clear any existing reranker config
if "moduleConfig" in schema:
schema["moduleConfig"] = {
k: v for k, v in schema["moduleConfig"].items() if "reranker" not in k
}
self.__add_to_module_config(
schema,
self.rerankerConfig.reranker.value,
self.rerankerConfig._to_dict(),
)
if self.vectorizerConfig is not None:
if isinstance(self.vectorizerConfig, _VectorIndexConfigUpdate):
self.__check_quantizers(
self.vectorizerConfig.quantizer, schema["vectorIndexConfig"]
)
schema["vectorIndexConfig"] = self.vectorizerConfig.merge_with_existing(
schema["vectorIndexConfig"]
)
else:
for vc in self.vectorizerConfig:
if vc.name not in schema["vectorConfig"]:
raise WeaviateInvalidInputError(
f"Vector config with name {vc.name} does not exist in the existing vector config"
)
self.__check_quantizers(
vc.vectorIndexConfig.quantizer,
schema["vectorConfig"][vc.name]["vectorIndexConfig"],
)
schema["vectorConfig"][vc.name]["vectorIndexConfig"] = (
vc.vectorIndexConfig.merge_with_existing(
schema["vectorConfig"][vc.name]["vectorIndexConfig"]
)
)
schema["vectorConfig"][vc.name]["vectorIndexType"] = (
vc.vectorIndexConfig.vector_index_type()
)
if self.vectorConfig is not None:
vcs = (
[self.vectorConfig]
if isinstance(self.vectorConfig, _VectorConfigUpdate)
else self.vectorConfig
)
for vc in vcs:
if vc.name not in schema["vectorConfig"]:
raise WeaviateInvalidInputError(
f"Vector config with name {vc.name} does not exist in the existing vector config"
)
self.__check_quantizers(
vc.vectorIndexConfig.quantizer,
schema["vectorConfig"][vc.name]["vectorIndexConfig"],
)
schema["vectorConfig"][vc.name]["vectorIndexConfig"] = (
vc.vectorIndexConfig.merge_with_existing(
schema["vectorConfig"][vc.name]["vectorIndexConfig"]
)
)
schema["vectorConfig"][vc.name]["vectorIndexType"] = (
vc.vectorIndexConfig.vector_index_type()
)
return schema
@staticmethod
def __add_to_module_config(
return_dict: Dict[str, Any], addition_key: str, addition_val: Dict[str, Any]
) -> None:
if "moduleConfig" not in return_dict:
return_dict["moduleConfig"] = {addition_key: addition_val}
else:
return_dict["moduleConfig"][addition_key] = addition_val
[docs]
@dataclass
class _BM25Config(_ConfigBase):
b: float
k1: float
BM25Config = _BM25Config
[docs]
@dataclass
class _StopwordsConfig(_ConfigBase):
preset: StopwordsPreset
additions: Optional[List[str]]
removals: Optional[List[str]]
StopwordsConfig = _StopwordsConfig
StopwordsCreate = _StopwordsCreate
# Invariant: the read-side dataclass (_StopwordsConfig) and the write-side
# pydantic model (_StopwordsCreate) must carry the same set of field names so
# that values round-tripped from ``collection.config.get()`` can flow back into
# ``tokenization.text()`` without silent data loss. If a field is added to one
# but not the other, importing this module fails loudly; the read→write
# conversion in ``weaviate/tokenization/executor.py::_TokenizationExecutor.text``
# depends on this parity.
_read_fields = {f.name for f in _dataclass_fields(_StopwordsConfig)}
_write_fields = set(_StopwordsCreate.model_fields.keys())
if _read_fields != _write_fields:
raise RuntimeError(
"_StopwordsConfig / _StopwordsCreate field drift detected — "
f"read-only={_read_fields - _write_fields}, "
f"write-only={_write_fields - _read_fields}. "
"Update both classes together, or adapt the read→write conversion in "
"weaviate/tokenization/executor.py::_TokenizationExecutor.text."
)
del _read_fields, _write_fields
[docs]
@dataclass
class _InvertedIndexConfig(_ConfigBase):
bm25: BM25Config
cleanup_interval_seconds: int
index_null_state: bool
index_property_length: bool
index_timestamps: bool
stopwords: StopwordsConfig
stopword_presets: Optional[Dict[str, List[str]]] = None
InvertedIndexConfig = _InvertedIndexConfig
[docs]
@dataclass
class _MultiTenancyConfig(_ConfigBase):
enabled: bool
auto_tenant_creation: bool
auto_tenant_activation: bool
MultiTenancyConfig = _MultiTenancyConfig
[docs]
@dataclass
class _PropertyVectorizerConfig:
skip: bool
vectorize_property_name: bool
PropertyVectorizerConfig = _PropertyVectorizerConfig
[docs]
@dataclass
class _TextAnalyzerConfig(_ConfigBase):
ascii_fold: bool
ascii_fold_ignore: Optional[List[str]]
stopword_preset: Optional[str]
TextAnalyzerConfig = _TextAnalyzerConfig
[docs]
@dataclass
class _NestedProperty(_ConfigBase):
data_type: DataType
description: Optional[str]
index_filterable: bool
index_searchable: bool
name: str
nested_properties: Optional[List["NestedProperty"]]
text_analyzer: Optional[_TextAnalyzerConfig]
tokenization: Optional[Tokenization]
[docs]
def to_dict(self) -> Dict[str, Any]:
out = super().to_dict()
out["dataType"] = [str(self.data_type.value)]
if self.nested_properties is not None and len(self.nested_properties) > 0:
out["nestedProperties"] = [np.to_dict() for np in self.nested_properties]
return out
NestedProperty = _NestedProperty
[docs]
@dataclass
class _PropertyBase(_ConfigBase):
name: str
description: Optional[str]
[docs]
def to_dict(self) -> Dict[str, Any]:
out = {"name": self.name}
if self.description is not None:
out["description"] = self.description
return out
[docs]
@dataclass
class _Property(_PropertyBase):
data_type: DataType
index_filterable: bool
index_range_filters: bool
index_searchable: bool
nested_properties: Optional[List[NestedProperty]]
text_analyzer: Optional[_TextAnalyzerConfig]
tokenization: Optional[Tokenization]
vectorizer_config: Optional[PropertyVectorizerConfig]
vectorizer: Optional[str]
vectorizer_configs: Optional[Dict[str, PropertyVectorizerConfig]]
[docs]
def to_dict(self) -> Dict[str, Any]:
out = super().to_dict()
out["dataType"] = [self.data_type.value]
out["indexFilterable"] = self.index_filterable
out["indexSearchable"] = self.index_searchable
out["indexRangeFilters"] = self.index_range_filters
out["tokenization"] = self.tokenization.value if self.tokenization else None
if self.text_analyzer is not None:
out["textAnalyzer"] = self.text_analyzer.to_dict()
if self.nested_properties is not None and len(self.nested_properties) > 0:
out["nestedProperties"] = [np.to_dict() for np in self.nested_properties]
module_config: Dict[str, Any] = {}
if self.vectorizer is not None:
module_config[self.vectorizer] = {}
if self.vectorizer_config is not None:
assert self.vectorizer is not None
module_config[self.vectorizer] = {
"skip": self.vectorizer_config.skip,
"vectorizePropertyName": self.vectorizer_config.vectorize_property_name,
}
if self.vectorizer_configs is not None:
module_config = {
k: {"skip": v.skip, "vectorizePropertyName": v.vectorize_property_name}
for k, v in self.vectorizer_configs.items()
}
if len(module_config) > 0:
out["moduleConfig"] = module_config
return out
PropertyConfig = _Property
[docs]
@dataclass
class _ReferenceProperty(_PropertyBase):
target_collections: List[str]
[docs]
def to_dict(self) -> Dict[str, Any]:
out = super().to_dict()
out["dataType"] = self.target_collections
return out
ReferencePropertyConfig = _ReferenceProperty
[docs]
@dataclass
class _AsyncReplicationConfig(_ConfigBase):
max_workers: Optional[int]
hashtree_height: Optional[int]
frequency: Optional[int]
frequency_while_propagating: Optional[int]
alive_nodes_checking_frequency: Optional[int]
logging_frequency: Optional[int]
diff_batch_size: Optional[int]
diff_per_node_timeout: Optional[int]
pre_propagation_timeout: Optional[int]
propagation_timeout: Optional[int]
propagation_limit: Optional[int]
propagation_delay: Optional[int]
propagation_concurrency: Optional[int]
propagation_batch_size: Optional[int]
AsyncReplicationConfig = _AsyncReplicationConfig
[docs]
@dataclass
class _ReplicationConfig(_ConfigBase):
factor: int
async_enabled: bool
deletion_strategy: ReplicationDeletionStrategy
async_config: Optional[_AsyncReplicationConfig] = None
ReplicationConfig = _ReplicationConfig
[docs]
@dataclass
class _ShardingConfig(_ConfigBase):
virtual_per_physical: int
desired_count: int
actual_count: int
desired_virtual_count: int
actual_virtual_count: int
key: str
strategy: str
function: str
ShardingConfig = _ShardingConfig
[docs]
@dataclass
class _PQEncoderConfig(_ConfigBase):
type_: PQEncoderType
distribution: PQEncoderDistribution
[docs]
def to_dict(self) -> Dict[str, Any]:
ret_dict = super().to_dict()
ret_dict["type"] = str(ret_dict.pop("type"))
ret_dict["distribution"] = str(ret_dict.pop("distribution"))
return ret_dict
PQEncoderConfig = _PQEncoderConfig
[docs]
@dataclass
class _PQConfig(_ConfigBase):
internal_bit_compression: bool
segments: int
centroids: int
training_limit: int
encoder: PQEncoderConfig
@property
def bit_compression(self) -> bool:
_Warnings.bit_compression_in_pq_config()
return self.internal_bit_compression
PQConfig = _PQConfig
[docs]
@dataclass
class _BQConfig(_ConfigBase):
cache: Optional[bool]
rescore_limit: int
[docs]
@dataclass
class _SQConfig(_ConfigBase):
rescore_limit: int
training_limit: int
[docs]
@dataclass
class _RQConfig(_ConfigBase):
cache: Optional[bool]
bits: Optional[int]
rescore_limit: int
BQConfig = _BQConfig
SQConfig = _SQConfig
RQConfig = _RQConfig
[docs]
@dataclass
class _MuveraConfig(_ConfigBase):
enabled: Optional[bool]
ksim: Optional[int]
dprojections: Optional[int]
repetitions: Optional[int]
MuveraConfig = _MuveraConfig
[docs]
@dataclass
class _MultiVectorConfig(_ConfigBase):
encoding: Optional[_MuveraConfig]
aggregation: str
MultiVector = _MultiVectorConfig
[docs]
@dataclass
class _VectorIndexConfig(_ConfigBase):
multi_vector: Optional[_MultiVectorConfig]
quantizer: Optional[Union[PQConfig, BQConfig, SQConfig, RQConfig]]
[docs]
def to_dict(self) -> Dict[str, Any]:
out = super().to_dict()
if isinstance(self.quantizer, _PQConfig):
out["pq"] = {**out.pop("quantizer"), "enabled": True}
elif isinstance(self.quantizer, _BQConfig):
out["bq"] = {**out.pop("quantizer"), "enabled": True}
elif isinstance(self.quantizer, _SQConfig):
out["sq"] = {**out.pop("quantizer"), "enabled": True}
elif isinstance(self.quantizer, _RQConfig):
out["rq"] = {**out.pop("quantizer"), "enabled": True}
return out
[docs]
@dataclass
class _VectorIndexConfigHNSW(_VectorIndexConfig):
cleanup_interval_seconds: int
distance_metric: VectorDistances
dynamic_ef_min: int
dynamic_ef_max: int
dynamic_ef_factor: int
ef: int
ef_construction: int
filter_strategy: VectorFilterStrategy
flat_search_cutoff: int
max_connections: int
skip: bool
vector_cache_max_objects: int
[docs]
@staticmethod
def vector_index_type() -> str:
return VectorIndexType.HNSW.value
VectorIndexConfigHNSW = _VectorIndexConfigHNSW
[docs]
@dataclass
class _VectorIndexConfigHFresh(_VectorIndexConfig):
distance_metric: VectorDistances
max_posting_size_kb: int
replicas: int
search_probe: int
[docs]
@staticmethod
def vector_index_type() -> str:
return VectorIndexType.HFRESH.value
[docs]
def to_dict(self) -> Dict[str, Any]:
out = super().to_dict()
if "maxPostingSizeKb" in out:
out["maxPostingSizeKB"] = out.pop("maxPostingSizeKb")
return out
VectorIndexConfigHFresh = _VectorIndexConfigHFresh
[docs]
@dataclass
class _VectorIndexConfigFlat(_VectorIndexConfig):
distance_metric: VectorDistances
vector_cache_max_objects: int
[docs]
@staticmethod
def vector_index_type() -> str:
return VectorIndexType.FLAT.value
VectorIndexConfigFlat = _VectorIndexConfigFlat
[docs]
@dataclass
class _VectorIndexConfigDynamic(_ConfigBase):
distance_metric: VectorDistances
hnsw: Optional[VectorIndexConfigHNSW]
flat: Optional[VectorIndexConfigFlat]
threshold: Optional[int]
[docs]
@staticmethod
def vector_index_type() -> str:
return VectorIndexType.DYNAMIC.value
VectorIndexConfigDynamic = _VectorIndexConfigDynamic
[docs]
@dataclass
class _GenerativeConfig(_ConfigBase):
generative: Union[GenerativeSearches, str]
model: Dict[str, Any]
GenerativeConfig = _GenerativeConfig
[docs]
@dataclass
class _VectorizerConfig(_ConfigBase):
vectorizer: Union[Vectorizers, str]
model: Dict[str, Any]
vectorize_collection_name: bool
VectorizerConfig = _VectorizerConfig
[docs]
@dataclass
class _RerankerConfig(_ConfigBase):
model: Dict[str, Any]
reranker: Union[Rerankers, str]
RerankerConfig = _RerankerConfig
[docs]
@dataclass
class _NamedVectorizerConfig(_ConfigBase):
vectorizer: Union[Vectorizers, str]
model: Dict[str, Any]
source_properties: Optional[List[str]]
[docs]
def to_dict(self) -> Dict[str, Any]:
ret_dict = super().to_dict()
if "sourceProperties" in ret_dict:
ret_dict["properties"] = ret_dict.pop("sourceProperties")
return ret_dict
[docs]
@dataclass
class _NamedVectorConfig(_ConfigBase):
vectorizer: _NamedVectorizerConfig
vector_index_config: Union[
VectorIndexConfigHNSW,
VectorIndexConfigFlat,
VectorIndexConfigDynamic,
VectorIndexConfigHFresh,
]
[docs]
def to_dict(self) -> Dict:
ret_dict = super().to_dict()
ret_dict["vectorIndexType"] = self.vector_index_config.vector_index_type()
return ret_dict
NamedVectorConfig = _NamedVectorConfig
[docs]
@dataclass
class _ObjectTTLConfig(_ConfigBase):
enabled: bool
time_to_live: Optional[datetime.timedelta]
filter_expired_objects: bool
delete_on: Union[str, Literal["updateTime"], Literal["creationTime"]]
[docs]
def to_dict(self) -> dict:
delete_on = self.delete_on
if delete_on == "creationTime":
delete_on = "_creationTimeUnix"
elif delete_on == "updateTime":
delete_on = "_lastUpdateTimeUnix"
out: dict = {
"enabled": self.enabled,
"filterExpiredObjects": self.filter_expired_objects,
"deleteOn": delete_on,
}
if self.time_to_live is not None:
out["defaultTtl"] = int(self.time_to_live.total_seconds())
return out
ObjectTTLConfig = _ObjectTTLConfig
[docs]
@dataclass
class _CollectionConfig(_ConfigBase):
name: str
description: Optional[str]
generative_config: Optional[GenerativeConfig]
inverted_index_config: InvertedIndexConfig
multi_tenancy_config: MultiTenancyConfig
object_ttl_config: Optional[ObjectTTLConfig]
properties: List[PropertyConfig]
references: List[ReferencePropertyConfig]
replication_config: ReplicationConfig
reranker_config: Optional[RerankerConfig]
sharding_config: Optional[ShardingConfig]
vector_index_config: Union[
VectorIndexConfigHNSW,
VectorIndexConfigFlat,
VectorIndexConfigDynamic,
VectorIndexConfigHFresh,
None,
]
vector_index_type: Optional[VectorIndexType]
vectorizer_config: Optional[VectorizerConfig]
vectorizer: Optional[Union[Vectorizers, str]]
vector_config: Optional[Dict[str, _NamedVectorConfig]]
[docs]
def to_dict(self) -> dict:
out = super().to_dict()
out["class"] = out.pop("name")
out["moduleConfig"] = {}
for name in [
("generativeConfig", "generative"),
("vectorizerConfig", "vectorizer"),
("rerankerConfig", "reranker"),
]:
if name[0] not in out:
continue
val = out.pop(name[0])
module_name = val[name[1]]
out["moduleConfig"][module_name] = val.get("model", {})
vectorize_collection_name = val.get("vectorizeCollectionName", None)
if vectorize_collection_name is not None:
out["moduleConfig"][module_name]["vectorizeClassName"] = vectorize_collection_name
if "vectorConfig" in out:
for k, v in out["vectorConfig"].items():
extra_values = v["vectorizer"].pop("model", {})
vectorizer = v["vectorizer"].pop("vectorizer")
out["vectorConfig"][k]["vectorizer"] = {
vectorizer: {**extra_values, **v["vectorizer"]}
}
# remove default values for single vector setup
out.pop(
"vectorIndexType", None
) # if doesn't exist (in the case of named vectors) then do nothing
out.pop(
"vectorIndexConfig", None
) # if doesn't exist (in the case of named vectors) then do nothing
out["properties"] = [
*[prop.to_dict() for prop in self.properties],
*[prop.to_dict() for prop in self.references],
]
out.pop("references")
return out
CollectionConfig = _CollectionConfig
[docs]
@dataclass
class _CollectionConfigSimple(_ConfigBase):
name: str
description: Optional[str]
generative_config: Optional[GenerativeConfig]
properties: List[PropertyConfig]
references: List[ReferencePropertyConfig]
reranker_config: Optional[RerankerConfig]
vectorizer_config: Optional[VectorizerConfig]
vectorizer: Optional[Union[Vectorizers, str]]
vector_config: Optional[Dict[str, _NamedVectorConfig]]
object_ttl_config: Optional[ObjectTTLConfig]
CollectionConfigSimple = _CollectionConfigSimple
ShardTypes = Literal["READONLY", "READY", "INDEXING"]
[docs]
@dataclass
class _ShardStatus:
name: str
status: ShardTypes
vector_queue_size: int
ShardStatus = _ShardStatus
[docs]
class _TextAnalyzerConfigCreate(_ConfigCreateModel):
"""Text analysis options for a property.
Configures per-property text analysis for `text` and `text[]` properties that use an
inverted index (searchable or filterable). Supports ASCII folding (accent/diacritic
handling) and selecting a stopword preset that overrides the collection-level
`invertedIndexConfig.stopwords` setting for this property only.
Attributes:
ascii_fold: If True, accent/diacritic marks are folded to their base characters
during indexing and search (e.g. 'école' matches 'ecole'). If omitted, the
field is not sent to the server and the server default (False) applies.
ascii_fold_ignore: Optional list of characters that should be excluded from
ASCII folding (e.g. ['é'] keeps 'é' from being folded to 'e'). If omitted,
the field is not sent to the server.
stopword_preset: Stopword preset name. Overrides the collection-level
`invertedIndexConfig.stopwords` for this property. Only applies to
properties using `Tokenization.WORD`. Accepts a built-in preset
(`StopwordsPreset.EN` or `StopwordsPreset.NONE`) or the name of a
user-defined preset declared in
`Configure.inverted_index(stopword_presets=...)`.
All settings are immutable after the property is created.
"""
asciiFold: Optional[bool] = Field(default=None, alias="ascii_fold")
asciiFoldIgnore: Optional[List[str]] = Field(default=None, alias="ascii_fold_ignore")
stopwordPreset: Optional[Union[StopwordsPreset, str]] = Field(
default=None, alias="stopword_preset"
)
[docs]
@model_validator(mode="after")
def _validate_ascii_fold_ignore(self) -> "_TextAnalyzerConfigCreate":
if self.asciiFold is not True and self.asciiFoldIgnore is not None:
raise ValueError("asciiFoldIgnore cannot be set when asciiFold is not enabled")
return self
TextAnalyzerConfigCreate = _TextAnalyzerConfigCreate
[docs]
class Property(_ConfigCreateModel):
"""This class defines the structure of a data property that a collection can have within Weaviate.
Attributes:
name: The name of the property, REQUIRED.
data_type: The data type of the property, REQUIRED.
description: A description of the property.
index_filterable: Whether the property should be filterable in the inverted index.
index_range_filters: Whether the property should support range filters in the inverted index.
index_searchable: Whether the property should be searchable in the inverted index.
nested_properties: nested properties for data type OBJECT and OBJECT_ARRAY`.
skip_vectorization: Whether to skip vectorization of the property. Defaults to `False`.
text_analyzer: Text analysis options for the property. Configures ASCII folding
behavior for text and text[] properties using an inverted index. Immutable
after the property is created.
tokenization: The tokenization method to use for the inverted index. Defaults to `None`.
vectorize_property_name: Whether to vectorize the property name. Defaults to `True`.
"""
name: str
dataType: DataType = Field(default=..., alias="data_type")
description: Optional[str] = Field(default=None)
indexFilterable: Optional[bool] = Field(default=None, alias="index_filterable")
indexSearchable: Optional[bool] = Field(default=None, alias="index_searchable")
indexRangeFilters: Optional[bool] = Field(default=None, alias="index_range_filters")
nestedProperties: Optional[Union["Property", List["Property"]]] = Field(
default=None, alias="nested_properties"
)
skip_vectorization: bool = Field(default=False)
textAnalyzer: Optional[_TextAnalyzerConfigCreate] = Field(default=None, alias="text_analyzer")
tokenization: Optional[Tokenization] = Field(default=None)
vectorize_property_name: bool = Field(default=True)
[docs]
@field_validator("name")
def _check_name(cls, v: str) -> str:
if v == "vector":
raise ValueError(f"Property name '{v}' is reserved and cannot be used")
return v
[docs]
def _to_dict(
self, vectorizers: Optional[Sequence[Union[Vectorizers, _EnumLikeStr]]] = None
) -> Dict[str, Any]:
ret_dict = super()._to_dict()
ret_dict["dataType"] = [ret_dict["dataType"]]
if vectorizers is not None:
for vectorizer in vectorizers:
if vectorizer is not None and vectorizer != Vectorizers.NONE:
if "moduleConfig" not in ret_dict:
ret_dict["moduleConfig"] = {}
ret_dict["moduleConfig"][vectorizer.value] = {
"skip": self.skip_vectorization,
"vectorizePropertyName": self.vectorize_property_name,
}
del ret_dict["skip_vectorization"]
del ret_dict["vectorize_property_name"]
if self.nestedProperties is not None:
ret_dict["nestedProperties"] = (
[prop._to_dict() for prop in self.nestedProperties]
if isinstance(self.nestedProperties, list)
else [self.nestedProperties._to_dict()]
)
if self.textAnalyzer is not None:
ret_dict["textAnalyzer"] = self.textAnalyzer._to_dict()
return ret_dict
[docs]
class _ReferencePropertyBase(_ConfigCreateModel):
name: str
[docs]
@field_validator("name")
def check_name(cls, v: str) -> str:
if v in ["id", "vector"]:
raise ValueError(f"Property name '{v}' is reserved and cannot be used")
return v
[docs]
class _ReferencePropertyMultiTarget(_ReferencePropertyBase):
"""This class defines properties that are cross references to multiple target collections.
Use this class when you want to create a cross-reference in the collection's config that is capable
of having cross-references to multiple other collections at once.
Attributes:
name: The name of the property, REQUIRED.
target_collections: The names of the target collections, REQUIRED.
description: A description of the property.
"""
target_collections: List[str]
description: Optional[str] = Field(default=None)
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
ret_dict["dataType"] = [
_capitalize_first_letter(target) for target in self.target_collections
]
del ret_dict["target_collections"]
return ret_dict
[docs]
class ReferenceProperty(_ReferencePropertyBase):
"""This class defines properties that are cross references to a single target collection.
Use this class when you want to create a cross-reference in the collection's config that is capable
of having only cross-references to a single other collection.
Attributes:
name: The name of the property, REQUIRED.
target_collection: The name of the target collection, REQUIRED.
description: A description of the property.
"""
target_collection: str
description: Optional[str] = Field(default=None)
MultiTarget: ClassVar[Type[_ReferencePropertyMultiTarget]] = _ReferencePropertyMultiTarget
[docs]
def _to_dict(self) -> Dict[str, Any]:
ret_dict = super()._to_dict()
ret_dict["dataType"] = [_capitalize_first_letter(self.target_collection)]
del ret_dict["target_collection"]
return ret_dict
PropertyType = Union[Property, ReferenceProperty, _ReferencePropertyMultiTarget]
T = TypeVar("T", bound="_CollectionConfigCreate")
[docs]
class _CollectionConfigCreate(_ConfigCreateModel):
name: str
properties: Optional[Sequence[Property]] = Field(default=None)
references: Optional[List[_ReferencePropertyBase]] = Field(default=None)
description: Optional[str] = Field(default=None)
invertedIndexConfig: Optional[_InvertedIndexConfigCreate] = Field(
default=None, alias="inverted_index_config"
)
multiTenancyConfig: Optional[_MultiTenancyConfigCreate] = Field(
default=None, alias="multi_tenancy_config"
)
objectTtlConfig: Optional[_ObjectTTLConfigCreate] = Field(
default=None, alias="object_ttl_config"
)
replicationConfig: Optional[_ReplicationConfigCreate] = Field(
default=None, alias="replication_config"
)
shardingConfig: Optional[_ShardingConfigCreate] = Field(default=None, alias="sharding_config")
vectorIndexConfig: Optional[_VectorIndexConfigCreate] = Field(
default=None, alias="vector_index_config"
)
vectorizerConfig: Union[_VectorizerConfigCreate, List[_NamedVectorConfigCreate], None] = Field(
default=None, alias="vectorizer_config"
)
vectorConfig: Union[_VectorConfigCreate, List[_VectorConfigCreate], None] = Field(
default=None,
alias="vector_config",
validate_default=True,
)
generativeSearch: Optional[_GenerativeProvider] = Field(default=None, alias="generative_config")
rerankerConfig: Optional[_RerankerProvider] = Field(default=None, alias="reranker_config")
[docs]
def model_post_init(self, __context: Any) -> None:
self.name = _capitalize_first_letter(self.name)
[docs]
@field_validator("properties", mode="after")
@classmethod
def _check_top_level_property_names(
cls, v: Optional[Sequence["Property"]]
) -> Optional[Sequence["Property"]]:
if v is None:
return v
for prop in v:
if prop.name == "id":
raise WeaviateInsertInvalidPropertyError({"name": prop.name})
return v
[docs]
@field_validator("vectorizerConfig", "vectorConfig", mode="after")
@classmethod
def validate_vector_names(
cls,
v: Union[
_VectorizerConfigCreate,
_NamedVectorConfigCreate,
List[_NamedVectorConfigCreate],
],
info: ValidationInfo,
) -> Union[
_VectorizerConfigCreate,
_NamedVectorConfigCreate,
List[_NamedVectorConfigCreate],
]:
if isinstance(v, list):
names = [vc.name for vc in v]
if len(names) != len(set(names)):
dups = {name for name in names if names.count(name) > 1}
raise ValueError(f"Vector config names must be unique. Found duplicates: {dups}")
return v
[docs]
@field_validator("vectorConfig", mode="after")
@classmethod
def inject_vector_config_none(
cls,
v: Union[_VectorConfigCreate, List[_VectorConfigCreate], None],
info: ValidationInfo,
) -> Union[_VectorConfigCreate, List[_VectorConfigCreate], None]:
if (
v is None
and info.data.get("vectorizerConfig") is None
and info.data.get("vectorIndexConfig") is None
):
return _VectorConfigCreate(
name="default",
vectorizer=_VectorizerConfigCreate(vectorizer=Vectorizers.NONE),
)
return v
@staticmethod
def __add_to_module_config(
return_dict: Dict[str, Any], addition_key: str, addition_val: Dict[str, Any]
) -> None:
if "moduleConfig" not in return_dict:
return_dict["moduleConfig"] = {addition_key: addition_val}
else:
return_dict["moduleConfig"][addition_key] = addition_val
[docs]
def _to_dict(self, *, emit_default_vector_index_type: bool = True) -> Dict[str, Any]:
ret_dict: Dict[str, Any] = {}
for cls_field in type(self).model_fields:
val = getattr(self, cls_field)
if cls_field in ["name", "model", "properties", "references"] or val is None:
continue
elif isinstance(val, (bool, float, str, int)):
ret_dict[cls_field] = str(val)
elif isinstance(val, _GenerativeProvider):
self.__add_to_module_config(ret_dict, val.generative.value, val._to_dict())
elif isinstance(val, _RerankerProvider):
self.__add_to_module_config(ret_dict, val.reranker.value, val._to_dict())
elif isinstance(val, _VectorizerConfigCreate):
ret_dict["vectorizer"] = val.vectorizer.value
if val.vectorizer != Vectorizers.NONE:
self.__add_to_module_config(ret_dict, val.vectorizer.value, val._to_dict())
elif isinstance(val, _VectorIndexConfigCreate):
ret_dict["vectorIndexType"] = val.vector_index_type().value
ret_dict[cls_field] = val._to_dict()
elif isinstance(val, _VectorConfigCreate):
ret_dict["vectorConfig"] = {
val.name or "default": val._to_dict(
emit_default_vector_index_type=emit_default_vector_index_type
)
}
elif (
isinstance(val, list)
and len(val) > 0
and all(isinstance(item, _NamedVectorConfigCreate) for item in val)
):
val = cast(List[_NamedVectorConfigCreate], val)
ret_dict["vectorConfig"] = {
item.name: item._to_dict(
emit_default_vector_index_type=emit_default_vector_index_type
)
for item in val
}
elif (
isinstance(val, list)
and len(val) > 0
and all(isinstance(item, _VectorConfigCreate) for item in val)
):
val = cast(List[_VectorConfigCreate], val)
ret_dict["vectorConfig"] = {}
for item in val:
if item.name is None:
raise WeaviateInvalidInputError(
"Vector config name must be set when specifying multiple vectors"
)
ret_dict["vectorConfig"][item.name] = item._to_dict(
emit_default_vector_index_type=emit_default_vector_index_type
)
else:
assert isinstance(val, _ConfigCreateModel)
ret_dict[cls_field] = val._to_dict()
if (
self.vectorIndexConfig is None
and "vectorConfig" not in ret_dict
and emit_default_vector_index_type
):
ret_dict["vectorIndexType"] = VectorIndexType.HNSW
ret_dict["class"] = self.name
self.__add_props(self.properties, ret_dict)
self.__add_props(self.references, ret_dict)
return ret_dict
def __add_props(
self,
props: Optional[
Union[
Sequence[Union[Property, _ReferencePropertyBase]],
List[_ReferencePropertyBase],
]
],
ret_dict: Dict[str, Any],
) -> None:
if props is None:
return
existing_props = ret_dict.get("properties", [])
existing_props.extend(
[
(
prop._to_dict(
[self.vectorizerConfig.vectorizer]
if isinstance(self.vectorizerConfig, _VectorizerConfigCreate)
else (
None
if self.vectorizerConfig is None
else [conf.vectorizer.vectorizer for conf in self.vectorizerConfig]
)
)
if isinstance(prop, Property)
else prop._to_dict()
)
for prop in props
]
)
ret_dict["properties"] = existing_props
[docs]
class _Replication:
[docs]
@staticmethod
def async_config(
*,
max_workers: Optional[int] = None,
hashtree_height: Optional[int] = None,
frequency: Optional[int] = None,
frequency_while_propagating: Optional[int] = None,
alive_nodes_checking_frequency: Optional[int] = None,
logging_frequency: Optional[int] = None,
diff_batch_size: Optional[int] = None,
diff_per_node_timeout: Optional[int] = None,
pre_propagation_timeout: Optional[int] = None,
propagation_timeout: Optional[int] = None,
propagation_limit: Optional[int] = None,
propagation_delay: Optional[int] = None,
propagation_concurrency: Optional[int] = None,
propagation_batch_size: Optional[int] = None,
) -> _AsyncReplicationConfigCreate:
"""Create a configuration object create for async replication settings when creating a collection.
This is only available with WeaviateDB `>=v1.36.0`.
Note:
`max_workers` and `alive_nodes_checking_frequency` were removed from the
Weaviate server schema in v1.37.3. Passing them has no effect against any
server `>=v1.37.3` (the server silently drops them) and emits a
``DeprecationWarning``. Both arguments will be removed in a future release.
"""
if max_workers is not None:
_Warnings.async_replication_field_removed_server_side("max_workers")
if alive_nodes_checking_frequency is not None:
_Warnings.async_replication_field_removed_server_side("alive_nodes_checking_frequency")
return _AsyncReplicationConfigCreate(
maxWorkers=max_workers,
hashtreeHeight=hashtree_height,
frequency=frequency,
frequencyWhilePropagating=frequency_while_propagating,
aliveNodesCheckingFrequency=alive_nodes_checking_frequency,
loggingFrequency=logging_frequency,
diffBatchSize=diff_batch_size,
diffPerNodeTimeout=diff_per_node_timeout,
prePropagationTimeout=pre_propagation_timeout,
propagationTimeout=propagation_timeout,
propagationLimit=propagation_limit,
propagationDelay=propagation_delay,
propagationConcurrency=propagation_concurrency,
propagationBatchSize=propagation_batch_size,
)
[docs]
class _ReplicationUpdate:
[docs]
@staticmethod
def async_config(
*,
max_workers: Optional[int] = None,
hashtree_height: Optional[int] = None,
frequency: Optional[int] = None,
frequency_while_propagating: Optional[int] = None,
alive_nodes_checking_frequency: Optional[int] = None,
logging_frequency: Optional[int] = None,
diff_batch_size: Optional[int] = None,
diff_per_node_timeout: Optional[int] = None,
pre_propagation_timeout: Optional[int] = None,
propagation_timeout: Optional[int] = None,
propagation_limit: Optional[int] = None,
propagation_delay: Optional[int] = None,
propagation_concurrency: Optional[int] = None,
propagation_batch_size: Optional[int] = None,
) -> _AsyncReplicationConfigUpdate:
"""Create a configuration object for async replication settings when updating a collection.
This is only available with WeaviateDB `>=v1.36.0`.
Note:
`max_workers` and `alive_nodes_checking_frequency` were removed from the
Weaviate server schema in v1.37.3. Passing them has no effect against any
server `>=v1.37.3` (the server silently drops them) and emits a
``DeprecationWarning``. Both arguments will be removed in a future release.
"""
if max_workers is not None:
_Warnings.async_replication_field_removed_server_side("max_workers")
if alive_nodes_checking_frequency is not None:
_Warnings.async_replication_field_removed_server_side("alive_nodes_checking_frequency")
return _AsyncReplicationConfigUpdate(
maxWorkers=max_workers,
hashtreeHeight=hashtree_height,
frequency=frequency,
frequencyWhilePropagating=frequency_while_propagating,
aliveNodesCheckingFrequency=alive_nodes_checking_frequency,
loggingFrequency=logging_frequency,
diffBatchSize=diff_batch_size,
diffPerNodeTimeout=diff_per_node_timeout,
prePropagationTimeout=pre_propagation_timeout,
propagationTimeout=propagation_timeout,
propagationLimit=propagation_limit,
propagationDelay=propagation_delay,
propagationConcurrency=propagation_concurrency,
propagationBatchSize=propagation_batch_size,
)
[docs]
class _VectorIndexQuantizerUpdate:
[docs]
@staticmethod
def pq(
bit_compression: Optional[bool] = None,
centroids: Optional[int] = None,
encoder_distribution: Optional[PQEncoderDistribution] = None,
encoder_type: Optional[PQEncoderType] = None,
segments: Optional[int] = None,
training_limit: Optional[int] = None,
enabled: bool = True,
) -> _PQConfigUpdate:
"""Create a `_PQConfigUpdate` object to be used when updating the product quantization (PQ) configuration of Weaviate.
Use this method when defining the `quantizer` argument in the `vector_index` configuration in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/concepts/vector-index#hnsw-with-compression) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
if bit_compression is not None:
_Warnings.bit_compression_in_pq_config()
return _PQConfigUpdate(
enabled=enabled,
centroids=centroids,
segments=segments,
trainingLimit=training_limit,
encoder=(
_PQEncoderConfigUpdate(type_=encoder_type, distribution=encoder_distribution)
if encoder_type is not None or encoder_distribution is not None
else None
),
)
[docs]
@staticmethod
def bq(rescore_limit: Optional[int] = None, enabled: bool = True) -> _BQConfigUpdate:
"""Create a `_BQConfigUpdate` object to be used when updating the binary quantization (BQ) configuration of Weaviate.
Use this method when defining the `quantizer` argument in the `vector_index` configuration in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/concepts/vector-index#hnsw-with-compression) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _BQConfigUpdate(rescoreLimit=rescore_limit, enabled=enabled)
[docs]
@staticmethod
def sq(
rescore_limit: Optional[int] = None,
training_limit: Optional[int] = None,
enabled: bool = True,
) -> _SQConfigUpdate:
"""Create a `_SQConfigUpdate` object to be used when updating the scalar quantization (SQ) configuration of Weaviate.
Use this method when defining the `quantizer` argument in the `vector_index` configuration in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/concepts/vector-index#hnsw-with-compression) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _SQConfigUpdate(
enabled=enabled, rescoreLimit=rescore_limit, trainingLimit=training_limit
)
[docs]
@staticmethod
def rq(
rescore_limit: Optional[int] = None,
enabled: bool = True,
bits: Optional[int] = None,
) -> _RQConfigUpdate:
"""Create a `_RQConfigUpdate` object to be used when updating the Rotational quantization (RQ) configuration of Weaviate.
Use this method when defining the `quantizer` argument in the `vector_index` configuration in `collection.update()`.
Arguments:
See [the docs](https://weaviate.io/developers/weaviate/concepts/vector-index#hnsw-with-compression) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _RQConfigUpdate(enabled=enabled, rescoreLimit=rescore_limit, bits=bits)
[docs]
class _VectorIndexUpdate:
Quantizer = _VectorIndexQuantizerUpdate
[docs]
@staticmethod
def hnsw(
dynamic_ef_factor: Optional[int] = None,
dynamic_ef_min: Optional[int] = None,
dynamic_ef_max: Optional[int] = None,
ef: Optional[int] = None,
flat_search_cutoff: Optional[int] = None,
filter_strategy: Optional[VectorFilterStrategy] = None,
vector_cache_max_objects: Optional[int] = None,
quantizer: Optional[
Union[_PQConfigUpdate, _BQConfigUpdate, _SQConfigUpdate, _RQConfigUpdate]
] = None,
) -> _VectorIndexConfigHNSWUpdate:
"""Create an `_VectorIndexConfigHNSWUpdate` object to update the configuration of the HNSW vector index.
Use this method when defining the `vectorizer_config` argument in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/configuration/indexes#configure-the-inverted-index) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _VectorIndexConfigHNSWUpdate(
dynamicEfMin=dynamic_ef_min,
dynamicEfMax=dynamic_ef_max,
dynamicEfFactor=dynamic_ef_factor,
ef=ef,
filterStrategy=filter_strategy,
flatSearchCutoff=flat_search_cutoff,
vectorCacheMaxObjects=vector_cache_max_objects,
quantizer=quantizer,
)
[docs]
@staticmethod
def flat(
vector_cache_max_objects: Optional[int] = None,
quantizer: Optional[Union[_BQConfigUpdate, _RQConfigUpdate]] = None,
) -> _VectorIndexConfigFlatUpdate:
"""Create an `_VectorIndexConfigFlatUpdate` object to update the configuration of the FLAT vector index.
Use this method when defining the `vectorizer_config` argument in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/configuration/indexes#configure-the-inverted-index) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _VectorIndexConfigFlatUpdate(
vectorCacheMaxObjects=vector_cache_max_objects,
quantizer=quantizer,
)
[docs]
@staticmethod
def dynamic(
*,
threshold: Optional[int] = None,
hnsw: Optional[_VectorIndexConfigHNSWUpdate] = None,
flat: Optional[_VectorIndexConfigFlatUpdate] = None,
quantizer: Optional[_BQConfigUpdate] = None,
) -> _VectorIndexConfigDynamicUpdate:
"""Create an `_VectorIndexConfigDynamicUpdate` object to update the configuration of the Dynamic vector index.
Use this method when defining the `vectorizer_config` argument in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/configuration/indexes#configure-the-inverted-index) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _VectorIndexConfigDynamicUpdate(
threshold=threshold,
hnsw=hnsw,
flat=flat,
quantizer=quantizer,
)
[docs]
@staticmethod
def hfresh(
max_posting_size_kb: Optional[int] = None,
search_probe: Optional[int] = None,
quantizer: Optional[_RQConfigUpdate] = None,
) -> _VectorIndexConfigHFreshUpdate:
"""Create an `_VectorIndexConfigHFreshUpdate` object to update the configuration of the HFresh vector index.
Use this method when defining the `vectorizer_config` argument in `collection.update()`.
Args:
See [the docs](https://weaviate.io/developers/weaviate/configuration/indexes#configure-the-inverted-index) for a more detailed view!
""" # noqa: D417 (missing argument descriptions in the docstring)
return _VectorIndexConfigHFreshUpdate(
maxPostingSizeKB=max_posting_size_kb,
searchProbe=search_probe,
quantizer=quantizer,
)