evofabric.core.vectorstore

evofabric.core.vectorstore#

class evofabric.core.vectorstore.DBBase(BaseComponent, ABC)#

定义基本向量数据库接口的抽象基类。

参数:

  • collection_name (str) -- 数据库集合名称

  • persist_directory (str) -- 持久化存储目录

  • embedding (EmbedClientBase) -- 用于文本向量化的嵌入客户端

  • top_k (int) -- 相似性搜索的默认结果数量

async persist()#

将向量存储持久化到磁盘。

Raises:

NotImplementedError

async clear_db() → int#

清除整个向量存储。

返回:

删除的文档数量

返回类型:

int

Raises:

NotImplementedError

在向量存储上执行相似性搜索。

参数:

  • query (str) -- 要搜索的查询文本

  • k (int, optional) -- 返回的结果数量（默认：top_k）

  • filter (Optional[Dict[str, Any]], optional) -- 搜索的元数据过滤器

返回:

匹配查询的数据库项目列表

返回类型:

List[SearchResult]

Raises:

NotImplementedError

async add_texts(items: Sequence[DBItem] | Sequence[str], metadatas: Sequence[dict] | None = None, ids: Sequence[str] | None = None) → List[str]#

向向量存储添加新的数据库项目。

参数:

  • items (Union[Sequence[DBItem], Sequence[str]]) -- 要添加的 DBItem 或文本列表

  • metadatas (Optional[Sequence[dict]], optional) -- 元数据列表（可选）

  • ids (Optional[Sequence[str]], optional) -- 文档 ID 列表（可选）

返回:

添加项目的 ID 列表

返回类型:

List[str]

Raises:

NotImplementedError

async delete_by_ids(ids: List[str]) → None#

根据其原始 ID 删除项目。

参数:

ids (List[str]) -- 要删除的 ID 列表

Raises:

NotImplementedError

class evofabric.core.vectorstore.VectorDB(DBBase, ABC)#

具有高级功能的增强型向量数据库操作的抽象类。

参数:

  • collection_name (str) -- 集合名称

  • persist_directory (str, optional) -- 持久化存储目录（可选）。如果未设置，使用内存模式。

  • embedding (EmbedClientBase, optional) -- 嵌入函数（可选）

  • top_k (int) -- 默认搜索 top_k

get_vector_count() → int#

获取数据库中的向量数量。

返回:

向量数量

返回类型:

int

Raises:

NotImplementedError

get_collection_info() → Dict[str, Any]#

获取当前集合的信息。

返回:

集合信息字典

返回类型:

Dict[str, Any]

Raises:

NotImplementedError

class evofabric.core.vectorstore.ChromaDB(VectorDB)#

基于原生 ChromaDB 的向量数据库实现。

继承自 VectorDB 并实现所有抽象方法。

参数:

  • collection_name (str) -- ChromaDB 集合名称

  • persist_directory (str, optional) -- 持久化存储目录（可选）。如果未设置，使用内存模式。

  • embedding (EmbedClientBase) -- 用于文本向量化的嵌入客户端（必需）

  • top_k (int) -- 相似性搜索的默认结果数量

嵌入客户端要求：

embedding 参数必须是 EmbedClientBase 的实例。框架提供两个主要实现：

  1. SentenceTransformerEmbed 用于本地句子转换器模型

  2. OpenAIEmbedClient 基于 OpenAI API 的嵌入

model_post_init(context: Any, /) → None#

在 pydantic 验证后初始化 ChromaDB。

参数:

context (Any) -- Pydantic 验证上下文

async persist()#

持久化数据。在 ChromaDB 中，数据会自动持久化。

返回:

None

async clear_db() → int#

通过删除所有文档同时保留集合来清除向量存储。

返回:

删除的文档数量

返回类型:

int

执行相似性搜索并返回 SearchResult 对象列表。

参数:

  • query (str) -- 查询文本

  • k (int, optional) -- 返回的结果数量（默认 top_k）

  • filter (Optional[Dict[str, Any]], optional) -- 元数据过滤器

返回:

搜索结果列表

返回类型:

List[SearchResult]

async add_texts(items: Sequence[DBItem] | Sequence[str], metadatas: Sequence[dict] | None = None, ids: Sequence[str] | None = None) → List[str]#

向向量存储添加新的数据库项目。

参数:

  • items (Union[Sequence[DBItem], Sequence[str]]) -- DBItem 或文本列表

  • metadatas (Optional[Sequence[dict]], optional) -- 元数据列表（可选）

  • ids (Optional[Sequence[str]], optional) -- 文档 ID 列表（可选）

返回:

添加后的 ID 列表

返回类型:

List[str]

async delete_by_ids(ids: List[str]) → bool#

根据其 ID 删除向量。

参数:

ids (List[str]) -- 要删除的 ID 列表

返回:

删除是否成功

返回类型:

bool

get_vector_count() → int#

获取集合中存储的向量数量。

返回:

向量数量

返回类型:

int

get_collection_info() → Dict[str, Any]#

获取当前集合的信息。

返回:

集合信息字典

返回类型:

Dict[str, Any]

目录