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/json_store.py 
 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
141
142
143
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",
    ) -> 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"``.
        """
        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,
        embedded_text: str | None = None,
    ) -> None:
        """Persist an episode to the store.

        Appends one JSON line to the backing file. Does not rewrite existing
        content. ``embedded_text`` is accepted for interface compatibility
        but ignored — this store does not embed episodes.

        Args:
            episode (Episode): The completed episode to store.
            embedded_text (str | None): Ignored.
        """
        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)

    async def search(
        self,
        query: str,
        k: int,
        **kwargs: Any,
    ) -> list[Episode]:
        """Not implemented — similarity search is deferred to vector store.

        Args:
            query (str): The search query.
            k (int): Maximum number of episodes to return.
            **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')

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'
Source code in src/llm_agents_from_scratch/memory/json_store.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    dir: Path,
    filename: str = "episodes.jsonl",
) -> 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"``.
    """
    self.path = dir / filename
    self._episodes: list[Episode] = self._load()

write async 

write(episode, embedded_text=None)

Persist an episode to the store.

Appends one JSON line to the backing file. Does not rewrite existing content. embedded_text is accepted for interface compatibility but ignored — this store does not embed episodes.

Parameters:

Name Type Description Default
episode Episode

The completed episode to store.

 required
embedded_text str | None

Ignored.

 None
Source code in src/llm_agents_from_scratch/memory/json_store.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def write(
    self,
    episode: Episode,
    embedded_text: str | None = None,
) -> None:
    """Persist an episode to the store.

    Appends one JSON line to the backing file. Does not rewrite existing
    content. ``embedded_text`` is accepted for interface compatibility
    but ignored — this store does not embed episodes.

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

read_recent async 

read_recent(n)

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.

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/memory/json_store.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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]

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/json_store.py 
87
88
89
90
91
92
93
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/json_store.py 
 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
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)

search async 

search(query, k, **kwargs)

Not implemented — similarity search is deferred to vector store.

Parameters:

Name Type Description Default
query str

The search query.

 required
k int

Maximum number of episodes to return.

 required
**kwargs Any

Ignored.

 {}

Raises:

Type Description
NotImplementedError

Always. Use a vector-backed store for similarity search.
Source code in src/llm_agents_from_scratch/memory/json_store.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def search(
    self,
    query: str,
    k: int,
    **kwargs: Any,
) -> list[Episode]:
    """Not implemented — similarity search is deferred to vector store.

    Args:
        query (str): The search query.
        k (int): Maximum number of episodes to return.
        **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.",
    )