Skip to content

BaseMemoryStore

Base memory store class.

BaseMemoryStore

Bases: ABC

Base class for episodic memory stores.

Subclasses are responsible for two things:

  1. Durable persistence — write episodes to the substrate, including any preprocessing required by that substrate (embedding, indexing, tokenization).
  2. Retrieval primitives — implement _read_recent and _search so that the concrete search() method can dispatch based on recall_mode without knowing the underlying storage details.

The public retrieval interface is search(). Subclasses must not override it; they implement _read_recent and _search instead.

Attributes:

Name Type Description
max_results int

Default number of results returned by search() when no explicit count is supplied by the caller.
recall_mode RecallMode

Controls how search() retrieves episodes. RecallMode.RECENT delegates to _read_recent; RecallMode.SEARCH delegates to _search.
Source code in src/llm_agents_from_scratch/base/memory_store.py 
 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
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
173
174
175
176
class BaseMemoryStore(ABC):
    """Base class for episodic memory stores.

    Subclasses are responsible for two things:

    1. **Durable persistence** — write episodes to the substrate, including
       any preprocessing required by that substrate (embedding, indexing,
       tokenization).
    2. **Retrieval primitives** — implement ``_read_recent`` and ``_search``
       so that the concrete ``search()`` method can dispatch based on
       ``recall_mode`` without knowing the underlying storage details.

    The public retrieval interface is ``search()``. Subclasses must not
    override it; they implement ``_read_recent`` and ``_search`` instead.

    Attributes:
        max_results (int): Default number of results returned by
            ``search()`` when no explicit count is supplied by the caller.
        recall_mode (RecallMode): Controls how ``search()`` retrieves
            episodes. ``RecallMode.RECENT`` delegates to ``_read_recent``;
            ``RecallMode.SEARCH`` delegates to ``_search``.
    """

    def __init__(
        self,
        max_results: int = 5,
        recall_mode: RecallMode = RecallMode.SEARCH,
    ) -> None:
        """Initialise shared store state.

        Args:
            max_results (int): Default maximum number of episodes
                returned by retrieval operations. Defaults to 5.
            recall_mode (RecallMode): Retrieval strategy used by
                ``search()``. Defaults to ``RecallMode.SEARCH``.
        """
        self.max_results = max_results
        self.recall_mode = recall_mode

    @abstractmethod
    async def write(self, episode: Episode) -> None:
        """Persist an episode to the store.

        Subclasses must implement this method.

        Args:
            episode (Episode): The completed episode to store.
        """

    @abstractmethod
    async def _read_recent(self, n: int) -> list[Episode]:
        """Return the N most recently recorded episodes.

        Subclasses must implement this method. It is called by the
        concrete ``search()`` when ``recall_mode`` is
        ``RecallMode.RECENT``.

        Args:
            n (int): Maximum number of episodes to return.

        Returns:
            list[Episode]: Episodes ordered from most recent to oldest.
        """

    @abstractmethod
    async def _search(
        self,
        query: str,
        **kwargs: Any,
    ) -> list[Episode]:
        """Return the most relevant episodes for a query.

        Subclasses must implement this method. It is called by the
        concrete ``search()`` when ``recall_mode`` is
        ``RecallMode.SEARCH``. The number of results is controlled by
        ``self.max_results``.

        Args:
            query (str): The search query (e.g. the task instruction).
            **kwargs: Optional substrate-specific search parameters
                (e.g. filters, score thresholds).

        Returns:
            list[Episode]: Episodes ordered by relevance to the query.
        """

    async def recall(
        self,
        query: str,
        **kwargs: Any,
    ) -> list[Episode]:
        """Return episodes according to ``recall_mode``.

        Dispatches to ``_read_recent`` when ``recall_mode`` is
        ``RecallMode.RECENT``, or to ``_search`` when it is
        ``RecallMode.SEARCH``. Subclasses must not override this method;
        implement ``_read_recent`` and ``_search`` instead.

        Args:
            query (str): The search query (e.g. the task instruction).
                Ignored when ``recall_mode`` is ``RecallMode.RECENT``.
            **kwargs: Optional substrate-specific search parameters
                forwarded to ``_search``.

        Returns:
            list[Episode]: Episodes ordered by recency or relevance
                depending on ``recall_mode``.
        """
        if self.recall_mode == RecallMode.RECENT:
            results = await self._read_recent(self.max_results)
        else:
            results = await self._search(query, **kwargs)
        if len(results) > self.max_results:
            warnings.warn(
                f"{type(self).__name__}._search returned {len(results)} results"
                f" but max_results={self.max_results}.",
                MaxResultsExceededWarning,
                stacklevel=2,
            )
        return results

    @abstractmethod
    async def delete(self, id_: str) -> None:
        """Delete an episode by its unique identifier.

        Subclasses must implement this method. Raises
        ``EpisodeNotFoundError`` if no episode with ``id_`` exists.

        Args:
            id_ (str): The ``Episode.id_`` of the episode to remove.
        """

    @abstractmethod
    async def update(self, episode: Episode) -> None:
        """Replace an existing episode with an updated version.

        Subclasses must implement this method. Matches the stored episode
        by ``episode.id_`` and replaces it in-place. Raises
        ``EpisodeNotFoundError`` if no matching episode exists.

        Args:
            episode (Episode): The updated episode. Matched by ``id_``.
        """

    @abstractmethod
    async def count(self) -> int:
        """Return the total number of episodes in the store.

        Subclasses must implement this method.

        Returns:
            int: Episode count.
        """

    @abstractmethod
    async def summary(self) -> str:
        """Return a human-readable summary of the store contents.

        Subclasses must implement this method. Should describe the
        substrate, episode count, and the oldest and newest episodes.
        Intended for inspection and debugging.

        Returns:
            str: Multi-line summary of the store.
        """

