CAMEL - Build Multi-Agent AI Systems

WeaviateStorage

class WeaviateStorage(BaseVectorStorage):

An implementation of the BaseVectorStorage for interacting with Weaviate, a cloud-native vector search engine. This class provides multiple ways to connect to Weaviate instances:

Weaviate Cloud (WCD)
Local Docker/Kubernetes instances
Embedded Weaviate
Custom connection parameters

Parameters:

vector_dim (int): The dimension of storing vectors.
collection_name (Optional[str], optional): Name for the collection in Weaviate. If not provided, generates a unique name based on current timestamp. (default: :obj:None)
connection_type (ConnectionType, optional): Type of connection to use. Supported types: ‘local’, ‘cloud’, ‘embedded’, ‘custom’. (default: :obj:"local") # Weaviate Cloud parameters
wcd_cluster_url (Optional[str], optional): Weaviate Cloud cluster URL. Required when connection_type=‘cloud’.
wcd_api_key (Optional[str], optional): Weaviate Cloud API key. Required when connection_type=‘cloud’. # Local instance parameters
local_host (str, optional): Local Weaviate host. (default: :obj:"localhost")
local_port (int, optional): Local Weaviate HTTP port. (default: :obj:8080)
local_grpc_port (int, optional): Local Weaviate gRPC port. (default: :obj:50051)
local_auth_credentials (Optional[Union[str, Any]], optional): Authentication credentials for local instance. Can be an API key string or Auth object. (default: :obj:None) # Embedded Weaviate parameters
embedded_hostname (str, optional): Embedded instance hostname. (default: :obj:"127.0.0.1")
embedded_port (int, optional): Embedded instance HTTP port. (default: :obj:8079)
embedded_grpc_port (int, optional): Embedded instance gRPC port. (default: :obj:50050)
embedded_version (Optional[str], optional): Weaviate version for embedded instance. If None, uses the default version. (default: :obj:None)
embedded_persistence_data_path (Optional[str], optional): Directory for embedded database files. (default: :obj:None)
embedded_binary_path (Optional[str], optional): Directory for Weaviate binary. (default: :obj:None)
embedded_environment_variables (Optional[Dict[str, str]], optional): Environment variables for embedded instance. (default: :obj:None) # Custom connection parameters
custom_http_host (Optional[str], optional): Custom HTTP host.
custom_http_port (Optional[int], optional): Custom HTTP port.
custom_http_secure (Optional[bool], optional): Use HTTPS.
custom_grpc_host (Optional[str], optional): Custom gRPC host.
custom_grpc_port (Optional[int], optional): Custom gRPC port.
custom_grpc_secure (Optional[bool], optional): Use secure gRPC.
custom_auth_credentials (Optional[Any], optional): Custom auth. # Vector index configuration parameters
vector_index_type (VectorIndexType, optional): Vector index type. Supported types: ‘hnsw’, ‘flat’. (default: :obj:"hnsw")
distance_metric (DistanceMetric, optional): Distance metric for vector similarity. Supported metrics: ‘cosine’, ‘dot’, ‘l2-squared’, ‘hamming’, ‘manhattan’. (default: :obj:"cosine") # Common parameters for all connection types
headers (Optional[Dict[str, str]], optional): Additional headers for third-party API keys (e.g., OpenAI, Cohere). (default: :obj:None)
additional_config (Optional[Any], optional): Advanced configuration options like timeouts. (default: :obj:None)
skip_init_checks (bool, optional): Skip initialization checks. (default: :obj:False)

Note: This implementation supports synchronous operations only. The client connection is automatically handled and closed when the storage instance is destroyed.

init

