BaseGraphDB Interface Class

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

memora.graph_db.base.BaseGraphDB

Bases: ABC

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

This class provides a standardized interface for graph database operations, including creating, retrieving, and deleting memory nodes and relationships.

Functions

close abstractmethod async

close() -> None

Closes the database connection.

Source code in memora/graph_db/base.py

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

create_agent abstractmethod async

create_agent(
    org_id: str,
    agent_label: str,
    user_id: Optional[str] = None,
) -> models.Agent

Creates a new agent in the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_label	Label/name for the agent. TYPE: str
user_id	Optional Short UUID of the user. This is used when the agent is created specifically for a user, indicating that both the organization and the user will have this agent. TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Agent	Agent containing: org_id: Short UUID string user_id: Optional Short UUID string agent_id: Short UUID string agent_label: Agent label/name created_at: DateTime object of when the agent was created

Source code in memora/graph_db/base.py

@abstractmethod
async def create_agent(
    self, org_id: str, agent_label: str, user_id: Optional[str] = None
) -> models.Agent:
    """
    Creates a new agent in the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_label (str): Label/name for the agent.
        user_id (Optional[str]): Optional Short UUID of the user. This is used when the agent is created
            specifically for a user, indicating that both the organization and the
            user will have this agent.

    Returns:
        Agent containing:

            + org_id: Short UUID string
            + user_id: Optional Short UUID string
            + agent_id: Short UUID string
            + agent_label: Agent label/name
            + created_at: DateTime object of when the agent was created
    """
    pass

create_organization abstractmethod async

create_organization(org_name: str) -> models.Organization

Creates a new organization in the graph database.

PARAMETER	DESCRIPTION
org_name	The name of the organization to create. TYPE: str

RETURNS	DESCRIPTION
Organization	Organization object containing: org_id: Short UUID string org_name: Organization name created_at: DateTime object of when the organization was created

Source code in memora/graph_db/base.py

@abstractmethod
async def create_organization(self, org_name: str) -> models.Organization:
    """
    Creates a new organization in the graph database.

    Args:
        org_name (str): The name of the organization to create.

    Returns:
        Organization object containing:

            + org_id: Short UUID string
            + org_name: Organization name
            + created_at: DateTime object of when the organization was created
    """
    pass

create_user abstractmethod async

create_user(org_id: str, user_name: str) -> models.User

Creates a new user in the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_name	Name for the user. TYPE: str

RETURNS	DESCRIPTION
User	User containing: org_id: Short UUID string user_id: Short UUID string user_name: User's name created_at: DateTime object of when the user was created

Source code in memora/graph_db/base.py

@abstractmethod
async def create_user(self, org_id: str, user_name: str) -> models.User:
    """
    Creates a new user in the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_name (str): Name for the user.

    Returns:
        User containing:

            + org_id: Short UUID string
            + user_id: Short UUID string
            + user_name: User's name
            + created_at: DateTime object of when the user was created
    """
    pass

delete_agent abstractmethod async

delete_agent(org_id: str, agent_id: str) -> None

Deletes an agent from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_id	Short UUID string identifying the agent to delete. TYPE: str

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_agent(self, org_id: str, agent_id: str) -> None:
    """
    Deletes an agent from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_id (str): Short UUID string identifying the agent to delete.
    """
    pass

delete_all_user_interactions_and_their_memories abstractmethod async

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

Deletes all interactions and their associated memories for a specific user in an organization.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user whose interactions should be deleted TYPE: str

Note

If the graph database is associated with a vector database, the memories are also deleted there for data consistency.

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_all_user_interactions_and_their_memories(
    self,
    org_id: str,
    user_id: str,
) -> None:
    """
    Deletes all interactions and their associated memories for a specific user in an organization.

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user whose interactions should be deleted

    Note:
        If the graph database is associated with a vector database, the memories are also deleted there for data consistency.
    """
    pass

delete_all_user_memories abstractmethod async

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

Deletes all memories of a specific user.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user TYPE: str

Note

If the graph database is associated with a vector database, the memories are also deleted there for data consistency.