__init__ 

__init__(max_results=5, recall_mode=RecallMode.SEARCH)

Initialise shared store state.

Parameters:

Name Type Description Default
max_results int

Default maximum number of episodes returned by retrieval operations. Defaults to 5.

 5
recall_mode RecallMode

Retrieval strategy used by search(). Defaults to RecallMode.SEARCH.

 SEARCH
Source code in src/llm_agents_from_scratch/base/memory_store.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    max_results: int = 5,
    recall_mode: RecallMode = RecallMode.SEARCH,
) -> None:
    """Initialise shared store state.

    Args:
        max_results (int): Default maximum number of episodes
            returned by retrieval operations. Defaults to 5.
        recall_mode (RecallMode): Retrieval strategy used by
            ``search()``. Defaults to ``RecallMode.SEARCH``.
    """
    self.max_results = max_results
    self.recall_mode = recall_mode

write abstractmethod async 

write(episode)

Persist an episode to the store.

Subclasses must implement this method.

Parameters:

Name Type Description Default
episode Episode

The completed episode to store.

 required
Source code in src/llm_agents_from_scratch/base/memory_store.py 
51
52
53
54
55
56
57
58
59
@abstractmethod
async def write(self, episode: Episode) -> None:
    """Persist an episode to the store.

    Subclasses must implement this method.

    Args:
        episode (Episode): The completed episode to store.
    """

recall async 

recall(query, **kwargs)

Return episodes according to recall_mode.

Dispatches to _read_recent when recall_mode is RecallMode.RECENT, or to _search when it is RecallMode.SEARCH. Subclasses must not override this method; implement _read_recent and _search instead.

Parameters:

Name Type Description Default
query str

The search query (e.g. the task instruction). Ignored when recall_mode is RecallMode.RECENT.

 required
**kwargs Any

Optional substrate-specific search parameters forwarded to _search.

 {}

Returns:

Type Description
list[Episode]

