Episodic memory for agents

An agent without a memory starts over every time. Episodic memory = saved events: «2026-09-14: the user asked for help with gradient descent, got stuck on the learning rate, solved it after hint 2».

The design:

• What: a summary per session (not the raw text), with the time, the subject, the outcome, the emotional tone where relevant. Written by the model with a fixed template.
• How it is fetched: embeddings + a time weight + a subject filter → the top k episodes into the system prompt as «Earlier:».
• When: at the start of a session (the most recent + the most relevant) and at a change of subject.
• Measure: A/B with and without the memory on simulated users; probe questions; the share of answers where the model refers to an episode correctly.

The risk is not that the memory is missing — it is that it is used wrongly: an episode from last month is interpreted as the present, or two users are mixed up. Every episode carries a user_id, a date and a «valid until».

Privacy: the user should be able to see, delete and switch off. Do not save more than the purpose requires (data minimisation), and set a retention.

import time, json

TEMPLATE = "Summarise the session in ≤ 60 words: the subject, what the user managed, where they got stuck, preferences that were expressed. No sensitive information."

class EpisodicMemory:
    def __init__(self, store, embed, llm):
        self.store, self.embed, self.llm = store, embed, llm     # store: a vector DB with payload filters

    def write(self, user_id, transcript, subject):
        text = self.llm(TEMPLATE + "\n\n" + transcript, temperature=0)
        self.store.upsert(vector=self.embed(text), payload={"user": user_id, "t": time.time(), "subject": subject, "text": text})

    def fetch(self, user_id, query, k=4, half_life_d=30):
        cand = self.store.search(self.embed(query), k=20, filter={"user": user_id})
        now = time.time()
        def score(c):
            age_d = (now - c.payload["t"]) / 86400
            return c.score * 0.5 ** (age_d / half_life_d)          # the newer weighs more
        top = sorted(cand, key=score, reverse=True)[:k]
        return "\n".join(f"- [{time.strftime('%Y-%m-%d', time.localtime(c.payload['t']))}] {c.payload['text']}" for c in top)

    def delete(self, user_id):
        self.store.delete(filter={"user": user_id})

The system prompt: «Earlier sessions (they may be out of date — confirm where needed):\n{episodes}».