跳到主要内容

agentchat.contrib.vectordb.couchbase

CouchbaseVectorDB​

class CouchbaseVectorDB(VectorDB)

一个使用Couchbase作为后端的向量数据库实现。

__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)

初始化向量数据库。

参数:

  • connection_string str - 用于连接的Couchbase连接字符串。默认为'couchbase://localhost'。
  • username str - Couchbase 认证的用户名。默认为 'Administrator'。
  • password str - Couchbase认证的密码。默认值为'password'。
  • bucket_name str - 桶的名称。默认为'vector_db'。
  • embedding_function 可调用对象 - 用于生成向量表示的嵌入函数。默认是 SentenceTransformer("all-MiniLM-L6-v2").encode。
  • scope_name str - 范围名称。默认为'_default'。
  • collection_name str - 要为此向量数据库创建的集合名称。默认值为'_default'。
  • index_name str - 向量数据库的索引名称。默认为 None。
  • overwrite bool - 是否覆盖现有数据。默认为False。
  • wait_until_index_ready float | None - 阻塞调用，等待数据库索引准备就绪。None 表示不等待。默认为 None。
  • wait_until_document_ready float | None - 阻塞调用，等待直到数据库文档准备就绪。None表示不等待。默认为None。

search_index_exists​

def search_index_exists(index_name: str)

检查指定索引是否已准备好

创建集合​

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

在向量数据库中创建一个集合，并在集合中创建向量搜索索引。

参数:

  • collection_name - str | 集合的名称。
  • overwrite - bool | 如果集合存在，是否覆盖它。默认值为 False。
  • get_or_create - bool | 获取或创建集合。默认为True

create_index_if_not_exists​

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

在Couchbase的指定集合上创建一个向量搜索索引。

参数:

  • index_name str, 可选 - 要创建的向量搜索索引的名称。默认为"vector_search_index"。
  • collection 集合，可选 - 要创建索引的Couchbase集合。默认为None。

get_collection​

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

从向量数据库中获取集合。

参数:

  • collection_name - str | 集合的名称。默认为 None。如果为 None，则返回当前活动的集合。

返回:

集合 | 集合对象。

delete_collection​

def delete_collection(collection_name: str) -> None

从向量数据库中删除集合。

参数:

  • collection_name - str | 集合的名称。

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

在集合中创建一个向量搜索索引。

insert_docs​

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

将文档和向量嵌入插入向量数据库的集合中。在所有情况下，文档都会被更新插入。

更新文档​

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

更新集合中的文档，包括它们的嵌入。

删除文档​

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

从向量数据库的集合中删除文档。

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]

根据ID从向量数据库的集合中检索文档。

retrieve_docs​

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

根据查询从向量数据库的集合中检索文档。 注意：Couchbase FTS不支持距离阈值。