Skip to content

aaa.memory

Protocol and types

Memory

Bases: Protocol

Long-term memory. Implementations must be safe to call concurrently.

add async 

add(
    entries: Sequence[MemoryEntry],
    scope: Scope,
    *,
    key: str | None = None,
) -> AddResult

Remember entries for scope. key is an idempotency key: repeating a call with the same key must not store the same thing twice (backends that cannot honour that set idempotent = False, and AAA then refuses to retry their writes automatically).

search async 

search(
    query: str,
    scope: Scope,
    *,
    k: int = 5,
    filters: dict[str, Any] | None = None,
    as_of: datetime | None = None,
) -> list[MemoryItem]

The k memories of scope most relevant to query, best first. as_of asks for what was true at that moment (only meaningful on temporal backends; others ignore it).

Scope

Bases: BaseModel

Whose memory. All fields optional; the more that are set, the narrower the scope.

tenant is the isolation boundary; user and agent partition inside it; session is a single conversation or run; namespace is free-form extra partitioning (e.g. a project).

key 

key(*, include_session: bool = False) -> str

A stable string for this scope, safe to use as a group id / container tag / key prefix. The session is excluded by default: long-term memory outlives the conversation that made it.

slug 

slug(*, max_len: int | None = None) -> str

An identifier made only of [A-Za-z0-9_-], unique per (tenant, user, agent, namespace), for backends that restrict ids (Graphiti group ids, Supermemory container tags, LangGraph namespace labels, Mem0 entity ids). Parts are ordered tenant, user, agent, namespace, so a tenant's slugs all start with tenant_prefix(). Long slugs are shortened with a hash tail.

tenant_prefix 

tenant_prefix() -> str

The slug prefix shared by everything in this scope's tenant.

contains 

contains(other: Scope) -> bool

True if other is this scope or narrower (so forgetting self removes other's data). The tenant must match exactly (an untenanted scope never reaches into a tenant's data).

MemoryEntry

Bases: BaseModel

One thing to remember: a statement, or a conversation turn (role set).

MemoryItem

Bases: BaseModel

One recalled memory.

AddResult

Bases: BaseModel

Capabilities

Updatable

Bases: Protocol

Deletable

Bases: Protocol

delete async 

delete(ids: Sequence[str], scope: Scope) -> int

Delete memories by id; returns how many were removed.

Forgettable

Bases: Protocol

forget async 

forget(scope: Scope) -> int

Delete everything stored for scope and anything narrower (right to erasure).

Profiled

Bases: Protocol

profile async 

profile(scope: Scope) -> dict[str, Any]

A synthesized summary of the scope (stable facts, recent context...).

Listable

Bases: Protocol

require 

require(memory: object, *capabilities: type) -> None

Raise MissingMemoryCapability naming everything memory lacks (fail at startup, not mid-run).

Runtime

MemoryHandler 

MemoryHandler(
    memory: Memory,
    *,
    scope_for: Callable[[Event], Scope] = default_scope,
    write_back: WriteBack | None = None,
    delivery: Delivery | None = None,
    retry: RetryPolicy | None = None,
    concurrency: int = 8,
    timeout_s: float = 20.0,
)

WriteBack dataclass 

WriteBack(
    build: EntryBuilder = default_entries,
    only_success: bool = True,
)

Remember what happened when a run finishes. build(started, completed) chooses what; the default stores the task and the answer. Only successful runs are written unless only_success=False.

memory_tools 

memory_tools(
    memory: Memory,
    *,
    k: int = 5,
    scope_for: Callable[
        [ToolContext], Scope
    ] = scope_from_tool_context,
    allow_forget: bool = False,
) -> list[Tool]

remember(text), recall(query) and (opt-in) forget(memory_id) tools bound to memory.

RecallConfig dataclass 

RecallConfig(
    k: int = 5,
    query: Callable[[dict[str, Any]], str] = _default_query,
    max_chars: int = 2000,
    as_of: str | None = None,
)

Look up long-term memory before the first model call and put it in the prompt.

Needs a MemoryHandler attached to the engine (without one, recall is skipped with a warning). A failing or slow memory backend never fails the run: the loop simply continues without memories.

Backends

NativeMemory 

NativeMemory(
    index: Index | None = None,
    *,
    embedder: Embedder | None = None,
    extractor: Extractor | None = None,
)

llm_extractor 

llm_extractor(
    client: LLMClient, model: str | None = None
) -> Extractor

An extractor that distils a conversation into facts with one LLM call.

Note this calls client directly: it is not routed through the engine's LLMHandler, so its tokens are not in the run's budget ledger. If that matters, do the extraction as an ordinary step of your algorithm and pass the resulting facts to add.

CallableMemory 

CallableMemory(
    add: Callable[..., Any],
    search: Callable[..., Any],
    *,
    update: Callable[..., Any] | None = None,
    delete: Callable[..., Any] | None = None,
    forget: Callable[..., Any] | None = None,
    profile: Callable[..., Any] | None = None,
    list: Callable[..., Any] | None = None,
    aclose: Callable[[], Awaitable[None] | None]
    | None = None,
    idempotent: bool = False,
)

FakeMemory 

FakeMemory()

Scoped, idempotent, fully featured. calls records every operation for assertions.

Mem0Memory 

Mem0Memory(
    client: Any,
    *,
    infer: bool = True,
    threshold: float | None = None,
)

GraphitiMemory 

GraphitiMemory(
    graphiti: Any,
    *,
    source_description: str = "aaa",
    current_only: bool = True,
)

SupermemoryMemory 

SupermemoryMemory(
    api_key: str,
    *,
    base_url: str = "https://api.supermemory.ai",
    ingest: Literal["documents", "memories"] = "documents",
    search_mode: Literal[
        "memories", "hybrid", "documents"
    ] = "memories",
    wait: bool = False,
    wait_timeout_s: float = 60.0,
    threshold: float = 0.0,
    client: AsyncClient | None = None,
)

LangGraphStoreMemory 

LangGraphStoreMemory(store: Any, *, root: str = 'memories')

namespace_for 

namespace_for(scope: Scope) -> tuple[str, ...]

The store namespace holding exactly this scope's memories (also what LangMem should use).