Skip to main content

agentchat.contrib.vectordb.base

ItemID​

chromadb不支持整数id，VikingDB支持

Document​

class Document(TypedDict)

Document是向量数据库中的记录。

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

VectorDB​

@runtime_checkable
class VectorDB(Protocol)

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

属性：

  • active_collection - Any | 向量数据库中的活动集合。加快获取集合的速度。默认为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从向量数据库的集合中检索文档。

create_collection​

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。

返回值：

Any | 集合对象。

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 | 集合的名称。

返回值：

任意类型

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 | 附加的关键字参数。

返回值：

None

update_docs​

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

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

参数：

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

返回值：

None

delete_docs​

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

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

参数：

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

返回值：

None

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。