weaviatedb

langroid/vector_store/weaviatedb.py

WeaviateDB(config=WeaviateDBConfig())

Bases: VectorStore

Source code in langroid/vector_store/weaviatedb.py

def __init__(self, config: WeaviateDBConfig = WeaviateDBConfig()):
    super().__init__(config)
    self.config: WeaviateDBConfig = config
    self.embedding_fn: EmbeddingFunction = self.embedding_model.embedding_fn()
    self.embedding_dim = self.embedding_model.embedding_dims
    load_dotenv()
    key = os.getenv("WEAVIATE_API_KEY")
    url = os.getenv("WEAVIATE_API_URL")
    if None in [key, url]:
        logger.warning(
            """WEAVIATE_API_KEY, WEAVIATE_API_URL env variable must be set to use
            WeaviateDB in cloud mode. Please set these values
            in your .env file.
            """
        )
    self.client = weaviate.connect_to_weaviate_cloud(
        cluster_url=url,
        auth_credentials=Auth.api_key(key),
    )
    if config.collection_name is not None:
        WeaviateDB.validate_and_format_collection_name(config.collection_name)

validate_and_format_collection_name(name) staticmethod

Formats the collection name to comply with Weaviate's naming rules: - Name must start with a capital letter. - Name can only contain letters, numbers, and underscores. - Replaces invalid characters with underscores.

Source code in langroid/vector_store/weaviatedb.py

@staticmethod
def validate_and_format_collection_name(name: str) -> str:
    """
    Formats the collection name to comply with Weaviate's naming rules:
    - Name must start with a capital letter.
    - Name can only contain letters, numbers, and underscores.
    - Replaces invalid characters with underscores.
    """
    if not name:
        raise ValueError("Collection name cannot be empty.")

    formatted_name = re.sub(r"[^a-zA-Z0-9_]", "_", name)

    # Ensure the first letter is capitalized
    if not formatted_name[0].isupper():
        formatted_name = formatted_name.capitalize()

    # Check if the name now meets the criteria
    if not re.match(r"^[A-Z][A-Za-z0-9_]*$", formatted_name):
        raise ValueError(
            f"Invalid collection name '{name}'."
            " Names must start with a capital letter "
            "and contain only letters, numbers, and underscores."
        )

    if formatted_name != name:
        logger.warning(
            f"Collection name '{name}' was reformatted to '{formatted_name}' "
            "to comply with Weaviate's rules."
        )

    return formatted_name