Memory

Short-term memory, conversation stores, vector stores, and user memory.

Short-term memory

ShortTermMemory

class ShortTermMemory(max_tokens: int = 40000)

Sliding-window conversation buffer for a single agent run.

Stores the ordered message history and automatically trims to fit within a token budget when requested. Uses the heuristic chars / 4 ≈ tokens when no token-counting transport is available.

Example:

memory = ShortTermMemory(max_tokens=8000)
memory.add_user("Hello, how are you?")
memory.add_assistant(completion)
msgs = memory.messages()  # trimmed to budget

Parameters:

ShortTermMemory.add_user

def add_user(self, content: str | list[Any]) -> None

Append a user message to the buffer.

Parameters:

ShortTermMemory.add_assistant

def add_assistant(self, completion: Any) -> None

Append an assistant completion to the buffer.

Accepts a Completion dataclass (with .content and .tool_calls attributes) or a plain dict.

Parameters:

ShortTermMemory.add_tool_result

def add_tool_result(self, result: Any) -> None

Append a tool result message to the buffer.

Accepts a ToolResult dataclass or a plain dict.

Parameters:

ShortTermMemory.set_summary

def set_summary(self, text: str) -> None

Store text as the conversation summary.

Called by the runner after a summarisation LLM call completes. The summary is persisted via snapshot() / restore() so resumed sessions carry it forward.

Parameters:

ShortTermMemory.messages_to_summarize

def messages_to_summarize(self, keep_recent: int = 6) -> list[Any]

Return the slice of messages that should be compressed.

Returns the oldest (total - keep_recent) non-system messages. System messages are excluded because they are already managed separately (they are never dropped by messages() either).

Parameters:

Returns: list[Any] — List of messages to feed to the summarisation LLM call.

ShortTermMemory.trim_to_recent

def trim_to_recent(self, keep_recent: int = 6) -> None

Drop all but the most-recent keep_recent non-system messages.

Called by the runner after the summarisation call so the buffer only holds recent turns while the older context lives in self._summary.

Parameters:

ShortTermMemory.messages

def messages(self) -> list[Any]

Return the current message list, trimmed to fit the token window.

The trim is applied in-place on a copy; the internal buffer is NOT modified. Call trim_to_fit() explicitly to mutate the buffer.

Returns: list[Message] — Ordered list of messages within the token budget.

ShortTermMemory.trim_to_fit

def trim_to_fit(self, max_tokens: int) -> None

Drop oldest non-system messages until the token estimate fits.

Unlike messages() this mutates the internal buffer.

Parameters:

ShortTermMemory.clear

def clear(self) -> None

Clear all messages from the buffer.

Returns: None — None

ShortTermMemory.snapshot

def snapshot(self) -> Any

Return a deep copy of the current memory state.

The returned object includes both the message list and the conversation summary (if any). It is independent of the internal buffer — mutations do not affect the memory.

The format is a dict with "messages" and "summary" keys so that resumed sessions carry the summary forward. Old snapshots that are plain list objects are still accepted by restore() for backward compatibility.

Returns: dict[str, Any] — Snapshot dict {"messages": [...], "summary": str | None}.

ShortTermMemory.restore

def restore(self, data: Any) -> None

Restore the memory buffer from a snapshot.

Accepts both the new dict snapshot format ({"messages": [...], "summary": ...}) and the legacy plain list format produced by older versions of snapshot().

Parameters:

Conversation store

ConversationStore

class ConversationStore

Protocol for persisting and retrieving full conversation histories.

Keyed by an arbitrary string conversation_id (typically a session or user identifier).

ConversationStore.load

def load(self, conversation_id: str) -> list[Any]

Load the message history for conversation_id.

Parameters:

Returns: list[Message] — Ordered list of Message objects (empty list when not found).

ConversationStore.save

def save(self, conversation_id: str, messages: list[Any]) -> None

Persist the message history for conversation_id.

Overwrites any existing history for that ID.

Parameters:

ConversationStore.delete

def delete(self, conversation_id: str) -> None

Delete the history for conversation_id.

