Episodic Memory — Research Accumulation and Persistence¶

This notebook demonstrates the recency_memory() factory, which gives an LLMAgent episodic memory that accumulates across tasks and survives agent restarts.

Two properties are illustrated:

• Recall across tasks. After each task the agent records an episode. Subsequent tasks receive the most recent max_results episodes injected into their system prompt, so the agent can answer synthesis questions without repeating tool calls.
• Persistence across restarts. Episodes are written to a JSONL file on disk. A brand-new agent instance pointing to the same file picks up all prior episodes immediately.

This notebook uses the same PokéAPI tool as ch04/pokemon_comparison.ipynb. Reading that notebook first is recommended.

In [ ]:

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI
# !pip install llm-agents-from-scratch

# Uncomment the line below to install `llm-agents-from-scratch` from PyPI # !pip install llm-agents-from-scratch

Running an Ollama service¶

To execute the code provided in this notebook, you'll need to have Ollama installed on your local machine and have its LLM hosting service running. To download Ollama, follow the instructions found on this page: https://ollama.com/download. After downloading and installing Ollama, you can start a service by opening a terminal and running the command ollama serve.

Setup¶

In [1]:

import os
import shutil
import subprocess
import time
import urllib.error
import urllib.request


def ensure_ollama(host="http://localhost:11434", timeout=15):
    """Start Ollama if not already running and wait until responsive."""

    def _up():
        try:
            urllib.request.urlopen(f"{host}/api/tags", timeout=1)
            return True
        except (urllib.error.URLError, ConnectionError, TimeoutError):
            return False

    if _up():
        return print(f"✓ Ollama already running at {host}")

    # Lightning persistent path first, then standard locations
    ollama_path = shutil.which("ollama")
    if ollama_path is None:
        for candidate in [
            "/teamspace/studios/this_studio/.local/bin/ollama",
            "/usr/local/bin/ollama",
            "/usr/bin/ollama",
        ]:
            if os.path.exists(candidate):
                ollama_path = candidate
                break
    if ollama_path is None:
        raise RuntimeError(
            "Could not find the ollama binary. Install with: "
            "curl -fsSL https://ollama.com/install.sh | sh",
        )

    print(f"Starting Ollama server ({ollama_path})...")
    subprocess.Popen(
        [ollama_path, "serve"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    deadline = time.time() + timeout
    while time.time() < deadline:
        if _up():
            return print(f"✓ Ollama up and running at {host}")
        time.sleep(0.5)

    raise RuntimeError(f"Ollama did not start within {timeout}s")


ensure_ollama()

import os import shutil import subprocess import time import urllib.error import urllib.request def ensure_ollama(host="http://localhost:11434", timeout=15): """Start Ollama if not already running and wait until responsive.""" def _up(): try: urllib.request.urlopen(f"{host}/api/tags", timeout=1) return True except (urllib.error.URLError, ConnectionError, TimeoutError): return False if _up(): return print(f"✓ Ollama already running at {host}") # Lightning persistent path first, then standard locations ollama_path = shutil.which("ollama") if ollama_path is None: for candidate in [ "/teamspace/studios/this_studio/.local/bin/ollama", "/usr/local/bin/ollama", "/usr/bin/ollama", ]: if os.path.exists(candidate): ollama_path = candidate break if ollama_path is None: raise RuntimeError( "Could not find the ollama binary. Install with: " "curl -fsSL https://ollama.com/install.sh | sh", ) print(f"Starting Ollama server ({ollama_path})...") subprocess.Popen( [ollama_path, "serve"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) deadline = time.time() + timeout while time.time() < deadline: if _up(): return print(f"✓ Ollama up and running at {host}") time.sleep(0.5) raise RuntimeError(f"Ollama did not start within {timeout}s") ensure_ollama()

✓ Ollama already running at http://localhost:11434

In [2]:

import json
import logging
import urllib.error
import urllib.request
from pathlib import Path

from llm_agents_from_scratch import LLMAgent
from llm_agents_from_scratch.data_structures import Task
from llm_agents_from_scratch.data_structures.memory import Episode
from llm_agents_from_scratch.llms import OllamaLLM
from llm_agents_from_scratch.logger import enable_console_logging
from llm_agents_from_scratch.memory.recipes import recency_memory
from llm_agents_from_scratch.tools.simple_function import SimpleFunctionTool

enable_console_logging(logging.INFO)

import json import logging import urllib.error import urllib.request from pathlib import Path from llm_agents_from_scratch import LLMAgent from llm_agents_from_scratch.data_structures import Task from llm_agents_from_scratch.data_structures.memory import Episode from llm_agents_from_scratch.llms import OllamaLLM from llm_agents_from_scratch.logger import enable_console_logging from llm_agents_from_scratch.memory.recipes import recency_memory from llm_agents_from_scratch.tools.simple_function import SimpleFunctionTool enable_console_logging(logging.INFO)

Defining the Tool¶

In [3]:

def get_pokemon(name: str) -> str:
    """Look up a Pokémon by name and return its types and base stats."""
    url = f"https://pokeapi.co/api/v2/pokemon/{name.lower().strip()}"
    req = urllib.request.Request(
        url,
        headers={"User-Agent": "llm-agents-from-scratch/1.0"},
    )
    try:
        with urllib.request.urlopen(req) as resp:
            data = json.loads(resp.read())
    except urllib.error.HTTPError as e:
        if e.code == 404:  # noqa: PLR2004
            raise ValueError(
                f"Pokémon '{name}' not found. "
                "Check the spelling and try again.",
            ) from e
        raise
    types = [t["type"]["name"] for t in data["types"]]
    stats = {s["stat"]["name"]: s["base_stat"] for s in data["stats"]}
    return json.dumps({"name": data["name"], "types": types, "stats": stats})


get_pokemon_tool = SimpleFunctionTool(func=get_pokemon)

def get_pokemon(name: str) -> str: """Look up a Pokémon by name and return its types and base stats.""" url = f"https://pokeapi.co/api/v2/pokemon/{name.lower().strip()}" req = urllib.request.Request( url, headers={"User-Agent": "llm-agents-from-scratch/1.0"}, ) try: with urllib.request.urlopen(req) as resp: data = json.loads(resp.read()) except urllib.error.HTTPError as e: if e.code == 404: # noqa: PLR2004 raise ValueError( f"Pokémon '{name}' not found. " "Check the spelling and try again.", ) from e raise types = [t["type"]["name"] for t in data["types"]] stats = {s["stat"]["name"]: s["base_stat"] for s in data["stats"]} return json.dumps({"name": data["name"], "types": types, "stats": stats}) get_pokemon_tool = SimpleFunctionTool(func=get_pokemon)

In [4]:

STORE_DIR = Path(".")
STORE_DIR.mkdir(exist_ok=True)
(STORE_DIR / "episodes.jsonl").unlink(
    missing_ok=True,
)  # start clean on each full run

STORE_DIR = Path(".") STORE_DIR.mkdir(exist_ok=True) (STORE_DIR / "episodes.jsonl").unlink( missing_ok=True, ) # start clean on each full run

Creating the Agent with Memory¶

recency_memory() returns a Memory instance backed by a JSONL file at STORE_DIR. It recalls the max_results most recent episodes and formats them for the system prompt. The agent holds the memory in its memories list; TaskHandler calls recall at task start and record at task end automatically.

In [5]:

memory = recency_memory(path=STORE_DIR, max_results=3)
llm = OllamaLLM(model="qwen3:14b", think=False)
agent = LLMAgent(llm=llm, tools=[get_pokemon_tool], memories=[memory])

memory = recency_memory(path=STORE_DIR, max_results=3) llm = OllamaLLM(model="qwen3:14b", think=False) agent = LLMAgent(llm=llm, tools=[get_pokemon_tool], memories=[memory])

Part 1 — Building Up Episodic Memory¶

Three sequential Pokémon lookups. Each task calls get_pokemon once and records an episode at the end. The memory summary after each task shows the store count growing from 0 to 3. By the time the fourth task runs, all three episodes will be recalled into its system prompt.

In [6]:

task1 = Task(
    instruction=(
        "What are Pikachu's base stats? "
        "Use the get_pokemon tool. Do not rely on prior knowledge."
    ),
)
result1 = await agent.run(task1)
print(result1)

task1 = Task( instruction=( "What are Pikachu's base stats? " "Use the get_pokemon tool. Do not rely on prior knowledge." ), ) result1 = await agent.run(task1) print(result1)

INFO (llm_agents_fs.LLMAgent) :      🚀 Starting task: What are Pikachu's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: What are Pikachu's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: I need to call the get_pokemon tool with the parameter "name" set to "Pikachu".
INFO (llm_agents_fs.TaskHandler) :      🧠 New Step: Call the get_pokemon tool with the parameter 'name' set to 'Pikachu'.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: Call the get_pokemon tool with the parameter 'name' set to 'Pikachu'.
INFO (llm_agents_fs.TaskHandler) :      🛠️ Executing Tool Call: get_pokemon
INFO (llm_agents_fs.TaskHandler) :      ✅ Successful Tool Call: {"name": "pikachu", "types": ["electric"], "stats": {"hp": 35, "attack": 55, "defense": 40, "special-attack": 50, "special-def...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: The base stats for Pikachu are as follows:
- HP: 35
- Attack: 55
- Defense: 40
- Special Attack: 50
- Special Defense: 50
- Speed: 90
INFO (llm_agents_fs.TaskHandler) :      No new step required.
INFO (llm_agents_fs.LLMAgent) :      🏁 Task completed: The base stats for Pikachu are as follows:
- HP: 35
- Attack: 55
- Defense: 40
- Special Attack: 50
- Special Defense: 50
- Speed: 9...[TRUNCATED]
The base stats for Pikachu are as follows:
- HP: 35
- Attack: 55
- Defense: 40
- Special Attack: 50
- Special Defense: 50
- Speed: 90

In [7]:

print(await agent.memories[0].summary())

print(await agent.memories[0].summary())

JSONMemoryStore: 1 episodes | path=episodes.jsonl
  newest: 2026-06-03 00:05:34 | What are Pikachu's base stats? Use the get_pokemon tool. Do 
  oldest: 2026-06-03 00:05:34 | What are Pikachu's base stats? Use the get_pokemon tool. Do 

In [8]:

task2 = Task(
    instruction=(
        "What are Gengar's base stats? "
        "Use the get_pokemon tool. Do not rely on prior knowledge."
    ),
)
result2 = await agent.run(task2)
print(result2)

task2 = Task( instruction=( "What are Gengar's base stats? " "Use the get_pokemon tool. Do not rely on prior knowledge." ), ) result2 = await agent.run(task2) print(result2)

INFO (llm_agents_fs.LLMAgent) :      🚀 Starting task: What are Gengar's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: What are Gengar's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: I need to call the get_pokemon tool with the name "Gengar" to retrieve its base stats.
INFO (llm_agents_fs.TaskHandler) :      🧠 New Step: Call the get_pokemon tool with the name 'Gengar' to retrieve its base stats.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: Call the get_pokemon tool with the name 'Gengar' to retrieve its base stats.
INFO (llm_agents_fs.TaskHandler) :      🛠️ Executing Tool Call: get_pokemon
INFO (llm_agents_fs.TaskHandler) :      ✅ Successful Tool Call: {"name": "gengar", "types": ["ghost", "poison"], "stats": {"hp": 60, "attack": 65, "defense": 60, "special-attack": 130, "spec...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: The base stats for Gengar are as follows:
- HP: 60
- Attack: 65
- Defense: 60
- Special Attack: 130
- Special Defense: 75
- Speed: 110
INFO (llm_agents_fs.TaskHandler) :      No new step required.
INFO (llm_agents_fs.LLMAgent) :      🏁 Task completed: The base stats for Gengar are as follows:
- HP: 60
- Attack: 65
- Defense: 60
- Special Attack: 130
- Special Defense: 75
- Speed: 1...[TRUNCATED]
The base stats for Gengar are as follows:
- HP: 60
- Attack: 65
- Defense: 60
- Special Attack: 130
- Special Defense: 75
- Speed: 110

In [9]:

print(await agent.memories[0].summary())

print(await agent.memories[0].summary())

JSONMemoryStore: 2 episodes | path=episodes.jsonl
  newest: 2026-06-03 00:06:33 | What are Gengar's base stats? Use the get_pokemon tool. Do n
  oldest: 2026-06-03 00:05:34 | What are Pikachu's base stats? Use the get_pokemon tool. Do 

In [10]:

task3 = Task(
    instruction=(
        "What are Mewtwo's base stats? "
        "Use the get_pokemon tool. Do not rely on prior knowledge."
    ),
)
result3 = await agent.run(task3)
print(result3)

task3 = Task( instruction=( "What are Mewtwo's base stats? " "Use the get_pokemon tool. Do not rely on prior knowledge." ), ) result3 = await agent.run(task3) print(result3)

INFO (llm_agents_fs.LLMAgent) :      🚀 Starting task: What are Mewtwo's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: What are Mewtwo's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: I need to call the get_pokemon tool with the parameter "name" set to "Mewtwo".
INFO (llm_agents_fs.TaskHandler) :      🧠 New Step: Call the get_pokemon tool with the parameter 'name' set to 'Mewtwo'.
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: Call the get_pokemon tool with the parameter 'name' set to 'Mewtwo'.
INFO (llm_agents_fs.TaskHandler) :      🛠️ Executing Tool Call: get_pokemon
INFO (llm_agents_fs.TaskHandler) :      ✅ Successful Tool Call: {"name": "mewtwo", "types": ["psychic"], "stats": {"hp": 106, "attack": 110, "defense": 90, "special-attack": 154, "special-de...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: The base stats for Mewtwo are as follows:
- HP: 106
- Attack: 110
- Defense: 90
- Special Attack: 154
- Special Defense: 90
- Speed: 13...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      No new step required.
INFO (llm_agents_fs.LLMAgent) :      🏁 Task completed: The base stats for Mewtwo are as follows:
- HP: 106
- Attack: 110
- Defense: 90
- Special Attack: 154
- Special Defense: 90
- Speed:...[TRUNCATED]
The base stats for Mewtwo are as follows:
- HP: 106
- Attack: 110
- Defense: 90
- Special Attack: 154
- Special Defense: 90
- Speed: 130

In [11]:

print(await agent.memories[0].summary())

print(await agent.memories[0].summary())

JSONMemoryStore: 3 episodes | path=episodes.jsonl
  newest: 2026-06-03 00:07:14 | What are Mewtwo's base stats? Use the get_pokemon tool. Do n
  oldest: 2026-06-03 00:05:34 | What are Pikachu's base stats? Use the get_pokemon tool. Do 

Part 2 — Synthesis from Recalled Episodes¶

The synthesis task asks the agent to compare all three Pokémon by speed. No stats are provided in the task instruction. The information is only available through the recalled episodes injected into the system prompt at the start of this run. A working memory means the agent can answer without calling get_pokemon again. Watch the logs: you should see no 🛠️ Executing Tool Call lines.

In [12]:

task4 = Task(
    instruction=(
        "Based on the Pokémon you have already researched, "
        "which one has the highest base speed? "
        "You do not need to call any tools. Use only what you already know."
    ),
)
result4 = await agent.run(task4)

task4 = Task( instruction=( "Based on the Pokémon you have already researched, " "which one has the highest base speed? " "You do not need to call any tools. Use only what you already know." ), ) result4 = await agent.run(task4)

INFO (llm_agents_fs.LLMAgent) :      🚀 Starting task: Based on the Pokémon you have already researched, which one has the highest base speed? You do not need to call any tools. Use only w...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: Based on the Pokémon you have already researched, which one has the highest base speed? You do not need to call any tools. Use onl...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: From the Pokémon I have researched:

- Mewtwo has a base speed of 130.
- Gengar has a base speed of 110.
- Pikachu has a base speed of ...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      No new step required.
INFO (llm_agents_fs.LLMAgent) :      🏁 Task completed: From the Pokémon I have researched:

- Mewtwo has a base speed of 130.
- Gengar has a base speed of 110.
- Pikachu has a base speed ...[TRUNCATED]

In [13]:

print(result4)

print(result4)

From the Pokémon I have researched:

- Mewtwo has a base speed of 130.
- Gengar has a base speed of 110.
- Pikachu has a base speed of 90.

Mewtwo has the highest base speed among the researched Pokémon.

Part 3 — Inspecting the Store on Disk¶

JSONMemoryStore writes one JSON line per episode. The cell below reads the raw file so you can see exactly what is persisted between tasks and what a new agent instance will load on the next run.

In [14]:

lines = (STORE_DIR / "episodes.jsonl").read_text().splitlines()
print(f"{len(lines)} episode(s) on disk\n")
for line in lines:
    episode = Episode.model_validate_json(line)
    print(str(episode))
    print()

lines = (STORE_DIR / "episodes.jsonl").read_text().splitlines() print(f"{len(lines)} episode(s) on disk\n") for line in lines: episode = Episode.model_validate_json(line) print(str(episode)) print()

4 episode(s) on disk

  <episode>
    <task>What are Pikachu's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.</task>
    <result>The base stats for Pikachu are as follows:
- HP: 35
- Attack: 55
- Defense: 40
- Special Attack: 50
- Special Defense: 50
- Speed: 90
    </result>
    <completed_at>2026-06-03 00:05:34</completed_at>
  </episode>

  <episode>
    <task>What are Gengar's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.</task>
    <result>The base stats for Gengar are as follows:
- HP: 60
- Attack: 65
- Defense: 60
- Special Attack: 130
- Special Defense: 75
- Speed: 110
    </result>
    <completed_at>2026-06-03 00:06:33</completed_at>
  </episode>

  <episode>
    <task>What are Mewtwo's base stats? Use the get_pokemon tool. Do not rely on prior knowledge.</task>
    <result>The base stats for Mewtwo are as follows:
- HP: 106
- Attack: 110
- Defense: 90
- Special Attack: 154
- Special Defense: 90
- Speed: 130
    </result>
    <completed_at>2026-06-03 00:07:14</completed_at>
  </episode>

  <episode>
    <task>Based on the Pokémon you have already researched, which one has the highest base speed? You do not need to call any tools. Use only what you already know.</task>
    <result>From the Pokémon I have researched:

- Mewtwo has a base speed of 130.
- Gengar has a base speed of 110.
- Pikachu has a base speed of 90.

Mewtwo has the highest base speed among the researched Pokémon.
    </result>
    <completed_at>2026-06-03 00:07:30</completed_at>
  </episode>

Part 4 — Persistence Across Agent Restarts¶

A fresh LLMAgent is created pointing at the same STORE_PATH. No state is shared with the previous instance. The new agent loads its episodes entirely from disk. The summary shows it starts with three episodes already in memory, and it can answer the synthesis question without calling any tools.

In [15]:

memory2 = recency_memory(path=STORE_DIR, max_results=3)
llm2 = OllamaLLM(model="qwen3:14b", think=False)
agent2 = LLMAgent(llm=llm2, tools=[get_pokemon_tool], memories=[memory2])

memory2 = recency_memory(path=STORE_DIR, max_results=3) llm2 = OllamaLLM(model="qwen3:14b", think=False) agent2 = LLMAgent(llm=llm2, tools=[get_pokemon_tool], memories=[memory2])

In [16]:

print(await agent2.memories[0].summary())

print(await agent2.memories[0].summary())

JSONMemoryStore: 4 episodes | path=episodes.jsonl
  newest: 2026-06-03 00:07:30 | Based on the Pokémon you have already researched, which one 
  oldest: 2026-06-03 00:05:34 | What are Pikachu's base stats? Use the get_pokemon tool. Do 

In [17]:

task5 = Task(
    instruction=(
        "Which Pokémon have you previously researched, and which one had "
        "the highest base speed? "
        "You do not need to call any tools. Use only what you already know."
    ),
)
result5 = await agent2.run(task5)

task5 = Task( instruction=( "Which Pokémon have you previously researched, and which one had " "the highest base speed? " "You do not need to call any tools. Use only what you already know." ), ) result5 = await agent2.run(task5)

INFO (llm_agents_fs.LLMAgent) :      🚀 Starting task: Which Pokémon have you previously researched, and which one had the highest base speed? You do not need to call any tools. Use only w...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ⚙️ Processing Step: Which Pokémon have you previously researched, and which one had the highest base speed? You do not need to call any tools. Use onl...[TRUNCATED]
INFO (llm_agents_fs.TaskHandler) :      ✅ Step Result: I have previously researched Mewtwo, Gengar, and Pikachu. Among these Pokémon, Mewtwo has the highest base speed with a value of 130.
INFO (llm_agents_fs.TaskHandler) :      No new step required.
INFO (llm_agents_fs.LLMAgent) :      🏁 Task completed: I have previously researched Mewtwo, Gengar, and Pikachu. Among these Pokémon, Mewtwo has the highest base speed with a value of 130...[TRUNCATED]

In [18]:

print(result5)

print(result5)

I have previously researched Mewtwo, Gengar, and Pikachu. Among these Pokémon, Mewtwo has the highest base speed with a value of 130.

Key Takeaway¶

recency_memory() gives the agent two things it would otherwise lack:

• Context across tasks. Every task after the first starts with the max_results most recent episodes injected into its system prompt. The agent can synthesise across past work without repeating tool calls (see Part 2 and Part 4).
• Persistence across restarts. Episodes are written to a JSONL file immediately after each task. Any new agent pointing at the same path resumes with full context — nothing is lost when the process ends.

The factory bundles the two layers that deliver these properties: a recency-based recall strategy that selects the N most recent episodes, and a file-backed store that handles persistence. Both are configured through the factory — path sets the storage directory; max_results controls the recall window.