Source code in memora/graph_db/base.py

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

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user

    Note:
        If the graph database is associated with a vector database, the memories are also deleted there for data consistency.
    """
    pass

delete_organization abstractmethod async

delete_organization(org_id: str) -> None

Deletes an organization from the graph database.

Warning

This operation will delete all nodes and relationships from this organization including users, agents, memories, interactions etc.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization to delete. TYPE: str

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_organization(self, org_id: str) -> None:
    """
    Deletes an organization from the graph database.

    Warning:
        This operation will delete all nodes and relationships from this organization
        including users, agents, memories, interactions etc.

    Args:
        org_id (str): Short UUID string identifying the organization to delete.
    """
    pass

delete_user abstractmethod async

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

Deletes a user from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user to delete. TYPE: str

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_user(self, org_id: str, user_id: str) -> None:
    """
    Deletes a user from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user to delete.
    """
    pass

delete_user_interaction_and_its_memories abstractmethod async

delete_user_interaction_and_its_memories(
    org_id: str, user_id: str, interaction_id: str
) -> None

Deletes an interaction record and its associated memories.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str
interaction_id	Short UUID string identifying the interaction to delete. TYPE: str

Note

If the graph database is associated with a vector database, the memories are also deleted there for data consistency.

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_user_interaction_and_its_memories(
    self,
    org_id: str,
    user_id: str,
    interaction_id: str,
) -> None:
    """
    Deletes an interaction record and its associated memories.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user.
        interaction_id (str): Short UUID string identifying the interaction to delete.

    Note:
        If the graph database is associated with a vector database, the memories are also deleted there for data consistency.
    """
    pass

delete_user_memory abstractmethod async

delete_user_memory(
    org_id: str, user_id: str, memory_id: str
) -> None

Deletes a specific memory.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user TYPE: str
memory_id	UUID string identifying the memory to delete TYPE: str

Note

If the graph database is associated with a vector database, the memory is also deleted there for data consistency.

Source code in memora/graph_db/base.py

@abstractmethod
async def delete_user_memory(
    self,
    org_id: str,
    user_id: str,
    memory_id: str,
) -> None:
    """
    Deletes a specific memory.

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user
        memory_id (str): UUID string identifying the memory to delete

    Note:
        If the graph database is associated with a vector database, the memory is also deleted there for data consistency.
    """
    pass

fetch_user_memories_resolved abstractmethod async

fetch_user_memories_resolved(
    org_user_mem_ids: List[Dict[str, str]]
) -> List[models.Memory]

Fetches memories from the GraphDB by their IDs, resolves any contrary updates, and replaces user/agent placeholders with actual names.

This method performs several operations

1. Retrieves memories using (org_id, user_id, memory_ids)
2. If a memory has a CONTRARY_UPDATE relationship, uses the newer memory version
3. Replaces user_id & agent_id placeholders (e.g 'user_abc123' or 'agent_xyz789') in memories with actual user names / agent labels

PARAMETER	DESCRIPTION
org_user_mem_ids	List of Dicts containing org, user, and memory ids of the memories to fetch and process TYPE: List[Dict[str, str]]

RETURNS

DESCRIPTION

List[Memory]

List[Memory] containing memory details: org_id: Short UUID string identifying the organization agent_id: Short UUID string identifying the agent user_id: Short UUID string identifying the user interaction_id: Short UUID string identifying the interaction the memory was sourced from memory_id: Full UUID string identifying the memory memory: The resolved memory obtained_at: DateTime object of when the memory was obtained message_sources: List of messages in the interaction that triggered the memory

Example

>>> org_user_mem_ids = [{'memory_id': '443ac3a8-fe87-49a4-93d2-05d3eb58ddeb', 'org_id': 'gmDr4sUiWMNqbGAiV8ijbU', 'user_id': 'CcyKXxhi2skEcDpRzNZim7'}, ...]
>>> memories = graphInstance.fetch_memories_resolved(org_user_mem_ids)
>>> print([memoryObj.memory for memoryObj in memories])
["John asked for help with a wedding ring", "Sarah is allergic to peanuts"]

Note

• Org, user, and memory IDs are typically retrieved from a vector database before being passed to this method.
• A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def fetch_user_memories_resolved(
    self, org_user_mem_ids: List[Dict[str, str]]
) -> List[models.Memory]:
    """
    Fetches memories from the GraphDB by their IDs, resolves any contrary updates, and replaces user/agent placeholders with actual names.

    This method performs several operations:
      1. Retrieves memories using (org_id, user_id, memory_ids)
      2. If a memory has a CONTRARY_UPDATE relationship, uses the newer memory version
      3. Replaces user_id & agent_id placeholders (e.g 'user_abc123' or 'agent_xyz789') in memories with actual user names / agent labels

    Args:
        org_user_mem_ids (List[Dict[str, str]]): List of Dicts containing org, user, and memory ids of the memories to fetch and process

    Returns:
        List[Memory] containing memory details:

            + org_id: Short UUID string identifying the organization
            + agent_id: Short UUID string identifying the agent
            + user_id: Short UUID string identifying the user
            + interaction_id: Short UUID string identifying the interaction the memory was sourced from
            + memory_id: Full UUID string identifying the memory
            + memory: The resolved memory
            + obtained_at: DateTime object of when the memory was obtained
            + message_sources: List of messages in the interaction that triggered the memory

    Example:
        ```python
        >>> org_user_mem_ids = [{'memory_id': '443ac3a8-fe87-49a4-93d2-05d3eb58ddeb', 'org_id': 'gmDr4sUiWMNqbGAiV8ijbU', 'user_id': 'CcyKXxhi2skEcDpRzNZim7'}, ...]
        >>> memories = graphInstance.fetch_memories_resolved(org_user_mem_ids)
        >>> print([memoryObj.memory for memoryObj in memories])
        ["John asked for help with a wedding ring", "Sarah is allergic to peanuts"]
        ```

    Note:
        - Org, user, and memory IDs are typically retrieved from a vector database before being passed to this method.
        - A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

