Skip to content

Memory

Memory

Base class for Memory management.

This class acts as a high-level facade for memory operations, handling key scoping (namespacing) and delegating the actual storage logic to an underlying MemoryStore implementation (e.g., Redis or In-Memory).

Attributes:

Name Type Description
prefix str

The global prefix used for all keys generated by this instance.
store MemoryStore

The backend storage implementation.
Source code in blue/memories/memory.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
class Memory:
    """
    Base class for Memory management.

    This class acts as a high-level facade for memory operations, handling key scoping
    (namespacing) and delegating the actual storage logic to an underlying
    `MemoryStore` implementation (e.g., Redis or In-Memory).

    Attributes:
        prefix (str): The global prefix used for all keys generated by this instance.
        store (MemoryStore): The backend storage implementation.
    """

    def __init__(self, prefix: str, properties: Dict = None, store: MemoryStore = None):
        """
        Initialize the Memory instance.

        Determines the backing store based on the provided `store` instance or the
        configuration found in `properties`.

        Parameters:
            prefix (str): The base prefix for all keys generated by this instance
                (e.g., 'PLATFORM:default').
            properties (Dict, optional): Configuration dictionary containing connection details
                and protocol preferences (e.g., `{'memory_store.protocol': 'redis'}`).
                Defaults to None.
            store (MemoryStore, optional): An explicit, pre-initialized MemoryStore instance.
                If provided, `properties` are ignored for store initialization.

        Raises:
            Exception: If the 'memory_store.protocol' in `properties` is unknown or unsupported.
        """
        self.prefix = prefix

        if store:
            self.store = store
        else:
            # default to 'in-memory' if not specified, falling back safely if properties is None
            protocol = pydash.objects.get(properties, ['memory_store.protocol'], 'in-memory')
            if protocol == 'redis':
                self.store = RedisMemoryStore(properties)
            elif protocol == 'in-memory':
                self.store = InMemoryStore()
            else:
                raise Exception(f"Unknown memory store protocol: {protocol}")

        self.store.initialize()

    def _get_scoped_key(self, scope: str, identifier: str) -> str:
        """
        Generates a fully namespaced identifier (key) based on the hierarchy.

        The resulting key follows the pattern: `{prefix}:MEMORY:{scope}:{identifier}`.

        Parameters:
            scope (str): The logical scope of the memory (e.g., 'agent', 'session').
            identifier (str): The specific ID within that scope.

        Returns:
            str: The fully qualified key string.
        """
        return f"{self.prefix}:MEMORY:{scope}:{identifier}"

    # list delegates
    def _append_to_list(self, key: str, entry: Dict) -> bool:
        """
        Appends a dictionary entry to a list at the specified key.

        Delegates to the underlying `MemoryStore`.

        Parameters:
            key (str): The fully qualified key of the list.
            entry (Dict): The data to append.

        Returns:
            bool: True if the operation was successful, False otherwise.
        """
        return self.store.append_to_list(key, entry)

    def _get_list(self, key: str) -> List[Dict]:
        """
        Retrieves the entire list stored at the specified key.

        Delegates to the underlying `MemoryStore`.

        Parameters:
            key (str): The fully qualified key of the list.

        Returns:
            List[Dict]: The list of stored entries, or an empty list if not found.
        """
        return self.store.get_list(key)

    # indexed delegates
    def _store_indexed_entry(self, base_key: str, entry_id: str, entry: Dict, custom_key: Optional[str], timestamp: float) -> bool:
        """
        Atomically stores an entry and updates associated indices.

        Delegates to the underlying `MemoryStore`. This usually handles:
        1. Storing the actual data.
        2. Updating a secondary index for `custom_key` lookups.
        3. Updating a sorted set for time-range lookups.

        Parameters:
            base_key (str): The base key for the collection.
            entry_id (str): Unique ID for the specific entry.
            entry (Dict): The data payload to store.
            custom_key (Optional[str]): A secondary lookup key (e.g., a tag or label).
            timestamp (float): The timestamp associated with the entry (for time-range sorting).

        Returns:
            bool: True if the operation was successful, False otherwise.
        """
        return self.store.store_indexed_entry(base_key, entry_id, entry, custom_key, timestamp)

    def _get_entry_by_id(self, base_key: str, entry_id: str) -> Optional[Dict]:
        """
        Retrieves a specific entry by its unique ID.

        Delegates to the underlying `MemoryStore`.

        Parameters:
            base_key (str): The base key for the collection.
            entry_id (str): The unique ID of the entry.

        Returns:
            Optional[Dict]: The entry data if found, None otherwise.
        """
        return self.store.get_entry_by_id(base_key, entry_id)

    def _get_entry_by_key_index(self, base_key: str, custom_key: str) -> Optional[Dict]:
        """
        Retrieves a specific entry using a secondary custom key (index).

        Delegates to the underlying `MemoryStore`.

        Parameters:
            base_key (str): The base key for the collection.
            custom_key (str): The secondary lookup key.

        Returns:
            Optional[Dict]: The entry data if found, None otherwise.
        """
        return self.store.get_entry_by_key_index(base_key, custom_key)

    def _get_entries_by_time_range(self, base_key: str, start: float, end: float) -> List[Dict]:
        """
        Retrieves a list of entries falling within a specific time range.

        Delegates to the underlying `MemoryStore`.

        Parameters:
            base_key (str): The base key for the collection.
            start (float): The start timestamp (inclusive).
            end (float): The end timestamp (inclusive).

        Returns:
            List[Dict]: A list of entries found within the range.
        """
        return self.store.get_entries_by_time_range(base_key, start, end)

