This module defines the core memory instance for a specific conversational context. This module provides the EpisodicMemory class, which acts as the primary orchestrator for an individual memory session. It integrates short-term (session) and long-term (declarative) memory stores to provide a unified interface for adding and retrieving conversational data. Key responsibilities include:
  • Managing the lifecycle of the memory instance through reference counting.
  • Adding new conversational Episode objects to both session and declarative memory.
  • Retrieving relevant context for a query by searching both memory types.
  • Interacting with a language model for memory-related tasks.
  • Each instance is uniquely identified by a MemoryContext and managed by the EpisodicMemoryManager.

​
Initialization

To get started, you’ll need to initialize the MemMachineClient with your API key. 
from memmachine import MemMachineClient

client = MemMachineClient(api_key="your_api_key")

​
Creating an Episode

To create a new episode, use the create_episode method. 
episode_data = {
    "title": "My First Episode",
    "description": "This is a test episode for the documentation."
}

try:
    new_episode = client.episodic_memory.create_episode(episode_data)
    print(f"Created new episode with ID: {new_episode.id}")
except Exception as e:
    print(f"Error creating episode: {e}")

​
EpisodicMemory Objects

class EpisodicMemory()
Represents a single, isolated memory instance for a specific context. This class orchestrates the interaction between short-term (session) memory and long-term (declarative) memory. It manages the lifecycle of the memory, handles adding new information (episodes), and provides methods to retrieve contextual information for queries. Each instance is tied to a unique MemoryContext (defined by group, agent, user, and session IDs) and is managed by a central EpisodicMemoryManager.

​
__init__

def __init__(manager, config: dict, memory_context: MemoryContext)
Initializes a EpisodicMemory instance. Arguments:
  • manager - The EpisodicMemoryManager that created this instance.
  • config - A dictionary containing the configuration for this memory instance.
  • memory_context - The unique context for this memory instance.

​
reference

async def reference() -> bool
Increments the reference count for this instance. Used by the manager to track how many clients are actively using this memory instance. Returns: True if the reference was successfully added, False if the instance is already closed.

​
add_memory_episode

async def add_memory_episode(producer: str,
                             produced_for: str,
                             episode_content: str | list[float],
                             episode_type: str,
                             content_type: ContentType,
                             timestamp: datetime | None = None,
                             metadata: dict | None = None) -> bool
Adds a new memory episode to both session and declarative memory. Validates that the producer and recipient of the episode are part of the current memory context. Arguments:
  • producer - The ID of the user or agent that created the episode.
  • produced_for - The ID of the intended recipient.
  • episode_content - The content of the episode (string or vector).
  • episode_type - The type of the episode (e.g., ‘message’, ‘thought’).
  • content_type - The type of the content (e.g., STRING).
  • timestamp - The timestamp of the episode. Defaults to now().
  • metadata - Optional dictionary of user-defined metadata.
Returns: True if the episode was added successfully, False otherwise.

​
close

async def close()
Decrements the reference count and closes the instance if it reaches zero. When the reference count is zero, it closes the underlying memory stores and notifies the manager to remove this instance from its registry.

​
delete_data

async def delete_data()
Deletes all data from both session and declarative memory for this context. This is a destructive operation.

​
query_memory

async def query_memory(
        query: str,
        limit: int | None = None) -> tuple[list[Episode], list[str]]
Retrieves relevant context for a given query from all memory stores. It fetches episodes from both short-term (session) and long-term (declarative) memory, deduplicates them, and returns them along with any available summary. Arguments:
  • query - The query string to find context for.
  • limit - The maximum number of episodes to return.
  • filter - A dictionary of properties to filter the search in declarative memory.
Returns: A tuple containing a list of relevant Episode objects and a summary string.

​
formalize_query_with_context

async def formalize_query_with_context(query: str,
                                       limit: int | None = None) -> str
Constructs a finalized query string that includes context from memory. The context (summary and recent episodes) is prepended to the original query, formatted with XML-like tags for the language model to parse. Arguments:
  • query - The original query string.
  • limit - The maximum number of episodes to include in the context.
  • filter - A dictionary of properties to filter the search.
Returns: A new query string enriched with context.