fetch_user_memories_resolved_batch abstractmethod async

fetch_user_memories_resolved_batch(
    batch_org_user_mem_ids: List[List[Dict[str, str]]]
) -> List[List[models.Memory]]

Fetches memories from the GraphDB by their IDs, resolves any contrary updates, and replaces user/agent placeholders with actual names.

This method performs several operations

1. Retrieves memories using (org_id, user_id, memory_ids)
2. If a memory has a CONTRARY_UPDATE relationship, uses the newer memory version
3. Replaces user_id & agent_id placeholders (e.g 'user_abc123' or 'agent_xyz789') in memories with actual user names / agent labels

PARAMETER	DESCRIPTION
batch_org_user_mem_ids	List of lists containing Dicts with org, user, and memory ids of the memories to fetch and process TYPE: List[List[Dict[str, str]]]

RETURNS

DESCRIPTION

List[List[Memory]]

List[List[Memory]] with memory details: org_id: Short UUID string identifying the organization agent_id: Short UUID string identifying the agent user_id: Short UUID string identifying the user interaction_id: Short UUID string identifying the interaction the memory was sourced from memory_id: Full UUID string identifying the memory memory: The resolved memory obtained_at: DateTime object of when the memory was obtained message_sources: List of messages in the interaction that triggered the memory

Example

>>> batch_org_user_mem_ids = [[{"memory_id": "413ac3a8-fe87-49a4-93d2-05d3eb58ddeb", "org_id": "gmDr4sUiWMNqbGAiV8ijbU", "user_id": "CcyKXxhi2skEcDpRzNZim7"}, ...], [{...}, ...]]
>>> batch_memories = graphInstance.fetch_memories_resolved_batch(batch_org_user_mem_ids)
>>> print([[memoryObj.memory for memoryObj in memories] for memories in batch_memories])
[["John asked for help with a wedding ring", "Sarah is allergic to peanuts"], ["John is about to propose to Sarah"]]

Note