__init__(prefix, properties=None, store=None)

Initialize the Memory instance.

Determines the backing store based on the provided store instance or the configuration found in properties.

Parameters:

Name Type Description Default
prefix str

The base prefix for all keys generated by this instance (e.g., 'PLATFORM:default').

 required
properties Dict

Configuration dictionary containing connection details and protocol preferences (e.g., {'memory_store.protocol': 'redis'}). Defaults to None.

 None
store MemoryStore

An explicit, pre-initialized MemoryStore instance. If provided, properties are ignored for store initialization.

 None

Raises:

Type Description
Exception

If the 'memory_store.protocol' in properties is unknown or unsupported.
Source code in blue/memories/memory.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(self, prefix: str, properties: Dict = None, store: MemoryStore = None):
    """
    Initialize the Memory instance.

    Determines the backing store based on the provided `store` instance or the
    configuration found in `properties`.

    Parameters:
        prefix (str): The base prefix for all keys generated by this instance
            (e.g., 'PLATFORM:default').
        properties (Dict, optional): Configuration dictionary containing connection details
            and protocol preferences (e.g., `{'memory_store.protocol': 'redis'}`).
            Defaults to None.
        store (MemoryStore, optional): An explicit, pre-initialized MemoryStore instance.
            If provided, `properties` are ignored for store initialization.

    Raises:
        Exception: If the 'memory_store.protocol' in `properties` is unknown or unsupported.
    """
    self.prefix = prefix

    if store:
        self.store = store
    else:
        # default to 'in-memory' if not specified, falling back safely if properties is None
        protocol = pydash.objects.get(properties, ['memory_store.protocol'], 'in-memory')
        if protocol == 'redis':
            self.store = RedisMemoryStore(properties)
        elif protocol == 'in-memory':
            self.store = InMemoryStore()
        else:
            raise Exception(f"Unknown memory store protocol: {protocol}")

    self.store.initialize()

SessionMemory

Bases: Memory

Specialized Memory management for individual Sessions.

This class extends the base Memory class to provide scoped storage for session-specific data. It handles the generation of unique keys for each session and provides high-level methods for storing and retrieving memory entries (e.g., chat logs, state, context) associated with a specific session ID.

Attributes:

Name Type Description
prefix str

Inherited from Memory. The base prefix for keys.
store MemoryStore

