Memory Module#

The memory system in Mesa-LLM provides different types of memory implementations that enable agents to store and retrieve past events (conversations, observations, actions, messages, plans, etc.). Memory serves as the foundation for creating agents with persistent, contextual awareness that enhances their decision-making capabilities. The memory module contains two classes.

Usage in Mesa Simulations#

from mesa_llm.llm_agent import LLMAgent
from mesa_llm.memory.st_lt_memory import STLTMemory

class MyAgent(LLMAgent):
   def __init__(self, model, reasoning, **kwargs):
      super().__init__(model, reasoning, **kwargs)

      # Override default memory with custom configuration
      self.memory = STLTMemory(
            agent=self,
            short_term_capacity=10,    # Store 10 recent experiences
            consolidation_capacity=3, # Consolidate when 13 total entries
            llm_model="openai/gpt-4o-mini",
            display=True,             # Display the memory entries in the console when they are added to the memory
            api_base=None,            # Set to a custom URL for self-hosted LLMs
      )

Core memory interfaces#

class MemoryEntry(content: dict, step: int | None, agent: LLMAgent)[source]#

Bases: object

A data structure that stores individual memory records with content, step number, and agent reference. Each entry includes rich formatting for display. Content is a nested dictionary of arbitrary depth containing the entry’s information. Each entry is designed to hold all the information of a given step for an agent, but can also be used to store a single event if needed.

content: dict#

step: int | None#

agent: LLMAgent#

display()[source]#

Bases: ABC

Generic parent class for memory backends.

Attributes:

agent : the agent that the memory belongs to llm_model : the model to use for the summarization if used display : whether to display the memory additive_event_types : event types that accumulate multiple values

within a step. Defaults to {"message", "action"}.

Content Addition

Before each agent step, the agent can add new events to the memory through add_to_memory(type, content) so that the memory can be used to reason about the most recent events as well as the past events.
During the step, content for types in additive_event_types is accumulated as a list; all other types overwrite the previous value for that step.
At the end of the step, the memory is processed via process_step(), managing when memory entries are added,consolidated, displayed, or removed

Default behavior

By default, additive_event_types == {"message", "action"}.
Repeated message or action entries within one step are accumulated as a list.
Repeated observation or plan entries within one step overwrite the previous value unless configured otherwise.

Initialize the memory

Args:: agent : the agent that the memory belongs to llm_model : the model to use for summarization display : whether to display memory entries in the console api_base : the API base URL to use for the LLM provider additive_event_types : event types that should accumulate multiple

values within the same step instead of overwriting. Defaults to {"message", "action"}. For example, message and action accumulate by default, while observation and plan overwrite unless explicitly included here.

step_content: dict#

abstractmethod get_prompt_ready() → str[source]#: Get the memory in a format that can be used for reasoning

abstractmethod get_communication_history() → str[source]#: Get the communication history in a format that can be used for reasoning

abstractmethod process_step(pre_step: bool = False)[source]#

A function that is called before and after the step of the agent is called. It is implemented to ensure that the memory is up to date when the agent is starting a new step.

/!If you consider that you do not need this function, you can write “pass” in its implementation.

async aprocess_step(pre_step: bool = False)[source]#

add_to_memory(type: str, content: dict)[source]#

Add a new entry to the memory.

Event types in self.additive_event_types accumulate multiple values within the same step. All other types use overwrite semantics. By default, self.additive_event_types == {"message", "action"}. For example, repeated message entries are stored as a list, while repeated observation entries overwrite the previous value.

async aadd_to_memory(type: str, content: dict)[source]#

Memory implementations#

class STLTMemory(agent: LLMAgent, short_term_capacity: int = 5, consolidation_capacity: int = 2, display: bool = True, llm_model: str | None = None, api_base: str | None = None, additive_event_types: list[str] | set[str] | tuple[str, ...] | None = None)[source]#

Bases: Memory

Implements a dual-memory system where recent experiences are stored in short-term memory with limited capacity, and older memories are consolidated into long-term summaries using LLM-based summarization.

Attributes:

agent : the agent that the memory belongs to

Memory is composed of

A short term memory who stores the n (int) most recent interactions (observations, planning, discussions)
A long term memory that is a summary of the memories that are removed from short term memory (summary

completed/refactored as it goes) - Event types in additive_event_types accumulate within a step.

Defaults to {"message", "action"}.

Logic behind the implementation

Short-term capacity: Configurable number of recent memory entries (default: short_term_capacity = 5)
Consolidation: When capacity is exceeded, oldest entries are summarized into long-term memory (number of entries to summarize is configurable, default: consolidation_capacity = 3)
LLM Summarization: Uses a separate LLM instance to create meaningful summaries of past experiences

Initialize the memory

Args:: short_term_capacity : the number of interactions to store in the short term memory llm_model : the model to use for the summarization api_base : the API base URL to use for the LLM provider agent : the agent that the memory belongs to additive_event_types : event types that accumulate multiple values

within a step instead of overwriting. Defaults to {"message", "action"}.

process_step(pre_step: bool = False)[source]#: Synchronous memory step handler

async aprocess_step(pre_step: bool = False)[source]#: Async memory step handler (non-blocking consolidation)

format_long_term() → str[source]#: Get the long term memory

format_short_term() → str[source]#: Get the short term memory

