Memory

Base memory and memory store classes.

BaseMemory

Bases: ABC

Base class for episodic memory.

Subclasses are responsible for three things:

1. Retrieval strategy — given a task and a store, decide which store primitive(s) to invoke and how to compose or score the results.
2. Prompt formatting — convert the retrieved episodes into the string that gets injected into the agent's system prompt.
3. Write-time transformation — optionally transform an episode before recording it (e.g. reflection, summarisation, fact extraction).

Source code in src/llm_agents_from_scratch/base/memory.py

class BaseMemory(ABC):
    """Base class for episodic memory.

    Subclasses are responsible for three things:

    1. **Retrieval strategy** — given a task and a store, decide which store
       primitive(s) to invoke and how to compose or score the results.
    2. **Prompt formatting** — convert the retrieved episodes into the string
       that gets injected into the agent's system prompt.
    3. **Write-time transformation** — optionally transform an episode before
       recording it (e.g. reflection, summarisation, fact extraction).
    """

    @abstractmethod
    async def recall(self, task: Task) -> str:
        """Retrieve relevant past episodes for a task.

        Called at the start of each task. Returns a formatted string ready to
        inject into the agent's system prompt under the recalled_episodes block.

        Args:
            task (Task): The incoming task used to select relevant episodes.

        Returns:
            str: Formatted episode context for the system prompt.
        """

    @abstractmethod
    async def record(self, episode: Episode) -> None:
        """Persist a completed episode.

        Called at the end of each task. Implementations may apply write-time
        transformations (e.g. reflection, summarisation) before storing.

        Args:
            episode (Episode): The completed task, trajectory, and result.
        """

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

        Useful for inspection and debugging — describes what is currently
        held in memory (e.g. episode count, date range, topics).

        Returns:
            str: A short summary of the memory contents.
        """

recall abstractmethod async

recall(task)

Retrieve relevant past episodes for a task.

Called at the start of each task. Returns a formatted string ready to inject into the agent's system prompt under the recalled_episodes block.

Parameters:

Name	Type	Description	Default
task	Task	The incoming task used to select relevant episodes.	required

Returns:

Name	Type	Description
str	str	Formatted episode context for the system prompt.

Source code in src/llm_agents_from_scratch/base/memory.py

@abstractmethod
async def recall(self, task: Task) -> str:
    """Retrieve relevant past episodes for a task.

    Called at the start of each task. Returns a formatted string ready to
    inject into the agent's system prompt under the recalled_episodes block.

    Args:
        task (Task): The incoming task used to select relevant episodes.

    Returns:
        str: Formatted episode context for the system prompt.
    """

record abstractmethod async

record(episode)

Persist a completed episode.

Called at the end of each task. Implementations may apply write-time transformations (e.g. reflection, summarisation) before storing.

Parameters:

Name	Type	Description	Default
episode	Episode	The completed task, trajectory, and result.	required

Source code in src/llm_agents_from_scratch/base/memory.py

@abstractmethod
async def record(self, episode: Episode) -> None:
    """Persist a completed episode.

    Called at the end of each task. Implementations may apply write-time
    transformations (e.g. reflection, summarisation) before storing.

    Args:
        episode (Episode): The completed task, trajectory, and result.
    """

summary abstractmethod async

summary()

Return a human-readable summary of the memory store.

Useful for inspection and debugging — describes what is currently held in memory (e.g. episode count, date range, topics).

Returns:

Name	Type	Description
str	str	A short summary of the memory contents.

Source code in src/llm_agents_from_scratch/base/memory.py

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

    Useful for inspection and debugging — describes what is currently
    held in memory (e.g. episode count, date range, topics).

    Returns:
        str: A short summary of the memory contents.
    """

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 — expose read_recent and search so that BaseMemory implementations can compose and score results without knowing the underlying storage details.

Source code in src/llm_agents_from_scratch/base/memory.py

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** — expose `read_recent` and `search` so that
       `BaseMemory` implementations can compose and score results without
       knowing the underlying storage details.
    """

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

        Args:
            episode (Episode): The completed episode to store.
            embedded_text (str | None): Pre-formatted text for
                substrates that embed episodes (e.g. vector stores).
                When provided, the store uses this text for embedding
                instead of formatting the episode itself. Substrates
                that do not embed (e.g. ``JSONMemoryStore``) ignore
                this argument. Defaults to ``None``.
        """

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

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

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

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

        Returns:
            int: Episode count.
        """

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

        Args:
            query (str): The search query (e.g. the task instruction).
            k (int): Maximum number of episodes to return.
            **kwargs: Optional substrate-specific search parameters
                (e.g. filters, score thresholds).

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

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

        Describes the substrate, episode count, and the oldest and newest
        episodes. Intended for inspection and debugging.

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

write abstractmethod async

write(episode, embedded_text=None)

Persist an episode to the store.

Parameters:

Name	Type	Description	Default
episode	Episode	The completed episode to store.	required
embedded_text	str | None	Pre-formatted text for substrates that embed episodes (e.g. vector stores). When provided, the store uses this text for embedding instead of formatting the episode itself. Substrates that do not embed (e.g. JSONMemoryStore) ignore this argument. Defaults to None.	None

Source code in src/llm_agents_from_scratch/base/memory.py

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

    Args:
        episode (Episode): The completed episode to store.
        embedded_text (str | None): Pre-formatted text for
            substrates that embed episodes (e.g. vector stores).
            When provided, the store uses this text for embedding
            instead of formatting the episode itself. Substrates
            that do not embed (e.g. ``JSONMemoryStore``) ignore
            this argument. Defaults to ``None``.
    """

read_recent abstractmethod async

read_recent(n)

Return the N most recently recorded episodes.

Parameters:

Name	Type	Description	Default
n	int	Maximum number of episodes to return.	required

Returns:

Type	Description
list[Episode]	list[Episode]: Episodes ordered from most recent to oldest.

Source code in src/llm_agents_from_scratch/base/memory.py

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

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

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

count abstractmethod async

count()

Return the total number of episodes in the store.

Returns:

Name	Type	Description
int	int	Episode count.

Source code in src/llm_agents_from_scratch/base/memory.py

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

    Returns:
        int: Episode count.
    """

search abstractmethod async

search(query, k, **kwargs)

Return the K episodes most relevant to a query.

Parameters:

Name	Type	Description	Default
query	str	The search query (e.g. the task instruction).	required
k	int	Maximum number of episodes to return.	required
**kwargs	Any	Optional substrate-specific search parameters (e.g. filters, score thresholds).	{}

Returns:

Type	Description
list[Episode]	list[Episode]: Episodes ordered by relevance to the query.

Source code in src/llm_agents_from_scratch/base/memory.py

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

    Args:
        query (str): The search query (e.g. the task instruction).
        k (int): Maximum number of episodes to return.
        **kwargs: Optional substrate-specific search parameters
            (e.g. filters, score thresholds).

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

summary abstractmethod async

summary()

Return a human-readable summary of the store contents.

Describes 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.py

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

    Describes the substrate, episode count, and the oldest and newest
    episodes. Intended for inspection and debugging.

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