• Batch org, user, and memory IDs are typically retrieved from a vector database before being passed to this method.
• A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def fetch_user_memories_resolved_batch(
    self, batch_org_user_mem_ids: List[List[Dict[str, str]]]
) -> List[List[models.Memory]]:
    """
    Fetches memories from the GraphDB by their IDs, resolves any contrary updates, and replaces user/agent placeholders with actual names.

    This method performs several operations:
      1. Retrieves memories using (org_id, user_id, memory_ids)
      2. If a memory has a CONTRARY_UPDATE relationship, uses the newer memory version
      3. Replaces user_id & agent_id placeholders (e.g 'user_abc123' or 'agent_xyz789') in memories with actual user names / agent labels

    Args:
        batch_org_user_mem_ids (List[List[Dict[str, str]]]): List of lists containing Dicts with org, user, and memory ids of the memories to fetch and process

    Returns:
        List[List[Memory]] with memory details:

            + org_id: Short UUID string identifying the organization
            + agent_id: Short UUID string identifying the agent
            + user_id: Short UUID string identifying the user
            + interaction_id: Short UUID string identifying the interaction the memory was sourced from
            + memory_id: Full UUID string identifying the memory
            + memory: The resolved memory
            + obtained_at: DateTime object of when the memory was obtained
            + message_sources: List of messages in the interaction that triggered the memory

    Example:
        ```python
        >>> batch_org_user_mem_ids = [[{"memory_id": "413ac3a8-fe87-49a4-93d2-05d3eb58ddeb", "org_id": "gmDr4sUiWMNqbGAiV8ijbU", "user_id": "CcyKXxhi2skEcDpRzNZim7"}, ...], [{...}, ...]]
        >>> batch_memories = graphInstance.fetch_memories_resolved_batch(batch_org_user_mem_ids)
        >>> print([[memoryObj.memory for memoryObj in memories] for memories in batch_memories])
        [["John asked for help with a wedding ring", "Sarah is allergic to peanuts"], ["John is about to propose to Sarah"]]
        ```

    Note:
        - Batch org, user, and memory IDs are typically retrieved from a vector database before being passed to this method.
        - A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

get_agent abstractmethod async

get_agent(org_id: str, agent_id: str) -> models.Agent

Gets a specific agent belonging to the specified organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_id	Short UUID string identifying the agent to retrieve. TYPE: str

RETURNS	DESCRIPTION
Agent	Agent containing: org_id: Short UUID string user_id: Optional Short UUID string agent_id: Short UUID string agent_label: Agent label/name created_at: DateTime object of when the agent was created

Source code in memora/graph_db/base.py

@abstractmethod
async def get_agent(self, org_id: str, agent_id: str) -> models.Agent:
    """
    Gets a specific agent belonging to the specified organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_id (str): Short UUID string identifying the agent to retrieve.

    Returns:
        Agent containing:

            + org_id: Short UUID string
            + user_id: Optional Short UUID string
            + agent_id: Short UUID string
            + agent_label: Agent label/name
            + created_at: DateTime object of when the agent was created
    """
    pass

get_all_org_agents abstractmethod async

get_all_org_agents(org_id: str) -> List[models.Agent]

Gets all agents belonging to the specified organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str

RETURNS	DESCRIPTION
List[Agent]	A List[Agent], each containing: org_id: Short UUID string user_id: Optional Short UUID string agent_id: Short UUID string agent_label: Agent label/name created_at: DateTime object of when the agent was created

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_org_agents(self, org_id: str) -> List[models.Agent]:
    """
    Gets all agents belonging to the specified organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.

    Returns:
        A List[Agent], each containing:

            + org_id: Short UUID string
            + user_id: Optional Short UUID string
            + agent_id: Short UUID string
            + agent_label: Agent label/name
            + created_at: DateTime object of when the agent was created
    """
    pass

get_all_org_users abstractmethod async

get_all_org_users(org_id: str) -> List[models.User]

Gets all users belonging to the specified organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str

RETURNS	DESCRIPTION
List[User]	List[User], each containing: org_id: Short UUID string user_id: Short UUID string user_name: User's name created_at: DateTime object of when the user was created.

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_org_users(self, org_id: str) -> List[models.User]:
    """
    Gets all users belonging to the specified organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.

    Returns:
        List[User], each containing:

            + org_id: Short UUID string
            + user_id: Short UUID string
            + user_name: User's name
            + created_at: DateTime object of when the user was created.
    """
    pass

get_all_organizations abstractmethod async

get_all_organizations() -> List[models.Organization]

Gets all organizations from the graph database.

RETURNS	DESCRIPTION
List[Organization]	List[Organization] each containing: org_id: Short UUID string org_name: Organization name created_at: DateTime object of when the organization was created

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_organizations(self) -> List[models.Organization]:
    """
    Gets all organizations from the graph database.

    Returns:
        List[Organization] each containing:

            + org_id: Short UUID string
            + org_name: Organization name
            + created_at: DateTime object of when the organization was created
    """
    pass

get_all_user_agents abstractmethod async

get_all_user_agents(
    org_id: str, user_id: str
) -> List[models.Agent]

Gets all agents for a user within an organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str

RETURNS	DESCRIPTION
List[Agent]	A List[Agent], each containing: org_id: Short UUID string user_id: Optional Short UUID string agent_id: Short UUID string agent_label: Agent label/name created_at: DateTime object of when the agent was created

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_user_agents(
    self, org_id: str, user_id: str
) -> List[models.Agent]:
    """
    Gets all agents for a user within an organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user.

    Returns:
        A List[Agent], each containing:

            + org_id: Short UUID string
            + user_id: Optional Short UUID string
            + agent_id: Short UUID string
            + agent_label: Agent label/name
            + created_at: DateTime object of when the agent was created
    """
    pass

get_all_user_interactions abstractmethod async

get_all_user_interactions(
    org_id: str,
    user_id: str,
    with_their_messages: bool = True,
    with_their_memories: bool = True,
    skip: int = 0,
    limit: int = 100,
) -> List[models.Interaction]

Retrieves all interactions for a specific user in an organization.

Note

Interactions are sorted in descending order by their updated at datetime. (So most recent interactions are first).

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str
with_their_messages	Whether to also retrieve messages of an interaction. TYPE: bool DEFAULT: True
with_their_memories	Whether to also retrieve memories gotten across all occurrences of an interaction. TYPE: bool DEFAULT: True
skip	Number of interactions to skip. (Useful for pagination) TYPE: int DEFAULT: 0
limit	Maximum number of interactions to retrieve. (Useful for pagination) TYPE: int DEFAULT: 100

RETURNS

DESCRIPTION

List[Interaction]

List[Interaction], each containing an Interaction with: org_id: Short UUID string identifying the organization. user_id: Short UUID string identifying the user. agent_id: Short UUID string identifying the agent. interaction_id: Short UUID string identifying the interaction. created_at: DateTime object of when the interaction was created. updated_at: DateTime object of when the interaction was last updated. messages (if with_their_messages = True): List of messages in the interaction. memories (if with_their_memories = True): List of memories gotten from all occurrences of this interaction.

Note

A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_user_interactions(
    self,
    org_id: str,
    user_id: str,
    with_their_messages: bool = True,
    with_their_memories: bool = True,
    skip: int = 0,
    limit: int = 100,
) -> List[models.Interaction]:
    """
    Retrieves all interactions for a specific user in an organization.

    Note:
        Interactions are sorted in descending order by their updated at datetime. (So most recent interactions are first).

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user.
        with_their_messages (bool): Whether to also retrieve messages of an interaction.
        with_their_memories (bool): Whether to also retrieve memories gotten across all occurrences of an interaction.
        skip (int): Number of interactions to skip. (Useful for pagination)
        limit (int): Maximum number of interactions to retrieve. (Useful for pagination)

    Returns:
        List[Interaction], each containing an Interaction with:

            + org_id: Short UUID string identifying the organization.
            + user_id: Short UUID string identifying the user.
            + agent_id: Short UUID string identifying the agent.
            + interaction_id: Short UUID string identifying the interaction.
            + created_at: DateTime object of when the interaction was created.
            + updated_at: DateTime object of when the interaction was last updated.
            + messages (if `with_their_messages` = True): List of messages in the interaction.
            + memories (if `with_their_memories` = True): List of memories gotten from all occurrences of this interaction.

    Note:
        A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

