跳到主要内容

agentchat.contrib.vectordb.base

项目ID

chromadb不支持int类型的id,VikingDB支持

文档

class Document(TypedDict)

文档是向量数据库中的一条记录。

id: ItemID | 文档的唯一标识符。 content: str | 文本块的内容。 metadata: Metadata, Optional | 包含文档的额外信息,如来源、日期等。 embedding: Vector, Optional | 内容的向量表示。

向量数据库

@runtime_checkable
class VectorDB(Protocol)

向量数据库的抽象类。向量数据库负责存储和检索文档。

属性:

  • active_collection - Any | 向量数据库中的活动集合。使 get_collection 更快。默认为 None。
  • type - str | 向量数据库的类型,如chroma、pgvector等。默认为""。

方法:

  • create_collection - Callable[[str, bool, bool], Any] | 在向量数据库中创建一个集合。
  • get_collection - Callable[[str], Any] | 从向量数据库中获取集合。
  • delete_collection - Callable[[str], Any] | 从向量数据库中删除集合。
  • insert_docs - Callable[[List[Document], str, bool], None] | 将文档插入到向量数据库的集合中。
  • update_docs - Callable[[List[Document], str], None] | 更新向量数据库集合中的文档。
  • delete_docs - Callable[[List[ItemID], str], None] | 从向量数据库的集合中删除文档。
  • retrieve_docs - Callable[[List[str], str, int, float], QueryResults] | 根据查询从向量数据库的集合中检索文档。
  • get_docs_by_ids - Callable[[List[ItemID], str], List[Document]] | 根据ID从向量数据库中检索文档。

创建集合

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

在向量数据库中创建一个集合。 情况1. 如果集合不存在,则创建该集合。 情况2. 如果集合存在且overwrite为True,则会覆盖该集合。 情况3. 如果集合存在且overwrite为False,若get_or_create为True,则会获取该集合, 否则会引发ValueError。

参数:

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

返回:

任何 | 集合对象。

get_collection

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

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

参数:

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

返回:

任何 | 集合对象。

delete_collection

def delete_collection(collection_name: str) -> Any

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

参数:

  • collection_name - str | 集合的名称。

返回:

任何

generate_chunk_ids

def generate_chunk_ids(chunks: List[str],
hash_length: int = HASH_LENGTH) -> List[ItemID]

生成块ID以确保上传不重复。

参数:

  • chunks 列表 - 要哈希的块(字符串)列表。
  • hash_length int - 哈希的期望长度。

返回:

  • list - 生成的分块ID列表。

insert_docs

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

将文档插入到向量数据库的集合中。

参数:

  • docs - List[Document] | 一个文档列表。每个文档都是一个TypedDict Document
  • collection_name - str | 集合的名称。默认值为 None。
  • upsert - bool | 如果文档存在,是否更新。默认为False。
  • kwargs - Dict | 额外的关键字参数。

返回:

更新文档

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

更新向量数据库集合中的文档。

参数:

  • docs - List[Document] | 文档列表。
  • collection_name - str | 集合的名称。默认值为 None。
  • kwargs - Dict | 额外的关键字参数。

返回:

删除文档

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

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

参数:

  • ids - List[ItemID] | 文档ID列表。每个ID都是类型化的ItemID
  • collection_name - str | 集合的名称。默认值为 None。
  • kwargs - Dict | 额外的关键字参数。

返回:

retrieve_docs

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

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

参数:

  • queries - List[str] | 查询列表。每个查询是一个字符串。
  • collection_name - str | 集合的名称。默认值为 None。
  • n_results - int | 返回的相关文档数量。默认值为10。
  • distance_threshold - float | 距离分数的阈值,只有小于该值的距离才会返回。如果小于0则不使用此过滤,默认值为-1。
  • kwargs - Dict | 额外的关键字参数。

返回:

QueryResults | 查询结果。每个查询结果是一个元组列表的列表,包含文档及其距离。

get_docs_by_ids

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

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

参数:

  • ids - List[ItemID] | 文档ID列表。如果为None,将返回所有文档。默认为None。
  • collection_name - str | 集合的名称。默认值为 None。
  • include - List[str] | 包含的字段。默认是 None。 如果为 None,将包含 ["metadatas", "documents"],ids 将始终被包含。这可能会根据实现有所不同。
  • kwargs - dict | 额外的关键字参数。

返回:

List[Document] | 结果。

VectorDBFactory

class VectorDBFactory()

用于创建向量数据库的工厂类。

创建向量数据库

@staticmethod
def create_vector_db(db_type: str, **kwargs) -> VectorDB

创建一个向量数据库。

参数:

  • db_type - str | 向量数据库的类型。
  • kwargs - Dict | 初始化向量数据库的关键字参数。

返回:

VectorDB | 向量数据库。