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] | 一个文档列表。每个文档都是一个TypedDictDocument
。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 | 向量数据库。