Parameters:

InMemoryConversationStore

class InMemoryConversationStore()

In-memory store for full conversation histories.

Implements the ConversationStore protocol. Each conversation is keyed by an arbitrary string identifier (typically a user ID or session UUID). Deep copies are used on both load and save so that the caller cannot inadvertently mutate stored data.

Example:

store = InMemoryConversationStore()
await store.save("session-abc", messages)
loaded = await store.load("session-abc")

InMemoryConversationStore.load

def load(self, conversation_id: str) -> Any

Load the conversation snapshot for conversation_id.

Returns an empty list when the conversation does not exist (backward compat — callers that check if prior: still work on empty lists).

Parameters:

Returns: dict[str, Any] | list[Any] — A deep copy of the stored snapshot. When the snapshot was created by ShortTermMemory.snapshot() this is a {"messages": [...], "summary": ...} dict; for legacy plain lists the raw list is returned.

InMemoryConversationStore.save

def save(self, conversation_id: str, snapshot: Any) -> None

Persist the conversation snapshot for conversation_id.

Overwrites any existing entry for that identifier. A deep copy is stored to prevent the caller from mutating the stored data.

Accepts both the new dict snapshot format ({"messages": [...], "summary": ...}) produced by ShortTermMemory.snapshot() and the legacy plain list[Message] format so that code written against the old API continues to work. Plain lists are automatically normalised to the dict format so that load() always returns a consistent shape.

Parameters:

InMemoryConversationStore.delete

def delete(self, conversation_id: str) -> None

Delete the history for conversation_id.

Silently does nothing when the conversation does not exist.

Parameters:

InMemoryConversationStore.list_conversations

def list_conversations(self) -> list[str]

Return a sorted list of all stored conversation identifiers.

Returns: list[str] — Sorted list of conversation IDs.

InMemoryConversationStore.clear

def clear(self) -> None

Remove all stored conversation histories.

Returns: None — None

Vector store

InMemoryVectorStore

class InMemoryVectorStore()

In-memory vector store using TF-IDF cosine similarity.

Implements the MemoryStore protocol. Suitable for development and testing; no external dependencies required.

Example:

store = InMemoryVectorStore()
doc_id = await store.upsert("The quick brown fox", metadata={"tag": "test"})
results = await store.search("quick fox", k=3)

InMemoryVectorStore.upsert

def upsert(self, content: str, id: str | None = None, metadata: dict[str, Any] | None = None, embedding: list[float] | None = None) -> str

Insert or update a document.

Parameters:

Returns: str — The document's identifier.

InMemoryVectorStore.search

def search(self, query: str, k: int = 5, filter: dict[str, Any] | None = None) -> list[MemoryResult]

Search for documents semantically similar to query.

Parameters:

Returns: list[MemoryResult] — Up to k results ordered by descending cosine similarity.

InMemoryVectorStore.get

def get(self, id: str) -> MemoryResult | None

Retrieve a document by its identifier.

Parameters:

Returns: MemoryResult | None — The MemoryResult, or None when not found.

InMemoryVectorStore.delete

def delete(self, ids: list[str]) -> None

Delete documents by their identifiers.

Parameters:

InMemoryVectorStore.clear

def clear(self) -> None

Remove all documents from the store.

Returns: None — None

User memory

MemoryFact

A single persisted fact about a user.

Facts are stored with a confidence score [0.0–1.0] that can decay over time if not reinforced.

UserMemoryStore

Protocol for user-level persistent memory stores.

InMemoryUserMemoryStore

In-process UserMemoryStore for testing and development.

Uses simple substring matching for search (no vector similarity).

@remember decorator

remember

Opt a @agent() class into automatic user memory extraction/injection.

Must be applied BELOW @agent():

@agent(model="claude-haiku-4-5")
@remember(store="user_memory", extract=True, inject=True, top_k=5)
class PersonalAssistant: ...

When inject=True, relevant memories are prepended to the system prompt before each LLM call.

When extract=True, new facts are extracted from each conversation turn and stored in the UserMemoryStore.

Parameters: