BaseVectorDB Interface Class

This interface defines the contract that all vector database implementations must follow.

memora.vector_db.base.BaseVectorDB

Bases: ABC

Abstract base class defining a common interface for different Vector DB implementations.

This class provides a standardized interface for vector database operations including adding, searching, and deleting memories.

Functions

add_memories abstractmethod async

add_memories(
    org_id: str,
    user_id: str,
    agent_id: str,
    memory_ids: List[uuid.UUID],
    memories: List[str],
    obtained_at: str,
) -> None

Add memories to collection with their org_id, user_id, agent_id, and obtained_at datetime as metadata.

PARAMETER	DESCRIPTION
org_id	Organization ID for the memories TYPE: str
user_id	User ID for the memories TYPE: str
agent_id	Agent ID for the memories TYPE: str
memory_ids	List of UUIDs for each memory TYPE: List[UUID]
memories	List of memory strings to add TYPE: List[str]
obtained_at	ISO format datetime string when the memories were obtained TYPE: str

Source code in memora/vector_db/base.py

@abstractmethod
async def add_memories(
    self,
    org_id: str,
    user_id: str,
    agent_id: str,
    memory_ids: List[uuid.UUID],
    memories: List[str],
    obtained_at: str,
) -> None:
    """
    Add memories to collection with their org_id, user_id, agent_id, and obtained_at datetime as metadata.

    Args:
        org_id (str): Organization ID for the memories
        user_id (str): User ID for the memories
        agent_id (str): Agent ID for the memories
        memory_ids (List[uuid.UUID]): List of UUIDs for each memory
        memories (List[str]): List of memory strings to add
        obtained_at (str): ISO format datetime string when the memories were obtained
    """
    pass

close abstractmethod async

close() -> None

Closes the database connection.

Source code in memora/vector_db/base.py

@abstractmethod
async def close(self) -> None:
    """Closes the database connection."""
    pass

delete_all_organization_memories abstractmethod async

delete_all_organization_memories(org_id: str) -> None

Delete all memories associated with an organization.

PARAMETER	DESCRIPTION
org_id	ID of the organization whose memories should be deleted TYPE: str

Source code in memora/vector_db/base.py

@abstractmethod
async def delete_all_organization_memories(self, org_id: str) -> None:
    """
    Delete all memories associated with an organization.

    Args:
        org_id (str): ID of the organization whose memories should be deleted
    """
    pass

delete_all_user_memories abstractmethod async

delete_all_user_memories(org_id: str, user_id: str) -> None

Delete all memories associated with a specific user.

PARAMETER	DESCRIPTION
org_id	Organization ID the user belongs to TYPE: str
user_id	ID of the user whose memories should be deleted TYPE: str

Source code in memora/vector_db/base.py

@abstractmethod
async def delete_all_user_memories(self, org_id: str, user_id: str) -> None:
    """
    Delete all memories associated with a specific user.

    Args:
        org_id (str): Organization ID the user belongs to
        user_id (str): ID of the user whose memories should be deleted
    """
    pass

delete_memories abstractmethod async

delete_memories(memory_ids: List[str]) -> None

Delete multiple memories by their IDs.

PARAMETER	DESCRIPTION
memory_ids	List of memory IDs to delete TYPE: List[str]

Source code in memora/vector_db/base.py

@abstractmethod
async def delete_memories(self, memory_ids: List[str]) -> None:
    """
    Delete multiple memories by their IDs.

    Args:
        memory_ids (List[str]): List of memory IDs to delete
    """
    pass

delete_memory abstractmethod async

delete_memory(memory_id: str) -> None

Delete a memory by its ID with optional org/user filtering.

PARAMETER	DESCRIPTION
memory_id	ID of the memory to delete TYPE: str

Source code in memora/vector_db/base.py

@abstractmethod
async def delete_memory(self, memory_id: str) -> None:
    """
    Delete a memory by its ID with optional org/user filtering.

    Args:
        memory_id (str): ID of the memory to delete
    """
    pass

search_memories abstractmethod async

search_memories(
    queries: List[str],
    memory_search_scope: MemorySearchScope,
    org_id: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> List[List[Tuple[models.Memory, float]]]

Batch memory search with optional user/agent filtering.

PARAMETER	DESCRIPTION
queries	List of search query strings TYPE: List[str]
memory_search_scope	Memory search scope (organization or user) TYPE: MemorySearchScope
org_id	Organization ID for filtering TYPE: str
user_id	Optional user ID for filtering TYPE: Optional[str] DEFAULT: None
agent_id	Optional agent ID for filtering TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[List[Tuple[Memory, float]]]	List[List[Tuple[models.Memory, float]]] of search results for each query, with a tuple containing: Memory: + org_id: str + agent_id: str + user_id: str + memory_id: str + memory: str + obtained_at: datetime float: Score of the memory

Source code in memora/vector_db/base.py

@abstractmethod
async def search_memories(
    self,
    queries: List[str],
    memory_search_scope: MemorySearchScope,
    org_id: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> List[List[Tuple[models.Memory, float]]]:
    """
    Batch memory search with optional user/agent filtering.

    Args:
        queries (List[str]): List of search query strings
        memory_search_scope (MemorySearchScope): Memory search scope (organization or user)
        org_id (str): Organization ID for filtering
        user_id (Optional[str]): Optional user ID for filtering
        agent_id (Optional[str]): Optional agent ID for filtering

    Returns:
        List[List[Tuple[models.Memory, float]]] of search results for each query, with a tuple containing:
            Memory:

                + org_id: str
                + agent_id: str
                + user_id: str
                + memory_id: str
                + memory: str
                + obtained_at: datetime

            float: Score of the memory
    """
    pass

search_memory abstractmethod async

search_memory(
    query: str,
    memory_search_scope: MemorySearchScope,
    org_id: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> List[Tuple[models.Memory, float]]

Memory search with optional user/agent filtering.

PARAMETER	DESCRIPTION
query	Search query string TYPE: str
memory_search_scope	Memory search scope (organization or user) TYPE: MemorySearchScope
org_id	Organization ID for filtering TYPE: str
user_id	Optional user ID for filtering TYPE: Optional[str] DEFAULT: None
agent_id	Optional agent ID for filtering TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
List[Tuple[Memory, float]]	List[Tuple[Memory, float]] containing tuple of search results and score: Memory: + org_id: str + agent_id: str + user_id: str + memory_id: str + memory: str + obtained_at: datetime float: Score of the memory

Source code in memora/vector_db/base.py

@abstractmethod
async def search_memory(
    self,
    query: str,
    memory_search_scope: MemorySearchScope,
    org_id: str,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> List[Tuple[models.Memory, float]]:
    """
    Memory search with optional user/agent filtering.

    Args:
        query (str): Search query string
        memory_search_scope (MemorySearchScope): Memory search scope (organization or user)
        org_id (str): Organization ID for filtering
        user_id (Optional[str]): Optional user ID for filtering
        agent_id (Optional[str]): Optional agent ID for filtering

    Returns:
        List[Tuple[Memory, float]] containing tuple of search results and score:
            Memory:

                + org_id: str
                + agent_id: str
                + user_id: str
                + memory_id: str
                + memory: str
                + obtained_at: datetime

            float: Score of the memory
    """
    pass

setup abstractmethod async

setup(*args, **kwargs) -> None

Setup the vector database by initializing collections, indices, etc.

Source code in memora/vector_db/base.py

@abstractmethod
async def setup(self, *args, **kwargs) -> None:
    """Setup the vector database by initializing collections, indices, etc."""
    pass