Skip to content

JSONMemoryStore

JSONL-file-backed episodic memory store.

JSONMemoryStore

Bases: BaseMemoryStore

Episodic memory store backed by a JSONL file on disk.

Each episode is stored as one JSON object per line (JSONL format). New episodes are appended to the file — no full rewrite on each write. The full file is read into memory on construction and on read_recent.

Attributes:

Name Type Description
path Path

Full path to the backing JSONL file (dir / filename).
Source code in src/llm_agents_from_scratch/memory_stores/json.py 
 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
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class JSONMemoryStore(BaseMemoryStore):
    """Episodic memory store backed by a JSONL file on disk.

    Each episode is stored as one JSON object per line (JSONL format).
    New episodes are appended to the file — no full rewrite on each write.
    The full file is read into memory on construction and on ``read_recent``.

    Attributes:
        path (Path): Full path to the backing JSONL file (``dir / filename``).
    """

    def __init__(
        self,
        dir: Path,
        filename: str = "episodes.jsonl",
        max_results: int = 5,
        recall_mode: RecallMode = RecallMode.RECENT,
    ) -> None:
        """Initialize a JSONMemoryStore.

        Loads any existing episodes from ``dir / filename`` on construction.
        The file is created on the first ``write`` call if it does not yet
        exist.

        Args:
            dir (Path): Directory in which the backing JSONL file is stored.
                The caller decides the location; no default is imposed by the
                library.
            filename (str): Name of the JSONL file within ``dir``. Defaults
                to ``"episodes.jsonl"``.
            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.RECENT`` since
                this store does not support similarity search.
        """
        super().__init__(max_results=max_results, recall_mode=recall_mode)
        self.path = dir / filename
        self._episodes: list[Episode] = self._load()

    def _load(self) -> list[Episode]:
        if not self.path.exists():
            return []
        with open(self.path) as f:
            return [
                Episode.model_validate_json(line) for line in f if line.strip()
            ]

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

        Appends one JSON line to the backing file. Does not rewrite existing
        content.

        Args:
            episode (Episode): The completed episode to store.
        """
        self._episodes.append(episode)
        with open(self.path, "a") as f:
            f.write(episode.model_dump_json() + "\n")

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

        Reads all episodes from the in-memory list. A production
        implementation would tail the file to avoid reading it in full.

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

        Returns:
            list[Episode]: Episodes ordered from most recent to oldest.
        """
        return sorted(
            self._episodes,
            key=lambda e: e.completed_at,
            reverse=True,
        )[:n]

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

        Returns:
            int: Episode count.
        """
        return len(self._episodes)

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

        Includes the backing file path, total episode count, and the
        instruction and timestamp of the newest and oldest episodes.

        Returns:
            str: Multi-line summary of the store.
        """
        total = len(self._episodes)
        lines = [f"JSONMemoryStore: {total} episodes | path={self.path}"]
        if total > 0:
            by_time = sorted(
                self._episodes,
                key=lambda e: e.completed_at,
            )
            oldest = by_time[0]
            newest = by_time[-1]
            lines.append(
                f"  newest: {str(newest.completed_at)[:19]}"
                f" | {newest.task.instruction[:60]}",
            )
            lines.append(
                f"  oldest: {str(oldest.completed_at)[:19]}"
                f" | {oldest.task.instruction[:60]}",
            )
        return "\n".join(lines)

    def _rewrite_jsonl(self) -> None:
        with open(self.path, "w") as f:
            for ep in self._episodes:
                f.write(ep.model_dump_json() + "\n")

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

        Raises ``EpisodeNotFoundError`` if no episode with ``id_`` exists.
        Otherwise removes it from the in-memory list and rewrites the
        backing file.

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

        Raises:
            EpisodeNotFoundError: If no episode with ``id_`` exists.
        """
        idx = next(
            (i for i, ep in enumerate(self._episodes) if ep.id_ == id_),
            None,
        )
        if idx is None:
            raise EpisodeNotFoundError(
                f"Episode '{id_}' not found in JSONMemoryStore.",
            )
        self._episodes.pop(idx)
        self._rewrite_jsonl()

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

        Matches by ``episode.id_``. Raises ``EpisodeNotFoundError`` if no
        matching episode exists. Otherwise replaces it in-place and rewrites
        the backing file.

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

        Raises:
            EpisodeNotFoundError: If no episode with ``episode.id_`` exists.
        """
        for i, ep in enumerate(self._episodes):
            if ep.id_ == episode.id_:
                self._episodes[i] = episode
                self._rewrite_jsonl()
                return
        raise EpisodeNotFoundError(
            f"Episode '{episode.id_}' not found in JSONMemoryStore.",
        )

    async def _search(
        self,
        query: str,
        **kwargs: Any,
    ) -> list[Episode]:
        """Raise ``NotImplementedError`` — this store does not embed episodes.

        Called by the base ``search()`` when ``recall_mode`` is
        ``RecallMode.SEARCH``. ``JSONMemoryStore`` does not support
        similarity search; use a vector-backed store instead.

        Args:
            query (str): Ignored.
            **kwargs: Ignored.

        Raises:
            NotImplementedError: Always. Use a vector-backed store for
                similarity search.
        """
        raise NotImplementedError(
            "JSONMemoryStore does not support similarity search. "
            "Use a vector-backed store instead.",
        )

__init__ 

__init__(
    dir,
    filename="episodes.jsonl",
    max_results=5,
    recall_mode=RecallMode.RECENT,
)

Initialize a JSONMemoryStore.