list[Episode]: Episodes ordered by recency or relevance depending on recall_mode.
Source code in src/llm_agents_from_scratch/base/memory_store.py 
 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
async def recall(
    self,
    query: str,
    **kwargs: Any,
) -> list[Episode]:
    """Return episodes according to ``recall_mode``.

    Dispatches to ``_read_recent`` when ``recall_mode`` is
    ``RecallMode.RECENT``, or to ``_search`` when it is
    ``RecallMode.SEARCH``. Subclasses must not override this method;
    implement ``_read_recent`` and ``_search`` instead.

    Args:
        query (str): The search query (e.g. the task instruction).
            Ignored when ``recall_mode`` is ``RecallMode.RECENT``.
        **kwargs: Optional substrate-specific search parameters
            forwarded to ``_search``.

    Returns:
        list[Episode]: Episodes ordered by recency or relevance
            depending on ``recall_mode``.
    """
    if self.recall_mode == RecallMode.RECENT:
        results = await self._read_recent(self.max_results)
    else:
        results = await self._search(query, **kwargs)
    if len(results) > self.max_results:
        warnings.warn(
            f"{type(self).__name__}._search returned {len(results)} results"
            f" but max_results={self.max_results}.",
            MaxResultsExceededWarning,
            stacklevel=2,
        )
    return results

delete abstractmethod async 

delete(id_)

Delete an episode by its unique identifier.

Subclasses must implement this method. Raises EpisodeNotFoundError if no episode with id_ exists.

Parameters:

Name Type Description Default
id_ str

The Episode.id_ of the episode to remove.

 required
Source code in src/llm_agents_from_scratch/base/memory_store.py 
133
134
135
136
137
138
139
140
141
142
@abstractmethod
async def delete(self, id_: str) -> None:
    """Delete an episode by its unique identifier.

    Subclasses must implement this method. Raises
    ``EpisodeNotFoundError`` if no episode with ``id_`` exists.

    Args:
        id_ (str): The ``Episode.id_`` of the episode to remove.
    """

update abstractmethod async 

update(episode)

Replace an existing episode with an updated version.

Subclasses must implement this method. Matches the stored episode by episode.id_ and replaces it in-place. Raises EpisodeNotFoundError if no matching episode exists.

Parameters:

Name Type Description Default
episode Episode

The updated episode. Matched by id_.

 required
Source code in src/llm_agents_from_scratch/base/memory_store.py 
144
145
146
147
148
149
150
151
152
153
154
@abstractmethod
async def update(self, episode: Episode) -> None:
    """Replace an existing episode with an updated version.

    Subclasses must implement this method. Matches the stored episode
    by ``episode.id_`` and replaces it in-place. Raises
    ``EpisodeNotFoundError`` if no matching episode exists.

    Args:
        episode (Episode): The updated episode. Matched by ``id_``.
    """

count abstractmethod async 

count()

Return the total number of episodes in the store.

Subclasses must implement this method.

Returns:

Name Type Description
int int

Episode count.
Source code in src/llm_agents_from_scratch/base/memory_store.py 
156
157
158
159
160
161
162
163
164
@abstractmethod
async def count(self) -> int:
    """Return the total number of episodes in the store.

    Subclasses must implement this method.

    Returns:
        int: Episode count.
    """

summary abstractmethod async 

summary()

Return a human-readable summary of the store contents.

Subclasses must implement this method. Should describe the substrate, episode count, and the oldest and newest episodes. Intended for inspection and debugging.

Returns:

Name Type Description
str str

Multi-line summary of the store.
Source code in src/llm_agents_from_scratch/base/memory_store.py 
166
167
168
169
170
171
172
173
174
175
176
@abstractmethod
async def summary(self) -> str:
    """Return a human-readable summary of the store contents.

    Subclasses must implement this method. Should describe the
    substrate, episode count, and the oldest and newest episodes.
    Intended for inspection and debugging.

    Returns:
        str: Multi-line summary of the store.
    """