Skip to main content

agentchat.contrib.vectordb.couchbase

CouchbaseVectorDB​

class CouchbaseVectorDB(VectorDB)

A vector database implementation that uses Couchbase as the backend.

__init__​

def __init__(connection_string: str = "couchbase://localhost",
             username: str = "Administrator",
             password: str = "password",
             bucket_name: str = "vector_db",
             embedding_function: Callable = SentenceTransformer(
                 "all-MiniLM-L6-v2").encode,
             scope_name: str = "_default",
             collection_name: str = "_default",
             index_name: str = None)

Initialize the vector database.

Arguments:

  • connection_string str - The Couchbase connection string to connect to. Default is 'couchbase://localhost'.
  • username str - The username for Couchbase authentication. Default is 'Administrator'.
  • password str - The password for Couchbase authentication. Default is 'password'.
  • bucket_name str - The name of the bucket. Default is 'vector_db'.
  • embedding_function Callable - The embedding function used to generate the vector representation. Default is SentenceTransformer("all-MiniLM-L6-v2").encode.
  • scope_name str - The name of the scope. Default is '_default'.
  • collection_name str - The name of the collection to create for this vector database. Default is '_default'.
  • index_name str - Index name for the vector database. Default is None.
  • overwrite bool - Whether to overwrite existing data. Default is False.
  • wait_until_index_ready float | None - Blocking call to wait until the database indexes are ready. None means no wait. Default is None.
  • wait_until_document_ready float | None - Blocking call to wait until the database documents are ready. None means no wait. Default is None.

search_index_exists​

def search_index_exists(index_name: str)

Check if the specified index is ready

create_collection​

def create_collection(collection_name: str,
                      overwrite: bool = False,
                      get_or_create: bool = True) -> Collection

Create a collection in the vector database and create a vector search index in the collection.

Arguments:

  • collection_name - str | The name of the collection.
  • overwrite - bool | Whether to overwrite the collection if it exists. Default is False.
  • get_or_create - bool | Whether to get or create the collection. Default is True

create_index_if_not_exists​

def create_index_if_not_exists(index_name: str = "vector_index",
                               collection=None) -> None

Creates a vector search index on the specified collection in Couchbase.

Arguments:

  • index_name str, optional - The name of the vector search index to create. Defaults to "vector_search_index".
  • collection Collection, optional - The Couchbase collection to create the index on. Defaults to None.

get_collection​

def get_collection(collection_name: str = None) -> Collection

Get the collection from the vector database.

Arguments:

  • collection_name - str | The name of the collection. Default is None. If None, return the current active collection.

Returns:

Collection | The collection object.

delete_collection​

def delete_collection(collection_name: str) -> None

Delete the collection from the vector database.

Arguments:

  • collection_name - str | The name of the collection.

create_vector_search_index​

def create_vector_search_index(
        collection,
        index_name: Union[str, None] = "vector_index",
        similarity: Literal["l2_norm", "dot_product"] = "dot_product") -> None

Create a vector search index in the collection.

insert_docs​

def insert_docs(docs: List[Document],
                collection_name: str = None,
                upsert: bool = False,
                batch_size=DEFAULT_BATCH_SIZE,
                **kwargs) -> None

Insert Documents and Vector Embeddings into the collection of the vector database. Documents are upserted in all cases.

update_docs​

def update_docs(docs: List[Document],
                collection_name: str = None,
                batch_size=DEFAULT_BATCH_SIZE,
                **kwargs: Any) -> None

Update documents, including their embeddings, in the Collection.

delete_docs​

def delete_docs(ids: List[ItemID],
                collection_name: str = None,
                batch_size=DEFAULT_BATCH_SIZE,
                **kwargs)

Delete documents from the collection of the vector database.

get_docs_by_ids​

def get_docs_by_ids(ids: List[ItemID] | None = None,
                    collection_name: str = None,
                    include: List[str] | None = None,
                    **kwargs) -> List[Document]

Retrieve documents from the collection of the vector database based on the ids.

retrieve_docs​

def retrieve_docs(queries: List[str],
                  collection_name: str = None,
                  n_results: int = 10,
                  distance_threshold: float = -1,
                  **kwargs) -> QueryResults

Retrieve documents from the collection of the vector database based on the queries. Note: Distance threshold is not supported in Couchbase FTS.