get_all_user_memories abstractmethod async

get_all_user_memories(
    org_id: str,
    user_id: str,
    agent_id: Optional[str] = None,
    skip: int = 0,
    limit: int = 1000,
) -> List[models.Memory]

Retrieves all memories associated with a specific user.

Note

Memories are sorted in descending order by their obtained at datetime. (So most recent memories are first).

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user TYPE: str
agent_id	Optional short UUID string identifying the agent. If provided, only memories obtained from interactions with this agent are returned. Otherwise, all memories associated with the user are returned. TYPE: Optional[str] DEFAULT: None
skip	Number of interactions to skip. (Useful for pagination) TYPE: int DEFAULT: 0
limit	Maximum number of interactions to retrieve. (Useful for pagination) TYPE: int DEFAULT: 1000

RETURNS

DESCRIPTION

List[Memory]

List[Memory] containing memory details: org_id: Short UUID string identifying the organization agent_id: Short UUID string identifying the agent user_id: Short UUID string identifying the user interaction_id: Short UUID string identifying the interaction the memory was sourced from memory_id: Full UUID string identifying the memory memory: The resolved memory obtained_at: DateTime object of when the memory was obtained message_sources: List of messages in the interaction that triggered the memory

Note

• A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def get_all_user_memories(
    self,
    org_id: str,
    user_id: str,
    agent_id: Optional[str] = None,
    skip: int = 0,
    limit: int = 1000,
) -> List[models.Memory]:
    """
    Retrieves all memories associated with a specific user.

    Note:
        Memories are sorted in descending order by their obtained at datetime. (So most recent memories are first).

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user
        agent_id (Optional[str]): Optional short UUID string identifying the agent. If provided, only memories obtained from
            interactions with this agent are returned.
            Otherwise, all memories associated with the user are returned.
        skip (int): Number of interactions to skip. (Useful for pagination)
        limit (int): Maximum number of interactions to retrieve. (Useful for pagination)

    Returns:
        List[Memory] containing memory details:

            + org_id: Short UUID string identifying the organization
            + agent_id: Short UUID string identifying the agent
            + user_id: Short UUID string identifying the user
            + interaction_id: Short UUID string identifying the interaction the memory was sourced from
            + memory_id: Full UUID string identifying the memory
            + memory: The resolved memory
            + obtained_at: DateTime object of when the memory was obtained
            + message_sources: List of messages in the interaction that triggered the memory

    Note:
        - A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

get_associated_vector_db abstractmethod

get_associated_vector_db() -> Optional[BaseVectorDB]

The vector database associated with the graph database, these is used inside the graph transactional blocks to ensure data consistency when handling memories across both stores (e.g., saving memories to the vector store and creating corresponding nodes in the graph db).

Source code in memora/graph_db/base.py

@abstractmethod
def get_associated_vector_db(self) -> Optional[BaseVectorDB]:
    """
    The vector database associated with the graph database, these is used inside the graph transactional blocks
    to ensure data consistency when handling memories across both stores (e.g., saving memories to the vector
    store and creating corresponding nodes in the graph db).
    """
    pass

get_interaction abstractmethod async

get_interaction(
    org_id: str,
    user_id: str,
    interaction_id: str,
    with_messages: bool = True,
    with_memories: bool = True,
) -> models.Interaction

Retrieves all messages associated with a specific interaction.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str
interaction_id	Short UUID string identifying the interaction. TYPE: str
with_messages	Whether to retrieve messages along with the interaction. TYPE: bool DEFAULT: True
with_memories	Whether to also retrieve memories gotten across all occurrences of this interaction. TYPE: bool DEFAULT: True

RETURNS

DESCRIPTION

Interaction

Interaction containing: org_id: Short UUID string identifying the organization. user_id: Short UUID string identifying the user. agent_id: Short UUID string identifying the agent. interaction_id: Short UUID string identifying the interaction. created_at: DateTime object of when the interaction was created. updated_at: DateTime object of when the interaction was last updated. messages (if with_messages = True): List of messages in the interaction. memories (if with_memories = True): List of memories gotten from all occurrences of this interaction.

Note

A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def get_interaction(
    self,
    org_id: str,
    user_id: str,
    interaction_id: str,
    with_messages: bool = True,
    with_memories: bool = True,
) -> models.Interaction:
    """
    Retrieves all messages associated with a specific interaction.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user.
        interaction_id (str): Short UUID string identifying the interaction.
        with_messages (bool): Whether to retrieve messages along with the interaction.
        with_memories (bool): Whether to also retrieve memories gotten across all occurrences of this interaction.

    Returns:
        Interaction containing:

            + org_id: Short UUID string identifying the organization.
            + user_id: Short UUID string identifying the user.
            + agent_id: Short UUID string identifying the agent.
            + interaction_id: Short UUID string identifying the interaction.
            + created_at: DateTime object of when the interaction was created.
            + updated_at: DateTime object of when the interaction was last updated.
            + messages (if `with_messages` = True): List of messages in the interaction.
            + memories (if `with_memories` = True): List of memories gotten from all occurrences of this interaction.

    Note:
        A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