Inherited from Memory. The underlying storage engine.
Source code in blue/memories/session_memory.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class SessionMemory(Memory):
    """
    Specialized Memory management for individual Sessions.

    This class extends the base `Memory` class to provide scoped storage for session-specific
    data. It handles the generation of unique keys for each session and provides high-level
    methods for storing and retrieving memory entries (e.g., chat logs, state, context)
    associated with a specific session ID.

    Attributes:
        prefix (str): Inherited from `Memory`. The base prefix for keys.
        store (MemoryStore): Inherited from `Memory`. The underlying storage engine.
    """

    def __init__(self, prefix: str, properties: Dict, store: MemoryStore = None):
        """
        Initialize the SessionMemory instance.

        Parameters:
            prefix (str): The base prefix for all keys (e.g., 'PLATFORM').
            properties (Dict): Configuration properties passed to the underlying store.
            store (MemoryStore, optional): Explicit store instance. Defaults to None.
        """
        super().__init__(prefix, properties, store)

    def _get_session_memory_key(self, session_id: str) -> str:
        """
        Generates the scoped base key for a specific session.

        Pattern: `{prefix}:MEMORY:SESSION:{session_id}`

        Parameters:
            session_id (str): The unique identifier of the session.

        Returns:
            str: The fully namespaced key for the session.
        """
        return self._get_scoped_key("SESSION", session_id)

    def store_session_memory(self, session_id: str, agent_id: str, data: Any, tags: List[str] = None, key: str = None) -> str:
        """
        Stores a new memory entry for a given session.

        This method automatically:
        1. Generates a unique entry ID (combining UUID and Agent ID).
        2. Captures the current timestamp.
        3. Wraps the payload in a standard envelope (metadata + data).
        4. Indexes the entry by ID, time, and optionally a custom key.

        Parameters:
            session_id (str): The session to attach this memory to.
            agent_id (str): The ID of the agent creating this memory.
            data (Any): The actual content to store (e.g., a message object, state dict).
            tags (List[str], optional): specific tags for categorization. Defaults to None.
            key (str, optional): A custom unique key for direct O(1) retrieval later.
                (e.g., 'latest_summary'). Defaults to None.

        Returns:
            str: The unique ID generated for the stored memory entry.
        """
        entry_id = f"memory_{uuid_utils.create_uuid()}_{agent_id}"
        timestamp = time.time()
        entry = {
            "id": entry_id,
            "agent_id": agent_id,
            "timestamp": timestamp,
            "data": data,
            "tags": tags or [],
            "key": key,  # optional custom key for direct lookup
        }
        self._store_indexed_entry(base_key=self._get_session_memory_key(session_id), entry_id=entry_id, entry=entry, custom_key=key, timestamp=timestamp)
        return entry_id

    def retrieve_session_memory_by_id(self, session_id: str, id: str) -> Optional[Dict]:
        """
        Retrieves a specific memory entry by its unique internal ID.

        Parameters:
            session_id (str): The session ID context.
            id (str): The unique entry ID (returned by `store_session_memory`).

        Returns:
            Optional[Dict]: The complete memory object if found, None otherwise.
        """
        return self._get_entry_by_id(base_key=self._get_session_memory_key(session_id), entry_id=id)

    def retrieve_session_memory_by_key(self, session_id: str, key: str) -> Optional[Dict]:
        """
        Retrieves a specific memory entry using a custom user-defined key.

        This is useful for fetching "singleton" memories like 'summary' or 'last_state'
        without knowing the generated UUID.

        Parameters:
            session_id (str): The session ID context.
            key (str): The custom key provided during storage.

        Returns:
            Optional[Dict]: The complete memory object if found, None otherwise.
        """
        return self._get_entry_by_key_index(base_key=self._get_session_memory_key(session_id), custom_key=key)

    def retrieve_session_memory_by_time(self, session_id: str, start_time: float, end_time: float) -> List[Dict]:
        """
        Retrieves all memory entries for a session within a specific time window.

        Parameters:
            session_id (str): The session ID context.
            start_time (float): The start timestamp (inclusive, Unix epoch).
            end_time (float): The end timestamp (inclusive, Unix epoch).

        Returns:
            List[Dict]: A list of memory objects sorted by time (depending on store implementation).
        """
        return self._get_entries_by_time_range(base_key=self._get_session_memory_key(session_id), start=start_time, end=end_time)

    def retrieve_session_memory_by_similarity(self, session_id: str, query_text: str, limit: int = 3) -> List[Dict]:
        """
        Retrieves memory entries based on semantic similarity to a query text.

        Note:
            This method is currently a placeholder and not yet implemented.

        Parameters:
            session_id (str): The session ID context.
            query_text (str): The text to compare against memory embeddings.
            limit (int, optional): The maximum number of results to return. Defaults to 3.

        Returns:
            List[Dict]: A list of most relevant memory objects (currently None).
        """
        # TODO: Implement vector search logic here
        return None