Loads any existing episodes from dir / filename on construction. The file is created on the first write call if it does not yet exist.

Parameters:

Name Type Description Default
dir Path

Directory in which the backing JSONL file is stored. The caller decides the location; no default is imposed by the library.

 required
filename str

Name of the JSONL file within dir. Defaults to "episodes.jsonl".

 'episodes.jsonl'
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.RECENT since this store does not support similarity search.

 RECENT
Source code in src/llm_agents_from_scratch/memory_stores/json.py 
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
def __init__(
    self,
    dir: Path,
    filename: str = "episodes.jsonl",
    max_results: int = 5,
    recall_mode: RecallMode = RecallMode.RECENT,
) -> None:
    """Initialize a JSONMemoryStore.

    Loads any existing episodes from ``dir / filename`` on construction.
    The file is created on the first ``write`` call if it does not yet
    exist.

    Args:
        dir (Path): Directory in which the backing JSONL file is stored.
            The caller decides the location; no default is imposed by the
            library.
        filename (str): Name of the JSONL file within ``dir``. Defaults
            to ``"episodes.jsonl"``.
        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.RECENT`` since
            this store does not support similarity search.
    """
    super().__init__(max_results=max_results, recall_mode=recall_mode)
    self.path = dir / filename
    self._episodes: list[Episode] = self._load()

write async 

write(episode)

Persist an episode to the store.

Appends one JSON line to the backing file. Does not rewrite existing content.

Parameters:

Name Type Description Default
episode Episode

The completed episode to store.

 required
Source code in src/llm_agents_from_scratch/memory_stores/json.py 
59
60
61
62
63
64
65
66
67
68
69
70
async def write(self, episode: Episode) -> None:
    """Persist an episode to the store.

    Appends one JSON line to the backing file. Does not rewrite existing
    content.

    Args:
        episode (Episode): The completed episode to store.
    """
    self._episodes.append(episode)
    with open(self.path, "a") as f:
        f.write(episode.model_dump_json() + "\n")

count 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/memory_stores/json.py 
90
91
92
93
94
95
96
async def count(self) -> int:
    """Return the total number of episodes in the store.

    Returns:
        int: Episode count.
    """
    return len(self._episodes)

summary async 

summary()

Return a human-readable summary of the store contents.

Includes the backing file path, total episode count, and the instruction and timestamp of the newest and oldest episodes.

Returns:

Name Type Description
str str

Multi-line summary of the store.
Source code in src/llm_agents_from_scratch/memory_stores/json.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
async def summary(self) -> str:
    """Return a human-readable summary of the store contents.

    Includes the backing file path, total episode count, and the
    instruction and timestamp of the newest and oldest episodes.

    Returns:
        str: Multi-line summary of the store.
    """
    total = len(self._episodes)
    lines = [f"JSONMemoryStore: {total} episodes | path={self.path}"]
    if total > 0:
        by_time = sorted(
            self._episodes,
            key=lambda e: e.completed_at,
        )
        oldest = by_time[0]
        newest = by_time[-1]
        lines.append(
            f"  newest: {str(newest.completed_at)[:19]}"
            f" | {newest.task.instruction[:60]}",
        )
        lines.append(
            f"  oldest: {str(oldest.completed_at)[:19]}"
            f" | {oldest.task.instruction[:60]}",
        )
    return "\n".join(lines)

delete async 

delete(id_)

Delete an episode by its unique identifier.

Raises EpisodeNotFoundError if no episode with id_ exists. Otherwise removes it from the in-memory list and rewrites the backing file.

Parameters:

Name Type Description Default
id_ str

The Episode.id_ of the episode to remove.

 required

Raises:

Type Description
EpisodeNotFoundError

If no episode with id_ exists.
Source code in src/llm_agents_from_scratch/memory_stores/json.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
async def delete(self, id_: str) -> None:
    """Delete an episode by its unique identifier.

    Raises ``EpisodeNotFoundError`` if no episode with ``id_`` exists.
    Otherwise removes it from the in-memory list and rewrites the
    backing file.

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

    Raises:
        EpisodeNotFoundError: If no episode with ``id_`` exists.
    """
    idx = next(
        (i for i, ep in enumerate(self._episodes) if ep.id_ == id_),
        None,
    )
    if idx is None:
        raise EpisodeNotFoundError(
            f"Episode '{id_}' not found in JSONMemoryStore.",
        )
    self._episodes.pop(idx)
    self._rewrite_jsonl()

update async 

update(episode)

Replace an existing episode with an updated version.

Matches by episode.id_. Raises EpisodeNotFoundError if no matching episode exists. Otherwise replaces it in-place and rewrites the backing file.

Parameters:

Name Type Description Default
episode Episode

The updated episode. Matched by id_.

 required

Raises:

Type Description
EpisodeNotFoundError

If no episode with episode.id_ exists.
Source code in src/llm_agents_from_scratch/memory_stores/json.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
async def update(self, episode: Episode) -> None:
    """Replace an existing episode with an updated version.

    Matches by ``episode.id_``. Raises ``EpisodeNotFoundError`` if no
    matching episode exists. Otherwise replaces it in-place and rewrites
    the backing file.

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

    Raises:
        EpisodeNotFoundError: If no episode with ``episode.id_`` exists.
    """
    for i, ep in enumerate(self._episodes):
        if ep.id_ == episode.id_:
            self._episodes[i] = episode
            self._rewrite_jsonl()
            return
    raise EpisodeNotFoundError(
        f"Episode '{episode.id_}' not found in JSONMemoryStore.",
    )