Skip to main content

Episodic Memory for Multi-Step Agents

A stateless agent that restarts a task from scratch every session wastes context. A long-running agent that can write its observations and decisions to a graph database — and read them back on resume — can pick up where it left off, avoid repeating work, and build compound reasoning over time.

This tutorial shows the episodic memory pattern: one GOAL record per task, linked STEP records capturing each intermediate action, and linked OBSERVATION records for tool outputs and discoveries.

Graph shape​

LabelWhat it represents
GOALThe agent's top-level objective for a session
STEPA single action the agent took (tool call, reasoning step, decision)
OBSERVATIONThe result or output of a step (tool output, search result, partial answer)

Step 1: Create a goal record when a session starts​

from rushdb import RushDB
import os
from datetime import datetime, timezone

db = RushDB(os.environ["RUSHDB_API_KEY"], base_url="https://api.rushdb.com/api/v1")


def start_session(task: str, agent_id: str):
    goal = db.records.create("GOAL", {
        "task": task,
        "agentId": agent_id,
        "status": "in_progress",
        "startedAt": datetime.now(timezone.utc).isoformat()
    })
    return goal


goal = start_session(
    "Research RushDB graph query patterns and summarize the top 5",
    "agent-001"
)
print("Session goal ID:", goal.id)

Step 2: Record each step as the agent executes it​

Each agent action — tool call, web search, or LLM reasoning step — becomes a STEP record linked to the GOAL.

def record_step(goal_id: str, step_type: str, action: str, index: int):
    step = db.records.create("STEP", {
        "type": step_type,
        "action": action,
        "index": index,
        "status": "executing",
        "startedAt": datetime.now(timezone.utc).isoformat()
    })

    goal_result = db.records.find({"labels": ["GOAL"], "where": {"__id": goal_id}})
    db.records.attach(
        goal_result.data[0].id,
        step.id,
        {"type": "HAS_STEP", "direction": "out"}
    )
    return step


step1 = record_step(goal.id, "tool_call", "Search docs: graph traversal patterns", 1)

Step 3: Store observations from tool outputs​

When a tool returns results, store the output as an OBSERVATION linked to the STEP that produced it.

def record_observation(step_id: str, content: str, obs_type: str):
    obs = db.records.create("OBSERVATION", {
        "content": content,
        "type": obs_type,
        "recordedAt": datetime.now(timezone.utc).isoformat()
    })

    step_result = db.records.find({"labels": ["STEP"], "where": {"__id": step_id}})
    db.records.attach(
        step_result.data[0].id,
        obs.id,
        {"type": "PRODUCED", "direction": "out"}
    )

    db.records.update(step_id, {"status": "completed"})
    return obs


obs1 = record_observation(
    step1.id,
    "Found 3 relevant tutorials: thinking-in-graphs, modeling-hierarchies, temporal-graphs",
    "search_result"
)

Step 4: Resume a session and retrieve accumulated context​

When the agent resumes — or when a new session needs to continue where a prior one left off — retrieve the full context for the GOAL.

def resume_session(goal_id: str) -> dict:
    goal_result = db.records.find({"labels": ["GOAL"], "where": {"__id": goal_id}})
    goal_rec = goal_result.data[0]

    steps = db.records.find({
        "labels": ["STEP"],
        "where": {
            "GOAL": {
                "$relation": {"type": "HAS_STEP", "direction": "in"},
                "__id": goal_id
            }
        },
        "orderBy": {"index": "asc"}
    })

    observations = db.records.find({
        "labels": ["OBSERVATION"],
        "where": {
            "STEP": {
                "$relation": {"type": "PRODUCED", "direction": "in"},
                "GOAL": {
                    "$relation": {"type": "HAS_STEP", "direction": "in"},
                    "__id": goal_id
                }
            }
        }
    })

    completed = sum(1 for s in steps.data if s.data.get("status") == "completed")
    return {
        "goal": {"task": goal_rec.data.get("task"), "status": goal_rec.data.get("status")},
        "completedSteps": completed,
        "totalSteps": steps.total,
        "observations": [o.data.get("content") for o in observations.data]
    }


context = resume_session(goal.id)
print(context)

Step 5: Mark the goal complete​

When the task finishes, update the GOAL record with its outcome.

from datetime import datetime, timezone

db.records.update(goal.id, {
    "status": "completed",
    "completedAt": datetime.now(timezone.utc).isoformat(),
    "summary": "Identified 5 top query patterns"
})

When to use episodic memory vs. team memory​

PatternUse case
Episodic memory (this tutorial)Per-session agent context: goals, steps, tool outputs
Team memory (Building Team Memory)Shared persistent knowledge: tickets, docs, decisions
Fact memory (RushDB as a Memory Layer)Long-lived facts about entities across all sessions

These patterns compose: an agent can write episodic steps to its own GOAL graph while also reading from shared team memory for background knowledge.

Production caveat​

Episodic records accumulate quickly during active agent use. Define a cleanup policy: archive GOAL records after N days, compress older OBSERVATION content (store summaries instead of full tool outputs), or use a separate RushDB project for ephemeral agent memory versus persistent team knowledge.

Next steps​