get_organization abstractmethod async

get_organization(org_id: str) -> models.Organization

Gets a specific organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization to retrieve. TYPE: str

RETURNS	DESCRIPTION
Organization	Organization object containing: org_id: Short UUID string org_name: Organization name created_at: DateTime object of when the organization was created

Source code in memora/graph_db/base.py

@abstractmethod
async def get_organization(self, org_id: str) -> models.Organization:
    """
    Gets a specific organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization to retrieve.

    Returns:
        Organization object containing:

            + org_id: Short UUID string
            + org_name: Organization name
            + created_at: DateTime object of when the organization was created
    """
    pass

get_user abstractmethod async

get_user(org_id: str, user_id: str) -> models.User

Gets a specific user belonging to the specified organization from the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user to retrieve. TYPE: str

RETURNS	DESCRIPTION
User	User containing: org_id: Short UUID string user_id: Short UUID string user_name: User's name created_at: DateTime object of when the user was created.

Source code in memora/graph_db/base.py

@abstractmethod
async def get_user(self, org_id: str, user_id: str) -> models.User:
    """
    Gets a specific user belonging to the specified organization from the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user to retrieve.

    Returns:
        User containing:

            + org_id: Short UUID string
            + user_id: Short UUID string
            + user_name: User's name
            + created_at: DateTime object of when the user was created.
    """
    pass

get_user_memory abstractmethod async

get_user_memory(
    org_id: str, user_id: str, memory_id: str
) -> models.Memory

Retrieves a specific memory.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user TYPE: str
memory_id	UUID string identifying the memory TYPE: str

RETURNS

DESCRIPTION

Memory

Memory containing memory details: org_id: Short UUID string identifying the organization agent_id: Short UUID string identifying the agent user_id: Short UUID string identifying the user interaction_id: Short UUID string identifying the interaction the memory was sourced from memory_id: Full UUID string identifying the memory memory: The resolved memory obtained_at: DateTime object of when the memory was obtained message_sources: List of messages in the interaction that triggered the memory

Note