def __init__(
    self,
    vector_dim: int,
    collection_name: Optional[str] = None,
    connection_type: ConnectionType = 'local',
    wcd_cluster_url: Optional[str] = None,
    wcd_api_key: Optional[str] = None,
    local_host: str = 'localhost',
    local_port: int = 8080,
    local_grpc_port: int = 50051,
    local_auth_credentials: Optional[Union[str, Any]] = None,
    embedded_hostname: str = '127.0.0.1',
    embedded_port: int = 8079,
    embedded_grpc_port: int = 50050,
    embedded_version: Optional[str] = None,
    embedded_persistence_data_path: Optional[str] = None,
    embedded_binary_path: Optional[str] = None,
    embedded_environment_variables: Optional[Dict[str, str]] = None,
    custom_http_host: Optional[str] = None,
    custom_http_port: Optional[int] = None,
    custom_http_secure: Optional[bool] = None,
    custom_grpc_host: Optional[str] = None,
    custom_grpc_port: Optional[int] = None,
    custom_grpc_secure: Optional[bool] = None,
    custom_auth_credentials: Optional[Any] = None,
    vector_index_type: VectorIndexType = 'hnsw',
    distance_metric: DistanceMetric = 'cosine',
    headers: Optional[Dict[str, str]] = None,
    additional_config: Optional[Any] = None,
    skip_init_checks: bool = False,
    **kwargs: Any
):

_get_connection_client

def _get_connection_client(self):

Get Weaviate client based on connection type and user settings.

_create_cloud_client

def _create_cloud_client(self, weaviate_module: Any):

Create a Weaviate Cloud client.

_create_local_client

def _create_local_client(self, weaviate_module: Any):

Create a local Weaviate client.

_create_embedded_client

def _create_embedded_client(self, weaviate_module: Any):

Create an embedded Weaviate client.

_create_custom_client

def _create_custom_client(self, weaviate_module: Any):

Create a custom Weaviate client.

del

def __del__(self):

Clean up client connection.

close

def close(self):

Explicitly close the client connection.

_generate_collection_name

def _generate_collection_name(self):

Generate a collection name if user doesn’t provide one.

_check_and_create_collection

def _check_and_create_collection(self, **kwargs: Any):

Check if collection exists and create if it doesn’t.

_collection_exists

def _collection_exists(self, collection_name: str):

Check if the collection exists.

_get_vector_index_config

def _get_vector_index_config(self, **kwargs: Any):

Get vector index configuration based on user settings.

_create_collection

def _create_collection(self, **kwargs: Any):

Create a new collection in Weaviate.

add

def add(self, records: List[VectorRecord], **kwargs: Any):

Saves a list of vector records to the storage. Parameters:

records (List[VectorRecord]): List of vector records to be saved. **kwargs (Any): Additional keyword arguments.

delete

def delete(self, ids: List[str], **kwargs: Any):

Deletes a list of vectors identified by their IDs from the storage. Parameters:

ids (List[str]): List of unique identifiers for the vectors to be deleted. **kwargs (Any): Additional keyword arguments.

_calculate_similarity_from_distance

def _calculate_similarity_from_distance(self, distance: Optional[float]):

Calculate similarity score based on distance metric. Parameters:

distance (Optional[float]): The distance value from Weaviate.

Returns: float: Normalized similarity score between 0 and 1.

status

def status(self):

Returns: VectorDBStatus: The vector database status.

query

def query(self, query: VectorDBQuery, **kwargs: Any):

Searches for similar vectors in the storage based on the provided query. Parameters:

query (VectorDBQuery): The query object containing the search vector and the number of top similar vectors to retrieve. **kwargs (Any): Additional keyword arguments.

Returns: List[VectorDBQueryResult]: A list of vectors retrieved from the storage based on similarity to the query vector.

clear

def clear(self):

Remove all vectors from the storage.

load

def load(self):

Load the collection hosted on cloud service.

client

def client(self):

Provides access to the underlying vector database client.

​WeaviateStorage

​init

​_get_connection_client

​_create_cloud_client

​_create_local_client

​_create_embedded_client

​_create_custom_client

​del

​close

​_generate_collection_name

​_check_and_create_collection

​_collection_exists

​_get_vector_index_config

​_create_collection

​add

​delete

​_calculate_similarity_from_distance

​status

​query

​clear

​load

​client