istar-memory 0.1.1

Generic multi-layer agent memory for LLM simulation loops (L0→L1→L2→L3)

istar

Generic multi-layer agent memory for LLM simulation loops.

Istar (singular) or Istari (plural) is the Quenya term for Wizards, which means "those who know", referring to five angelic beings known as Maiar sent to Middle-earth by the Valar to aid the Free Peoples against Sauron.

Istar borrows the L* tier hierarchy (L0 → L3) from TencentDB-Agent-Memory, which pioneered layered abstraction for agent memory.

---

LLM agents running over many turns need structured memory — not a raw transcript dump. istar provides a four-layer pipeline that compresses turn-by-turn data into increasingly abstract representations: raw records → discrete facts → strategic narratives → cross-run lessons.

The pipeline is domain-agnostic. You implement three small protocol classes that inject your domain's logic; istar handles the lifecycle, throttling, LLM calls, and persistence.

---

Architecture

┌──────────────────────────────────────────────────────────────┐
│  L3  Run Profile      Markdown file — persists across runs   │
│       ↑ one LLM call at run end                              │
├──────────────────────────────────────────────────────────────┤
│  L2  Episodes         ~150-token narrative — updated every   │
│                       25 turns by one LLM call               │
│       ↑ synthesised from L1 atoms + domain context           │
├──────────────────────────────────────────────────────────────┤
│  L1  Atoms            Timestamped facts — extracted every    │
│                       10 turns by deterministic rules        │
│       ↑ derived from L0 records, no LLM involved            │
├──────────────────────────────────────────────────────────────┤
│  L0  Turn Records     In-memory per-turn data                │
│                       (TurnRecord — normalised from your     │
│                        simulator's native response)          │
└──────────────────────────────────────────────────────────────┘

Only L2 and L3 involve LLM calls. L1 extraction is deterministic — no inference, no cost.

---

Installation

pip install istar-memory

---

Quick start

Implement three protocol classes for your domain, then wire them in:

from istar import AgentMemory, TurnRecord, L1Atom
from istar.models import EpisodePromptContext, RetroPromptContext

class MyAtomExtractor:
    def extract(self, window: list[TurnRecord], existing: list[L1Atom]) -> list[L1Atom]:
        atoms = []
        for i in range(1, len(window)):
            prev, curr = window[i - 1], window[i]
            # detect inflection points in your domain
            if prev.reward > 0 and curr.reward < 0:
                atoms.append(L1Atom(
                    id=f"loss-start-{curr.turn}",
                    turn=curr.turn,
                    type="market_event",
                    content=f"T{curr.turn}: Turned unprofitable (reward {curr.reward:.1f}).",
                    priority=85,
                    tags=["loss_start"],
                ))
        return atoms

class MyEpisodeClassifier:
    def build_context(self, recent_turns: list[TurnRecord], prev_turn: int) -> EpisodePromptContext:
        avg = sum(r.reward for r in recent_turns) / len(recent_turns) if recent_turns else 0
        return EpisodePromptContext(
            phase_names=["GROWTH", "PLATEAU", "DECLINE", "RECOVERY"],
            domain_preamble="You are tracking strategic phases for a trading agent.",
            context_summary=f"Average reward over last {len(recent_turns)} turns: {avg:.2f}.",
            phase_guidance="Classify based on the reward trend, not individual events.",
        )

class MyRetroClassifier:
    def build_context(self, all_turns: list[TurnRecord], episodes) -> RetroPromptContext:
        total = sum(r.reward for r in all_turns)
        return RetroPromptContext(
            domain_preamble="You are writing a retrospective for a trading agent.",
            outcome_summary=f"Total reward: {total:.1f} over {len(all_turns)} turns.",
            outcome_metric_name="reward",
        )

# Construct memory for one agent
memory = AgentMemory(
    atom_extractor=MyAtomExtractor(),
    episode_classifier=MyEpisodeClassifier(),
    retro_classifier=MyRetroClassifier(),
)

# Every turn
memory.record_turn(TurnRecord(
    turn=1, reward=42.0, reward_unit="pts",
    events=["price_spike"], actions_taken=[{"type": "buy", "qty": 10}],
    extra={"price": 105.0, "volatility": 0.3},
))
memory.extract_atoms(turn=1)  # self-throttles to every 10 turns

# Every 25 turns (async)
episode = await memory.update_episode(router, turn=25)

# Inject into agent observation
block = memory.memory_block()
# → {"current_phase": "[GROWTH | T1-T25]\n...", "key_observations": [...]}

# End of run (async)
await memory.summarise_run(router, profile_id="my-agent", profiles_dir=Path("profiles/"))

---

Domain injection protocols

Three protocol classes are the only domain-specific code you write:

AtomExtractor

Scans a window of TurnRecord objects and emits L1Atom objects for detected inflections. Called every 10 turns. No LLM. Should be fast and deterministic.

class AtomExtractor(Protocol):
    def extract(self, window: list[TurnRecord], existing_atoms: list[L1Atom]) -> list[L1Atom]: ...

Atom priorities (0–100) control eviction when the in-memory ring fills (default cap: 20 atoms). Higher priority atoms survive longer. Use ~80–90 for critical events, ~40–60 for routine ones.

EpisodeClassifier

Builds the prompt context for L2 episode synthesis. Called every 25 turns, just before the LLM call.

class EpisodeClassifier(Protocol):
    def build_context(self, recent_turns: list[TurnRecord], prev_episode_end_turn: int) -> EpisodePromptContext: ...

EpisodePromptContext carries: phase_names (your taxonomy), domain_preamble (who the LLM is), context_summary (domain metrics), phase_guidance (classification hints).

RetroClassifier

Builds the prompt context for the L3 end-of-run retrospective. Called once per run.

class RetroClassifier(Protocol):
    def build_context(self, all_turns: list[TurnRecord], episodes: list[L2Episode]) -> RetroPromptContext: ...

---

Data models

@dataclass
class TurnRecord:
    turn: int
    reward: float           # scalar outcome this turn
    reward_unit: str        # "EUR", "points", etc. — used in prompts
    events: list[str]       # named events that fired
    actions_taken: list[dict]
    extra: dict             # domain-specific fields; only your AtomExtractor reads these

@dataclass
class L1Atom:
    id: str                 # stable identifier — MD5 of (turn, tag)
    turn: int
    type: str               # "market_event" | "coordination_event" | "portfolio_event"
    content: str            # human-readable — injected into prompts
    priority: int           # 0–100; drives eviction order
    tags: list[str]         # machine-readable labels

@dataclass
class L2Episode:
    name: str               # phase name from your EpisodeClassifier
    turn_start: int
    turn_end: int
    narrative: str          # LLM-written description
    heat: int               # times injected into agent observations via memory_block()

---

AgentMemory reference

Method	When to call	LLM?
record_turn(record)	Every turn	No
extract_atoms(turn)	Every turn — self-throttles to every 10	No
update_episode(router, turn)	Every turn — self-throttles to every 25	Yes, 1 call
memory_block()	Every turn, to build agent observation	No
summarise_run(router, profile_id, profiles_dir)	End of run	Yes, 1 call

Constants (override by subclassing):

Constant	Default	Meaning
EXTRACT_EVERY	10	Atom extraction cadence (turns)
EPISODE_EVERY	25	Episode synthesis cadence (turns)
MAX_ATOMS	20	In-memory atom ring buffer size
MAX_EPISODES	5	Max episodes kept in memory

---

L3 cross-run learning

At the end of each run, summarise_run() writes a Markdown retrospective to profiles_dir/{profile_id}.md. To inject it into the next run's system prompt:

from istar.memory import load_prior_run_notes

notes = load_prior_run_notes("my-agent", profiles_dir=Path("profiles/"))
system_prompt = base_persona + (f"\n\n{notes}" if notes else "")

---

Reference implementation

The examples/energy_tycoon/ directory contains a complete, runnable implementation against a competitive multi-agent energy market simulator:

File	What it shows
examples/energy_tycoon/adapter.py	All three protocols — EnergyAtomExtractor, EnergyEpisodeClassifier, EnergyRetroClassifier — plus make_turn_record translating the simulator API to TurnRecord
examples/energy_tycoon/memory.py	AgentMemory subclass that pre-wires the adapters, overrides EXTRACT_EVERY = 3, and adapts the call signatures to the simulator's native API

---

License

MIT