Skip to main content
This module defines the core, isolated memory instance for one specific conversational context (like a single user’s chat with an AI). Think of it as the brain dedicated to one conversation. The EpisodicMemory class acts as the primary orchestrator for this session. It cleverly integrates short-term memory (recent chat history) and long-term memory (declarative facts and older information) to give you a unified view.

​
Key Responsibilities:

  • Unified Storage: Adds new conversational Episode objects to both short-term (Session) and long-term (Declarative) memory simultaneously.
  • Smart Retrieval: Finds the most relevant context for any query by searching both memory types and handling duplicates.
  • Context Generation: Formats retrieved memories into a structured (XML-like) query for seamless integration with your Language Models.
  • Lifecycle Management: Uses reference counting to ensure the memory instance is only active when needed, saving resources.

​
The EpisodicMemory Class

class EpisodicMemory()
Represents a single, isolated memory instance for a specific context. This class manages the interaction between short-term (session) memory and long-term (declarative) memory. Each instance is tied to a unique MemoryContext (defined by group, agent, user, and session IDs).

​
init

def __init__(manager, config: dict, memory_context: MemoryContext)
Initializes a new EpisodicMemory instance. This method handles all the setup, including configuring the internal short-term and long-term memory components based on your provided configuration dictionary. Arguments:
  • manager: The EpisodicMemoryManager that created this instance.
  • config: A dictionary containing all necessary configurations (model, prompts, and memory settings) for this memory instance.
  • memory_context: The unique context that defines this memory session (who is talking to whom).

​
reference

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

​
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 the short-term (session) and long-term (declarative) memory stores concurrently. This ensures that new information is instantly available for context retrieval. Arguments:
  • producer: The ID of the user or agent that created the episode (e.g., sent a message).
  • produced_for: The ID of the intended recipient.
  • episode_content: The content of the episode, which can be a raw text string or a pre-computed embedding vector (list[float]).
  • episode_type: The category of the episode (e.g., 'message', 'thought', 'summary').
  • content_type: Specifies the format of the episode_content (e.g., ContentType.STRING or ContentType.VECTOR).
  • timestamp: Optional. The timestamp of the episode. Defaults to the current time if not provided.
  • metadata: Optional. A dictionary for any custom, user-defined metadata you want to attach to the episode.
Returns: True if the episode was added successfully. (Will raise a ValueError if the producer or produced_for IDs are not part of the defined MemoryContext).

​
query_memory

async def query_memory(query: str,
                       limit: int | None = None,
                       property_filter: dict | None = None) -> tuple[list[Episode], list[Episode], list[str]]
The core search function. This method retrieves the most relevant contextual information for a given query from all memory stores (short-term and long-term). Arguments:
  • query: The question or query string used to search for context.
  • limit: Optional. The maximum number of episodes to return from each memory type. Defaults to 20.
  • property_filter: Optional. A dictionary of properties (key-value pairs) to filter the search in declarative memory.
Returns: A tuple containing three elements:
  1. A list of Episode objects found in short-term memory.
  2. A list of Episode objects found in long-term memory (already deduplicated).
  3. A list of summary strings generated from the short-term context.

​
formalize_query_with_context

async def formalize_query_with_context(query: str,
                                        limit: int | None = None,
                                        property_filter: dict | None = None) -> str
Constructs the final, complete query string to be sent to a Language Model. It calls query_memory to get the context, then formats the summary and episodes into a structured string (using <Summary>, <Episodes>, and <Query> tags) that is easy for the LLM to parse and reason over. Arguments:
  • query: The original query string.
  • limit: The maximum number of episodes to include in the context.
  • property_filter: Filters for the search.
Returns: A new query string enriched with context from memory, formatted for the LLM.

​
delete_data

async def delete_data()
CAUTION: This is a destructive operation. It deletes all associated data from both short-term (session) and long-term (declarative) memory for this specific MemoryContext.

​
close

async def close()
Decrements the internal reference count. If the count reaches zero, this method automatically closes the underlying memory stores, cleans up resources, and notifies the manager to remove the instance.

​
Async Context Manager

class AsyncEpisodicMemory
This helper class allows you to manage the EpisodicMemory lifecycle using Python’s asynchronous context manager (async with). This is the safest and cleanest way to ensure you call reference() and close() correctly. 
from memmachine.episodic_memory import AsyncEpisodicMemory
# Assuming you have an episodic_memory_instance from your manager...

async with AsyncEpisodicMemory(episodic_memory_instance) as memory:
    # 'memory' is the EpisodicMemory instance. 
    # reference() was called automatically when entering the block.
    await memory.add_memory_episode(...)
    # ...
# When exiting the 'async with' block, memory.close() is called automatically.