8a8d2ed574
原有的清理逻辑直接使用 delete 和 where 条件（timestamp: {"$lt": ...}）来删除过期记录。然而，ChromaDB 对元数据中复杂的查询操作符（如 $lt）的支持并不可靠。
为确保过期记录能被稳定地清除，本次提交将清理策略修改为更稳健的“获取-过滤-删除”模式：
1. 为向量数据库抽象层新增 `get` 方法，并为 ChromaDB 提供具体实现。
2. 在 `VectorInstantMemoryV2` 中，先获取指定聊天的所有记录。
3. 在应用代码中根据时间戳筛选出过期的记录ID。
4. 最后根据ID列表精确删除过期记录，确保了清理操作的准确性。
145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
class VectorDBBase(ABC):
|
|
"""
|
|
向量数据库的抽象基类 (ABC)，定义了所有向量数据库实现必须遵循的接口。
|
|
"""
|
|
|
|
@abstractmethod
|
|
def __init__(self, path: str, **kwargs: Any):
|
|
"""
|
|
初始化向量数据库客户端。
|
|
|
|
Args:
|
|
path (str): 数据库文件的存储路径。
|
|
**kwargs: 其他特定于实现的参数。
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_or_create_collection(self, name: str, **kwargs: Any) -> Any:
|
|
"""
|
|
获取或创建一个集合 (Collection)。
|
|
|
|
Args:
|
|
name (str): 集合的名称。
|
|
**kwargs: 其他特定于实现的参数 (例如 metadata)。
|
|
|
|
Returns:
|
|
Any: 代表集合的对象。
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def add(
|
|
self,
|
|
collection_name: str,
|
|
embeddings: List[List[float]],
|
|
documents: Optional[List[str]] = None,
|
|
metadatas: Optional[List[Dict[str, Any]]] = None,
|
|
ids: Optional[List[str]] = None,
|
|
) -> None:
|
|
"""
|
|
向指定集合中添加数据。
|
|
|
|
Args:
|
|
collection_name (str): 目标集合的名称。
|
|
embeddings (List[List[float]]): 向量列表。
|
|
documents (Optional[List[str]], optional): 文档列表。Defaults to None.
|
|
metadatas (Optional[List[Dict[str, Any]]], optional): 元数据列表。Defaults to None.
|
|
ids (Optional[List[str]], optional): ID 列表。Defaults to None.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def query(
|
|
self,
|
|
collection_name: str,
|
|
query_embeddings: List[List[float]],
|
|
n_results: int = 1,
|
|
where: Optional[Dict[str, Any]] = None,
|
|
**kwargs: Any,
|
|
) -> Dict[str, List[Any]]:
|
|
"""
|
|
在指定集合中查询相似向量。
|
|
|
|
Args:
|
|
collection_name (str): 目标集合的名称。
|
|
query_embeddings (List[List[float]]): 用于查询的向量列表。
|
|
n_results (int, optional): 返回结果的数量。Defaults to 1.
|
|
where (Optional[Dict[str, Any]], optional): 元数据过滤条件。Defaults to None.
|
|
**kwargs: 其他特定于实现的参数。
|
|
|
|
Returns:
|
|
Dict[str, List[Any]]: 查询结果，通常包含 ids, distances, metadatas, documents。
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete(
|
|
self,
|
|
collection_name: str,
|
|
ids: Optional[List[str]] = None,
|
|
where: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
"""
|
|
从指定集合中删除数据。
|
|
|
|
Args:
|
|
collection_name (str): 目标集合的名称。
|
|
ids (Optional[List[str]], optional): 要删除的条目的 ID 列表。Defaults to None.
|
|
where (Optional[Dict[str, Any]], optional): 基于元数据的过滤条件。Defaults to None.
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get(
|
|
self,
|
|
collection_name: str,
|
|
ids: Optional[List[str]] = None,
|
|
where: Optional[Dict[str, Any]] = None,
|
|
limit: Optional[int] = None,
|
|
offset: Optional[int] = None,
|
|
where_document: Optional[Dict[str, Any]] = None,
|
|
include: Optional[List[str]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
根据条件从集合中获取数据。
|
|
|
|
Args:
|
|
collection_name (str): 目标集合的名称。
|
|
ids (Optional[List[str]], optional): 要获取的条目的 ID 列表。Defaults to None.
|
|
where (Optional[Dict[str, Any]], optional): 基于元数据的过滤条件。Defaults to None.
|
|
limit (Optional[int], optional): 返回结果的数量限制。Defaults to None.
|
|
offset (Optional[int], optional): 返回结果的偏移量。Defaults to None.
|
|
where_document (Optional[Dict[str, Any]], optional): 基于文档内容的过滤条件。Defaults to None.
|
|
include (Optional[List[str]], optional): 指定返回的数据字段 (e.g., ["metadatas", "documents"])。Defaults to None.
|
|
|
|
Returns:
|
|
Dict[str, Any]: 获取到的数据。
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def count(self, collection_name: str) -> int:
|
|
"""
|
|
获取指定集合中的条目总数。
|
|
|
|
Args:
|
|
collection_name (str): 目标集合的名称。
|
|
|
|
Returns:
|
|
int: 条目总数。
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def delete_collection(self, name: str) -> None:
|
|
"""
|
|
删除一个集合。
|
|
|
|
Args:
|
|
name (str): 要删除的集合的名称。
|
|
"""
|
|
pass |