get_prompt_ready() → str[source]#: Get the memory in a format that can be used for reasoning

get_communication_history() → str[source]#: Get the communication history

class ShortTermMemory(agent: LLMAgent, n: int = 5, display: bool = True, additive_event_types: list[str] | set[str] | tuple[str, ...] | None = None)[source]#

Bases: Memory

Simple short-term memory implementation without consolidation (stores recent entries up to capacity limit). Same functionality as STLTMemory but without the long-term memory and consolidation mechanism.

Attributes:: agent : the agent that the memory belongs to n : positive number of short-term memories to remember display : whether to display the memory llm_model : the model to use for the summarization additive_event_types : event types accumulated as lists within a step.

Defaults to {"message", "action"}.

Initialize short-term memory.

Args:: agent : the agent that owns this memory n : maximum number of finalized short-term entries to keep display : whether memory entries should be displayed additive_event_types : event types that accumulate multiple values

within a step instead of overwriting. Defaults to {"message", "action"}.

async aprocess_step(pre_step: bool = False)[source]#: Asynchronous version of process_step

process_step(pre_step: bool = False)[source]#: Process the step of the agent : - Capture pre-step content into the current in-progress step entry - Merge current and post-step content into one finalized entry - Display the new entry

format_short_term() → str[source]#: Get the short term memory

get_prompt_ready() → str[source]#: Get the memory in a format that can be used for reasoning

get_communication_history() → str[source]#: Get the communication history

class LongTermMemory(agent: LLMAgent, display: bool = True, llm_model: str = 'openai/gpt-4o-mini', api_base: str | None = None, additive_event_types: list[str] | set[str] | tuple[str, ...] | None = None)[source]#

Bases: Memory

Purely long-term memory class that tries to store everything the agent experiences.

Attributes:: agent : the agent that the memory belongs to display : whether to display the memory llm_model : the model to use for the summarization additive_event_types : event types accumulated as lists within a step.

Defaults to {"message", "action"}.

Initialize long-term memory.

Args:: agent : the agent that owns this memory display : whether memory entries should be displayed llm_model : the model used for long-term summarization api_base : the API base URL to use for the LLM provider additive_event_types : event types that accumulate multiple values

within a step instead of overwriting. Defaults to {"message", "action"}.

process_step(pre_step: bool = False)[source]#: Process the step of the agent: - Merge the new entry into long term memory - Display the new entry (Will display it only when a new entry is created in this call)

async aprocess_step(pre_step: bool = False)[source]#: Asynchronous version of process_step (non-blocking)

format_long_term() → str[source]#: Get the long term memory

get_prompt_ready() → str[source]#: Get the memory in a format that can be used for reasoning

get_communication_history() → str[source]#: Get the communication history

class EventGrade(*, grade: int)[source]#

Bases: BaseModel

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

grade: int#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

normalize_dict_values(scores: dict, min_target: float, max_target: float) → dict[source]#

Normalize dictionary values to a target range with min-max scaling.

This mirrors the min-max helper used in the Generative Agents reference retrieval implementation: joonspk-research/generative_agents

class EpisodicMemory(agent: LLMAgent, llm_model: str | None = None, display: bool = True, max_capacity: int = 200, considered_entries: int = 30, recency_decay: float = 0.995, api_base: str | None = None)[source]#

Bases: Memory

Event-level memory with LLM-based importance scoring and recency-aware retrieval.

Credit / references: - Paper: Generative Agents: Interactive Simulacra of Human Behavior

https://arxiv.org/abs/2304.03442

Reference retrieval code: joonspk-research/generative_agents

This implementation is inspired by the paper’s retrieval scoring design (component-wise min-max normalization, then weighted combination). It is not a strict copy of the original code: relevance scoring via embeddings is not implemented yet, and recency is computed from step age.

Initialize the EpisodicMemory.

Args:: agent : the agent that owns this memory llm_model : the model used to grade event importance display : whether to display memory entries in the console max_capacity : maximum number of finalized episodic entries to keep considered_entries : number of entries to consider during retrieval recency_decay : exponential decay factor for recency scoring api_base : the API base URL to use for the LLM provider

grade_event_importance(type: str, content: dict) → float[source]#: Grade this event based on the content respect to the previous memory entries

async agrade_event_importance(type: str, content: dict) → float[source]#: Asynchronous version of grade_event_importance

retrieve_top_k_entries(k: int) → list[MemoryEntry][source]#

Retrieve the top-k entries using normalized importance and recency.

Notes: - Inspired by Generative Agents retrieval scoring:

recency/importance/relevance are normalized separately and combined.

This implementation currently combines importance + recency only. Relevance (embedding cosine similarity with a focal query) is pending.

add_to_memory(type: str, content: dict)[source]#: grading logic + adding to memory function call

async aadd_to_memory(type: str, content: dict)[source]#: Async version of add_to_memory + grading logic

get_prompt_ready() → str[source]#: Get the memory in a format that can be used for reasoning

get_communication_history() → str[source]#: Get the communication history

async aprocess_step(pre_step: bool = False)[source]#

Asynchronous version of process_step.

EpisodicMemory persists entries at add-time and does not use two-phase pre/post-step buffering.

process_step(pre_step: bool = False)[source]#

Process step hook (no-op for episodic memory).

EpisodicMemory persists entries at add-time and does not use two-phase pre/post-step buffering.