Spaces:
Paused
Paused
File size: 1,800 Bytes
9792ea7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | # -*- coding: utf-8 -*-
"""The embedding cache base class."""
from abc import abstractmethod
from typing import List, Any
from ..types import (
JSONSerializableObject,
Embedding,
)
class EmbeddingCacheBase:
"""Base class for embedding caches, which is responsible for storing and
retrieving embeddings."""
@abstractmethod
async def store(
self,
embeddings: List[Embedding],
identifier: JSONSerializableObject,
overwrite: bool = False,
**kwargs: Any,
) -> None:
"""Store the embeddings with the given identifier.
Args:
embeddings (`List[Embedding]`):
The embeddings to store.
identifier (`JSONSerializableObject`):
The identifier to distinguish the embeddings.
overwrite (`bool`, defaults to `False`):
Whether to overwrite existing embeddings with the same
identifier. If `True`, existing embeddings will be replaced.
"""
@abstractmethod
async def retrieve(
self,
identifier: JSONSerializableObject,
) -> List[Embedding] | None:
"""Retrieve the embeddings with the given identifier. If not
found, return `None`.
Args:
identifier (`JSONSerializableObject`):
The identifier to retrieve the embeddings.
"""
@abstractmethod
async def remove(
self,
identifier: JSONSerializableObject,
) -> None:
"""Remove the embeddings with the given identifier.
Args:
identifier (`JSONSerializableObject`):
The identifier to remove the embeddings.
"""
@abstractmethod
async def clear(self) -> None:
"""Clear all cached embeddings."""
|