__init__(prefix, properties, store=None)

Initialize the SessionMemory instance.

Parameters:

Name Type Description Default
prefix str

The base prefix for all keys (e.g., 'PLATFORM').

 required
properties Dict

Configuration properties passed to the underlying store.

 required
store MemoryStore

Explicit store instance. Defaults to None.

 None
Source code in blue/memories/session_memory.py 
22
23
24
25
26
27
28
29
30
31
def __init__(self, prefix: str, properties: Dict, store: MemoryStore = None):
    """
    Initialize the SessionMemory instance.

    Parameters:
        prefix (str): The base prefix for all keys (e.g., 'PLATFORM').
        properties (Dict): Configuration properties passed to the underlying store.
        store (MemoryStore, optional): Explicit store instance. Defaults to None.
    """
    super().__init__(prefix, properties, store)

retrieve_session_memory_by_id(session_id, id)

Retrieves a specific memory entry by its unique internal ID.

Parameters:

Name Type Description Default
session_id str

The session ID context.

 required
id str

The unique entry ID (returned by store_session_memory).

 required

Returns:

Type Description
Optional[Dict]

Optional[Dict]: The complete memory object if found, None otherwise.
Source code in blue/memories/session_memory.py 
81
82
83
84
85
86
87
88
89
90
91
92
def retrieve_session_memory_by_id(self, session_id: str, id: str) -> Optional[Dict]:
    """
    Retrieves a specific memory entry by its unique internal ID.

    Parameters:
        session_id (str): The session ID context.
        id (str): The unique entry ID (returned by `store_session_memory`).

    Returns:
        Optional[Dict]: The complete memory object if found, None otherwise.
    """
    return self._get_entry_by_id(base_key=self._get_session_memory_key(session_id), entry_id=id)

retrieve_session_memory_by_key(session_id, key)

Retrieves a specific memory entry using a custom user-defined key.

This is useful for fetching "singleton" memories like 'summary' or 'last_state' without knowing the generated UUID.

Parameters:

Name Type Description Default
session_id str

The session ID context.

 required
key str

The custom key provided during storage.

 required

Returns:

Type Description
Optional[Dict]

Optional[Dict]: The complete memory object if found, None otherwise.
Source code in blue/memories/session_memory.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def retrieve_session_memory_by_key(self, session_id: str, key: str) -> Optional[Dict]:
    """
    Retrieves a specific memory entry using a custom user-defined key.

    This is useful for fetching "singleton" memories like 'summary' or 'last_state'
    without knowing the generated UUID.

    Parameters:
        session_id (str): The session ID context.
        key (str): The custom key provided during storage.

    Returns:
        Optional[Dict]: The complete memory object if found, None otherwise.
    """
    return self._get_entry_by_key_index(base_key=self._get_session_memory_key(session_id), custom_key=key)

retrieve_session_memory_by_similarity(session_id, query_text, limit=3)

Retrieves memory entries based on semantic similarity to a query text.

Note

This method is currently a placeholder and not yet implemented.

Parameters:

Name Type Description Default
session_id str

The session ID context.

 required
query_text str

The text to compare against memory embeddings.

 required
limit int

The maximum number of results to return. Defaults to 3.

 3

Returns:

Type Description
List[Dict]