• The memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def get_user_memory(
    self, org_id: str, user_id: str, memory_id: str
) -> models.Memory:
    """
    Retrieves a specific memory.

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user
        memory_id (str): UUID string identifying the memory

    Returns:
        Memory containing memory details:

            + org_id: Short UUID string identifying the organization
            + agent_id: Short UUID string identifying the agent
            + user_id: Short UUID string identifying the user
            + interaction_id: Short UUID string identifying the interaction the memory was sourced from
            + memory_id: Full UUID string identifying the memory
            + memory: The resolved memory
            + obtained_at: DateTime object of when the memory was obtained
            + message_sources: List of messages in the interaction that triggered the memory

    Note:
        - The memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

get_user_memory_history abstractmethod async

get_user_memory_history(
    org_id: str, user_id: str, memory_id: str
) -> List[models.Memory]

Retrieves the history of a specific memory.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization TYPE: str
user_id	Short UUID string identifying the user TYPE: str
memory_id	UUID string identifying the memory TYPE: str

RETURNS

DESCRIPTION

List[Memory]

List[Memory] containing the history of memory details in descending order (starting with the current version, to the oldest version): org_id: Short UUID string identifying the organization agent_id: Short UUID string identifying the agent user_id: Short UUID string identifying the user interaction_id: Short UUID string identifying the interaction the memory was sourced from memory_id: Full UUID string identifying the memory memory: The resolved memory obtained_at: DateTime object of when the memory was obtained message_sources: List of messages in the interaction that triggered the memory

Note

• A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See graph.update_interaction_and_memories

Source code in memora/graph_db/base.py

@abstractmethod
async def get_user_memory_history(
    self, org_id: str, user_id: str, memory_id: str
) -> List[models.Memory]:
    """
    Retrieves the history of a specific memory.

    Args:
        org_id (str): Short UUID string identifying the organization
        user_id (str): Short UUID string identifying the user
        memory_id (str): UUID string identifying the memory

    Returns:
        List[Memory] containing the history of memory details in descending order (starting with the current version, to the oldest version):

            + org_id: Short UUID string identifying the organization
            + agent_id: Short UUID string identifying the agent
            + user_id: Short UUID string identifying the user
            + interaction_id: Short UUID string identifying the interaction the memory was sourced from
            + memory_id: Full UUID string identifying the memory
            + memory: The resolved memory
            + obtained_at: DateTime object of when the memory was obtained
            + message_sources: List of messages in the interaction that triggered the memory

    Note:
        - A memory won't have a message source, if its interaction was updated with a conflicting conversation thread that lead to truncation of the former thread. See `graph.update_interaction_and_memories`
    """
    pass

save_interaction_with_memories abstractmethod async

save_interaction_with_memories(
    org_id: str,
    agent_id: str,
    user_id: str,
    memories_and_interaction: MemoriesAndInteraction,
) -> Tuple[str, datetime]

Creates a new interaction record with associated memories.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_id	Short UUID string identifying the agent. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str
memories_and_interaction	Contains both the interaction and the associated memories. TYPE: MemoriesAndInteraction

Note

If the graph database is associated with a vector database, the memories are also stored there for data consistency.

RETURNS	DESCRIPTION
Tuple[str, datetime]	Tuple[str, datetime] containing: interaction_id: Short UUID string identifying the created interaction created_at: DateTime object of when the interaction was created.

Source code in memora/graph_db/base.py

@abstractmethod
async def save_interaction_with_memories(
    self,
    org_id: str,
    agent_id: str,
    user_id: str,
    memories_and_interaction: MemoriesAndInteraction,
) -> Tuple[str, datetime]:
    """
    Creates a new interaction record with associated memories.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_id (str): Short UUID string identifying the agent.
        user_id (str): Short UUID string identifying the user.
        memories_and_interaction (MemoriesAndInteraction): Contains both the interaction and the associated memories.

    Note:
        If the graph database is associated with a vector database, the memories are also stored there for data consistency.

    Returns:
        Tuple[str, datetime] containing:

            + interaction_id: Short UUID string identifying the created interaction
            + created_at: DateTime object of when the interaction was created.
    """
    pass

setup abstractmethod async

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

Sets up the database, e.g., creates indexes, constraints, etc.

Source code in memora/graph_db/base.py

@abstractmethod
async def setup(self, *args, **kwargs) -> None:
    """
    Sets up the database, e.g., creates indexes, constraints, etc.
    """
    pass

update_agent abstractmethod async

update_agent(
    org_id: str, agent_id: str, new_agent_label: str
) -> models.Agent

Updates an existing agent in the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_id	Short UUID string identifying the agent to update. TYPE: str
new_agent_label	New label/name for the agent. TYPE: str

RETURNS	DESCRIPTION
Agent	Agent containing: org_id: Short UUID string user_id: Optional Short UUID string agent_id: Short UUID string agent_label: Agent label/name created_at: DateTime object of when the agent was created

Source code in memora/graph_db/base.py

@abstractmethod
async def update_agent(
    self, org_id: str, agent_id: str, new_agent_label: str
) -> models.Agent:
    """
    Updates an existing agent in the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_id (str): Short UUID string identifying the agent to update.
        new_agent_label (str): New label/name for the agent.

    Returns:
        Agent containing:

            + org_id: Short UUID string
            + user_id: Optional Short UUID string
            + agent_id: Short UUID string
            + agent_label: Agent label/name
            + created_at: DateTime object of when the agent was created
    """
    pass

update_interaction_and_memories abstractmethod async

update_interaction_and_memories(
    org_id: str,
    agent_id: str,
    user_id: str,
    interaction_id: str,
    updated_memories_and_interaction: MemoriesAndInteraction,
) -> Tuple[str, datetime]

Update an existing interaction record and add new memories.

Compares updated interaction with existing one

• If differences are found, truncates existing record from that point and replaces with updated version. Old memories from truncated message(s) remain but become standalone (no longer linked to truncated messages).
• If no differences, appends new messages from the update.

New memories are always added, regardless of interaction changes.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
agent_id	Short UUID string identifying the agent in the updated interaction. TYPE: str
user_id	Short UUID string identifying the user. TYPE: str
interaction_id	Short UUID string identifying the interaction to update. TYPE: str
updated_memories_and_interaction	Contains both the updated interaction and the associated new memories. TYPE: MemoriesAndInteraction

Note

If the graph database is associated with a vector database, the memories are also stored there for data consistency.

RETURNS	DESCRIPTION
Tuple[str, datetime]	Tuple[str, datetime] containing: interaction_id: Short UUID string identifying the updated interaction updated_at: DateTime object of when the interaction was last updated.

Source code in memora/graph_db/base.py

@abstractmethod
async def update_interaction_and_memories(
    self,
    org_id: str,
    agent_id: str,
    user_id: str,
    interaction_id: str,
    updated_memories_and_interaction: MemoriesAndInteraction,
) -> Tuple[str, datetime]:
    """
    Update an existing interaction record and add new memories.

    Compares updated interaction with existing one:
        - If differences are found, truncates existing record from that point and
        replaces with updated version. Old memories from truncated message(s)
        remain but become standalone (no longer linked to truncated messages).
        - If no differences, appends new messages from the update.

    New memories are always added, regardless of interaction changes.

    Args:
        org_id (str): Short UUID string identifying the organization.
        agent_id (str): Short UUID string identifying the agent in the updated interaction.
        user_id (str): Short UUID string identifying the user.
        interaction_id (str): Short UUID string identifying the interaction to update.
        updated_memories_and_interaction (MemoriesAndInteraction): Contains both the updated interaction and the associated new memories.

    Note:
        If the graph database is associated with a vector database, the memories are also stored there for data consistency.

    Returns:
        Tuple[str, datetime] containing:

            + interaction_id: Short UUID string identifying the updated interaction
            + updated_at: DateTime object of when the interaction was last updated.
    """
    pass

update_organization abstractmethod async

update_organization(
    org_id: str, new_org_name: str
) -> models.Organization

Updates an existing organization in the graph database.

PARAMETER	DESCRIPTION
org_id	The Short UUID of the organization to update. TYPE: str
new_org_name	The new name for the organization. TYPE: str

RETURNS	DESCRIPTION
Organization	Organization object containing: org_id: Short UUID string org_name: Organization name created_at: DateTime object of when the organization was created

Source code in memora/graph_db/base.py

@abstractmethod
async def update_organization(
    self, org_id: str, new_org_name: str
) -> models.Organization:
    """
    Updates an existing organization in the graph database.

    Args:
        org_id (str): The Short UUID of the organization to update.
        new_org_name (str): The new name for the organization.

    Returns:
        Organization object containing:

            + org_id: Short UUID string
            + org_name: Organization name
            + created_at: DateTime object of when the organization was created
    """
    pass

update_user abstractmethod async

update_user(
    org_id: str, user_id: str, new_user_name: str
) -> models.User

Updates an existing user in the graph database.

PARAMETER	DESCRIPTION
org_id	Short UUID string identifying the organization. TYPE: str
user_id	Short UUID string identifying the user to update. TYPE: str
new_user_name	The new name for the user. TYPE: str

RETURNS	DESCRIPTION
User	User containing: org_id: Short UUID string user_id: Short UUID string user_name: User's name created_at: DateTime object of when the user was created.

Source code in memora/graph_db/base.py

@abstractmethod
async def update_user(
    self, org_id: str, user_id: str, new_user_name: str
) -> models.User:
    """
    Updates an existing user in the graph database.

    Args:
        org_id (str): Short UUID string identifying the organization.
        user_id (str): Short UUID string identifying the user to update.
        new_user_name (str): The new name for the user.

    Returns:
        User containing:

            + org_id: Short UUID string
            + user_id: Short UUID string
            + user_name: User's name
            + created_at: DateTime object of when the user was created.
    """
    pass