List[Dict]: A list of most relevant memory objects (currently None).
Source code in blue/memories/session_memory.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def retrieve_session_memory_by_similarity(self, session_id: str, query_text: str, limit: int = 3) -> List[Dict]:
    """
    Retrieves memory entries based on semantic similarity to a query text.

    Note:
        This method is currently a placeholder and not yet implemented.

    Parameters:
        session_id (str): The session ID context.
        query_text (str): The text to compare against memory embeddings.
        limit (int, optional): The maximum number of results to return. Defaults to 3.

    Returns:
        List[Dict]: A list of most relevant memory objects (currently None).
    """
    # TODO: Implement vector search logic here
    return None

retrieve_session_memory_by_time(session_id, start_time, end_time)

Retrieves all memory entries for a session within a specific time window.

Parameters:

Name Type Description Default
session_id str

The session ID context.

 required
start_time float

The start timestamp (inclusive, Unix epoch).

 required
end_time float

The end timestamp (inclusive, Unix epoch).

 required

Returns:

Type Description
List[Dict]

List[Dict]: A list of memory objects sorted by time (depending on store implementation).
Source code in blue/memories/session_memory.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
def retrieve_session_memory_by_time(self, session_id: str, start_time: float, end_time: float) -> List[Dict]:
    """
    Retrieves all memory entries for a session within a specific time window.

    Parameters:
        session_id (str): The session ID context.
        start_time (float): The start timestamp (inclusive, Unix epoch).
        end_time (float): The end timestamp (inclusive, Unix epoch).

    Returns:
        List[Dict]: A list of memory objects sorted by time (depending on store implementation).
    """
    return self._get_entries_by_time_range(base_key=self._get_session_memory_key(session_id), start=start_time, end=end_time)

store_session_memory(session_id, agent_id, data, tags=None, key=None)

Stores a new memory entry for a given session.

This method automatically: 1. Generates a unique entry ID (combining UUID and Agent ID). 2. Captures the current timestamp. 3. Wraps the payload in a standard envelope (metadata + data). 4. Indexes the entry by ID, time, and optionally a custom key.

Parameters:

Name Type Description Default
session_id str

The session to attach this memory to.

 required
agent_id str

The ID of the agent creating this memory.

 required
data Any

The actual content to store (e.g., a message object, state dict).

 required
tags List[str]

specific tags for categorization. Defaults to None.

 None
key str

A custom unique key for direct O(1) retrieval later. (e.g., 'latest_summary'). Defaults to None.

 None

Returns:

Name Type Description
str str

The unique ID generated for the stored memory entry.
Source code in blue/memories/session_memory.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def store_session_memory(self, session_id: str, agent_id: str, data: Any, tags: List[str] = None, key: str = None) -> str:
    """
    Stores a new memory entry for a given session.

    This method automatically:
    1. Generates a unique entry ID (combining UUID and Agent ID).
    2. Captures the current timestamp.
    3. Wraps the payload in a standard envelope (metadata + data).
    4. Indexes the entry by ID, time, and optionally a custom key.

    Parameters:
        session_id (str): The session to attach this memory to.
        agent_id (str): The ID of the agent creating this memory.
        data (Any): The actual content to store (e.g., a message object, state dict).
        tags (List[str], optional): specific tags for categorization. Defaults to None.
        key (str, optional): A custom unique key for direct O(1) retrieval later.
            (e.g., 'latest_summary'). Defaults to None.

    Returns:
        str: The unique ID generated for the stored memory entry.
    """
    entry_id = f"memory_{uuid_utils.create_uuid()}_{agent_id}"
    timestamp = time.time()
    entry = {
        "id": entry_id,
        "agent_id": agent_id,
        "timestamp": timestamp,
        "data": data,
        "tags": tags or [],
        "key": key,  # optional custom key for direct lookup
    }
    self._store_indexed_entry(base_key=self._get_session_memory_key(session_id), entry_id=entry_id, entry=entry, custom_key=key, timestamp=timestamp)
    return entry_id
Last update: 2026-02-11