agentkit.state.memory

The MemorySource Protocol plus the shipped implementations — CompositeMemory, SequentialMemory, VectorMemory — and the ScopedMemory / CompactedMemory / CachedMemory decorators.

agentkit.state.memory — the unified memory/RAG seam.

One Protocol (MemorySource), one composite (CompositeMemory), many backends. An agent declares Agent.memory: MemorySource | None; the cognition decides when to query and how to fold results into the working context.

First-party sources and decorators live in this package; concrete adapters land alongside as they're written:

• VectorMemory — vector-store backed (the canonical RAG case).
• FileMemory — file-system backed (a memory tool's read side).
• JournalMemory — wraps MutationJournal as a queryable source.
• ToolMemory — adapts any Tool into a MemorySource.
• ScopedMemory — Decorator enforcing ctx.scope at the boundary.
• CompactedMemory— Decorator summarising results via a Compactor.
• CachedMemory — Decorator caching hot queries with a TTL.
• ReadOnlyMemory — Decorator refusing (or dropping) writes, for a source that is read-only by policy rather than by backend.

DedupeMode module-attribute

DedupeMode = Literal['id', 'content']

How CompositeMemory decides two items are the same fact.

"id" merges items sharing a non-None MemoryItem.id OR identical content; "content" merges on content alone. None — the third value of the parameter, not of this alias — is the old pure concatenation.

OnWritePolicy module-attribute

OnWritePolicy = Literal['refuse', 'ignore']

What a ReadOnlyMemory does with a write. Closed Literal so a typo is a type error at the call site, not a runtime surprise; the tuple below re-derives the members with get_args so the runtime check and the type can never drift apart.

MemoryItem dataclass

MemoryItem(
    content: str,
    source: str,
    score: float | None = None,
    metadata: dict[str, Any] = dict(),
    *,
    id: str | None = None,
)

The uniform result shape every MemorySource returns.

content is what the agent reads (already-formatted text). source names the backend that produced it (the MemorySource.name) so consumers can attribute results and cognitions can format citations. id is the backend's own identifier for the RECORD — a vector store's chunk id, a journal row key. Stable within a source; not assumed comparable across sources unless the caller knows they share a keyspace (which is the usual case here, because the journal is normally what the vector store was built from). It exists so CompositeMemory can tell that two sources returned the SAME fact rather than two facts: without it the merge was pure concatenation, a reranker scored both copies, and the top-k the model saw was one fact occupying two slots. None is the honest answer for a backend that has no stable key (a tool-wrapped search, a scratchpad), and dedupe falls back to a content digest for those. score is the backend's relevance signal (cosine similarity for vector search, recency for journal, etc.); None when the backend doesn't rank. metadata carries backend-specific extras (chunk id, file path, timestamp) that the cognition or the rendering layer may consult. It is FROZEN at construction (see __post_init__) — still a dict for every consumer, but a backend that wants to annotate an item builds a new one: dataclasses.replace(item, metadata={**item.metadata, "path": p}).

id is KEYWORD-ONLY, and that is load-bearing rather than stylistic. This type is constructed positionally in tens of places (MemoryItem("c", "vector", 0.91, meta)), so declaring id third — where it reads most naturally, next to source — would have rebound 0.91 to id and meta to score at every one of them: no error, wrong data, silently. Appending it after metadata would have fixed today and left the same trap for the next field; kw_only closes it permanently and lets the declaration keep the order a reader wants.

MemorySource

Bases: Protocol

A queryable source of information the agent can reach for.

name is the stable diagnostic label — stamped on returned MemoryItem.source, also used for trace attribution. Implementations SHOULD set a sensible class-level default.

Implementations MUST honour ctx.scope if the underlying backend crosses tenants. Multi-tenant enforcement is the implementation's job OR a ScopedMemory decorator's job; the Protocol doesn't enforce it itself.

Reranker

Bases: Protocol

Optional second pass over a set of merged MemoryItems.

A composite memory may pull from many sources, each returning items scored on its own scale (vector cosine vs lexical match vs recency). A Reranker scores them on a common axis so the top-k cut is sensible.

The default reranker is score_sort_rerank — sort by score desc with None last. Real applications wire a cross-encoder or an LLM judge here.

CompositeMemory dataclass

CompositeMemory(
    sources: list[MemorySource],
    reranker: Reranker | None = None,
    name: str = "composite",
    dedupe: DedupeMode | None = "id",
)

Parallel fan-out across many sources.

Each query runs against every source concurrently. Results are merged, then optionally reranked. Top-k is taken from the reranked list. The default reranker is score_sort_rerank (sort by score desc); pass a cross-encoder or LLM judge for richer semantics.

Backpressure: asyncio.gather is bounded by Python's default semaphore behaviour — for very wide trees, prefer nesting CompositeMemory(CompositeMemory(...), ...) to keep fanout moderate at each level.

dedupe decides what happens when two sources hold the SAME fact, which in this composition is the normal case rather than an accident: ask a vector store and a journal the same question and they overlap, because the journal is usually what the store was built from. The merge used to be [item for batch in batches for item in batch] — pure concatenation — so a reranker scored both copies and the top-k the model saw was one fact occupying two slots. That is a WRONG answer, not a missing feature, which is why "id" is the DEFAULT and None is the opt-out a caller chooses rather than inherits:

• "id" — merge on a shared MemoryItem.id OR identical content. A union rather than "id, else content" because a per-item fallback would mean that populating id on a backend stops it matching backends that have none — the headline case would regress the moment VectorMemory started reporting its chunk ids. See _merge_duplicates.
• "content" — content only, for stores whose ids are not comparable (two stores can both call their first row "1", and trusting that merges two unrelated facts, which deletes one). The single behavioural difference from "id": a shared id with different text.
• None — the old concatenation, kept for callers who genuinely want every copy (a diversity reranker, a provenance audit).

A survivor of an actual collision is stamped with dedupe_sources (the distinct source names that produced a copy) and dedupe_count. Items that never collided come back exactly as their backend returned them.

Dedupe runs BEFORE rerank, which is the point of it: a cross-encoder scoring the same passage twice pays double and then still spends two top-k slots on it. Cost is one sha256 per id-less item (~1 µs for a 1 KB passage) against the network round trips that produced the pool.

accepts_writes property

accepts_writes: bool

True when at least one member can actually commit.

A composite is itself a MemorySource and composites nest, so without this the marker dies at the first composite boundary: an all-read-only CompositeMemory nested in an outer fan-out returns normally from write having stored nothing, and the outer split files it under accepted — the exact lie the refused bucket exists to prevent, one level up.

any rather than all because the bucket asks "did anything land here?", and a composite with one writable member did commit. An empty composite commits nothing, and any(()) is False, which is the honest answer.

write async

write(items: Iterable[MemoryItem], *, ctx: Ctx) -> None

Broadcast writes to every source. A backend that can't accept writes implements write as a no-op (see ToolMemory) and sets accepts_writes = False so the split can tell it apart from a backend that committed.

Partial failures surface as :class:CompositeWriteError with a per-source accepted/refused/failed split. Without return_exceptions=True the first child exception would propagate and the rest would be discarded, so a caller seeing a failure would have no way to know which backends had already committed.

CompositeWriteError

CompositeWriteError(
    *,
    accepted: list[str],
    failed: dict[str, BaseException],
    refused: list[str] | None = None,
)

Bases: RuntimeError

At least one source in a CompositeMemory.write fan-out failed.

Carries a structured per-source split so callers can compensate (retry the failed sources, or surface the partial-commit state to the operator). Raising the first child exception verbatim would discard everything else and postmortems would lose the information about which backends DID commit.

refused is a THIRD bucket, not a slice of accepted. A source that declares accepts_writes = False (ReadOnlyMemory with on_write="ignore") returns from write without raising and without storing anything, so the two-bucket split had exactly one place to put it and that place was a lie: accepted is what an operator reads to decide which backends NOT to replay after a partial commit, and replaying is skipped for a source that never committed. Keyword-only with a default so the constructor stays backward-compatible for callers that build the error themselves.

Attributes:

Name	Type	Description
accepted		list of source names whose write succeeded.
refused		list of source names that declared themselves read-only and dropped the write. Not an error — the fan-out still succeeds when nothing failed — but never conflated with accepted.
failed		dict mapping source name → BaseException raised by that source.

SequentialMemory dataclass

SequentialMemory(
    sources: list[MemorySource], name: str = "sequential"
)

Try sources in order; stop when k items are collected.

Use for cache-then-fallback patterns: a fast in-memory source first, an expensive vector source second. The second source isn't hit if the first answered. Items returned in the order sources were declared (preserves the "cache hit beats fresh fetch" intent).

NO dedupe here, deliberately, and it is not because the exposure is absent: a cache that returns 3 of 5 wanted items is followed by a vector lookup that can return copies of those same 3, so duplicates reach the caller exactly as they did in CompositeMemory. The difference is that the fix cannot be a merge pass. This class asks each source for k - len(collected) items, so collapsing a duplicate afterwards leaves a short list with nothing left to backfill from — the source that could have supplied the replacement has already been asked and answered. Doing it honestly means re-querying with a widened k (or over-fetching every tier by the expected overlap), which changes the round-trip count and the cost profile of the shape whose entire purpose is not making the second call. That is a design decision with a latency budget attached, not a merge, so it is left as its own item rather than smuggled in here. Wrap the chain in CompositeMemory(sources=[chain], dedupe="id") for a single-arm dedupe in the meantime.

accepts_writes property

accepts_writes: bool

Mirrors the FIRST source, because that is the only one write ever touches — downstream sources are read-only from this composite's point of view. A SequentialMemory whose cache tier is a ReadOnlyMemory stores nothing on write, and an enclosing fan-out must not report that as a commit. No sources means no write target, hence False.

write async

write(items: Iterable[MemoryItem], *, ctx: Ctx) -> None

Write to the FIRST source only — typical cache-tier pattern. Downstream sources are read-only from this composite's POV.

CachedMemory dataclass

CachedMemory(
    inner: MemorySource,
    ttl_seconds: float = 60.0,
    max_entries: int = 256,
    name: str = "",
    strict_scope: bool = False,
)

Decorator: cache query → results with a TTL.

The cache key is (query, k, frozenset(where or {})). Hits served without touching the inner source. Writes invalidate the cache (best-effort — a write may surface stale results to an in-flight query, which is the typical "read-after-write eventual consistency" trade-off).

ttl_seconds is wall-clock monotonic — the cache never serves an entry older than this. max_entries caps the cache size; eviction is LRU-ish (oldest insertion drops when full).

Not thread-safe; agents are single-flow per loop.

accepts_writes property

accepts_writes: bool

The cache never stores anything durably itself — it forwards writes — so whether this stack accepts writes is the inner source's answer.

CompactedMemory dataclass

CompactedMemory(
    inner: MemorySource,
    compactor: Any,
    max_items: int | None = None,
    name: str = "",
)

Decorator: shrink each MemoryItem.content via a Compactor.

Used when raw chunks would blow the prefix budget. Adapts the framework's Compactor capability (which operates on list[Message]) to operate on individual items: each item's content is wrapped as a single message, compacted, and the result becomes the new content.

max_items caps the post-compaction list — useful when the inner source returns more items than the cognition wants to fold into the prompt.

accepts_writes property

accepts_writes: bool

Compaction is a read-side concern; write is a straight pass-through, so the marker is the wrapped source's.

MemoryWriteRefused

MemoryWriteRefused(*, source: str, n: int)

Bases: AgentkitError

A ReadOnlyMemory turned away a write.

Subclasses AgentkitError rather than ValueError or a bare RuntimeError because kernel/errors.py makes one promise about this surface — "every framework-raised exception is a subclass of AgentkitError so a defensive except AgentkitError: catches the whole surface" — and a run boundary that catches the taxonomy should not need a second clause for this one.

Named rather than anonymous for the reason CompositeWriteError is: it lands inside CompositeWriteError.failed when a read-only member sits in a fan-out under the default policy, and an operator reading that split needs to tell "the registry is deliberately read-only" apart from "the registry is down". PermissionError was the other candidate and was rejected — ScopedMemory already raises it for a tenant-boundary violation, and collapsing "wrong tenant" into "right tenant, immutable source" would make the two indistinguishable at the catch site.

Attributes:

Name	Type	Description
source		the wrapped source's name — the label the write was aimed at, which is what identifies it inside a fan-out.
n		how many items the refused call carried. 0 is a real value; see ReadOnlyMemory.write for why an empty write still refuses.

ReadOnlyMemory dataclass

ReadOnlyMemory(
    inner: MemorySource,
    on_write: OnWritePolicy = "refuse",
    name: str = "",
)

Decorator: a source that is read-only by policy says so.

Every application has at least one source that must never be written — a curated knowledge base, an operator-maintained registry, a corpus of recorded facts the agent may consult and must not extend. Before this decorator the only protection was that nothing happened to call write, which is a property of the code as it currently stands rather than a rule: the first cognition taught to persist what it learned would have written into the registry and nothing would have complained. Wrapping the source turns the convention into an enforced boundary that survives the next refactor.

Single-responsibility like its neighbours: it constrains ONE verb. query is a byte-for-byte pass-through — same k, same where, same items back with their source stamp intact, which matters because that stamp is the identity a composite's merge and dedupe key on. A read-only member participates in recall exactly like any other source; that is the whole point of having it.

Two policies, because a fan-out forces the distinction:

• on_write="refuse" (default) — raise MemoryWriteRefused. The right answer for a source nothing has any business writing to; the exception names the source so a stack trace points at the policy rather than at the plumbing.
• on_write="ignore" — return normally, having written nothing. CompositeMemory.write broadcasts to every source, and one read-only member must not make the whole composite unwritable.

ignore is the dangerous one, because "silently succeeds" is the failure mode this package keeps writing tests against. It is accounted for three ways rather than none: refused_writes counts the turned-away calls on the instance, a memory.write_refused observation lands on the run's timeline with the source and the item count, and accepts_writes = False lets CompositeMemory file the member under a refused bucket instead of claiming in CompositeWriteError.accepted that it committed.

Nests either way round. ReadOnlyMemory(ScopedMemory(x)) refuses before the tenant guard runs (cheaper, and no PermissionError masking the real reason); ScopedMemory(ReadOnlyMemory(x)) checks the tenant first. Double wrapping is harmless — the outer refuses and the inner never runs.

name mirrors the wrapped source, like every other decorator here, so the guard is invisible in attribution.

query async

query(
    query: str,
    *,
    k: int,
    ctx: Ctx,
    where: dict[str, Any] | None = None,
) -> list[MemoryItem]

Untouched. No filtering, no re-stamping, no defensive copy — this decorator has an opinion about write and none at all about recall.

write async

write(items: Iterable[MemoryItem], *, ctx: Ctx) -> None

Refuse or drop, per on_write. self.inner.write is never awaited on either path — that is what makes "read-only" a fact about the wrapper rather than a hope about the backend, and it is why a wrapped source that raises on write is unreachable here.

items is materialised before the branch so the refusal can report n. It is the only reason to touch the iterable at all, and it costs one list build on a path that is about to raise or drop.

An EMPTY write still refuses. The policy is about the attempt, not about bytes moved: a caller that reaches write on a read-only source is running code that will carry items the moment its input is non-empty, and letting write([]) through means the violation is found in production instead of on the first test run. Nothing in the package pays for this — both CompositeMemory.write and SequentialMemory.write short-circuit on an empty list before reaching any source.

ScopedMemory dataclass

ScopedMemory(
    inner: MemorySource,
    enforce: Callable[[Ctx], Any] | None = None,
    name: str = "",
)

Fail-loud multi-tenant guard around any MemorySource.

Every query and every write runs enforce(ctx) first. If enforce is None, the default check fires — both ctx.scope.org_id and ctx.scope.domain_id must be set.

A custom enforce callable lets callers pin a stricter policy (e.g. "must match this specific tenant", "must be inside an approved checkpoint"). Returning a falsy value OR raising any exception inside enforce results in a PermissionError — the inner source is NOT touched.

The decorator's name mirrors the wrapped source so downstream consumers attributing MemoryItem.source see the same label they'd see without the wrapper — the guard is invisible on the happy path.

accepts_writes property

accepts_writes: bool

Mirror the wrapped source. A tenant guard around a read-only source is still read-only, and a composite that read True here would file the member under accepted.

FileMemory dataclass

FileMemory(files: Any, name: str = 'files')

Lexical MemorySource over a file-tree backend.

files is duck-typed: any object exposing async view(path), create(path, text), delete(path). The default is the in-tree InMemoryFiles; a durable / FS-backed implementation that conforms to the same surface drops in without changes.

query(query, k, ctx, where=None) — walks every file, scores by lowercase-substring match count, returns the top k items with metadata {"path": …}. Zero-match files are dropped. where={"path_prefix": "/foo"} scopes the walk to that prefix.

write(items, ctx) — writes each item via create. The target path comes from item.metadata["path"] if present, otherwise from a SHA-1 hash of the content.

JournalMemory dataclass

JournalMemory(
    journal: MutationJournal[T],
    render: Callable[[T], str],
    name: str = "journal",
)

Bases: Generic[T]

A MemorySource view on a MutationJournal[T].

render turns each journal entry into the text the cognition will read — pushed to the caller so the journal's entry type stays unconstrained.

Order semantics: query returns the most recent k entries (no relevance ranking — journals are temporal). Score is always None on returned items.

Filters

query — lexical filter; case-insensitive substring of the rendered text. Empty string disables the filter. where={"role": "..."} — keep only entries whose role attribute matches. Entries without a role attribute are skipped under this filter.

write async

write(items: Iterable[MemoryItem], *, ctx: Ctx) -> None

No-op — the journal's authored history is owned by journal.record(); a memory write seam would break that invariant.

ScratchpadMemory dataclass

ScratchpadMemory(
    context: WorkingContext, name: str = "scratchpad"
)

A MemorySource view on a WorkingContext.scratchpad.

Keys are inspected as-is; values are rendered via str(value) for matching. Returned items carry metadata={"key": key} so a downstream writer or reader can address the entry by name.

ToolMemory dataclass

ToolMemory(
    tool: Any,
    query_arg: str = "query",
    result_to_items: Callable[[Any, str], list[MemoryItem]]
    | None = None,
    name: str = "tool",
)

Adapter wrapping a FunctionTool (or any Tool-Protocol object) as a MemorySource. The tool becomes a knowledge probe the cognition can query alongside vector/file/journal sources.

The tool is invoked with a dict of args built from {query_arg: query, "limit": k, **(where or {})}. where keys are passed through unmodified so a tool with a typed parameter like site: can use them directly.

Parsing the tool's return into MemoryItems defaults to :func:default_parse and can be overridden per-instance with result_to_items=. Every emitted item has its source stamped with this instance's name so downstream consumers can attribute correctly.

Writes are no-ops. A wrapped tool is read-only as memory; the same underlying callable can be registered as a FunctionTool for LLM-decided side effects without conflict.

write async

write(items: Iterable[MemoryItem], *, ctx: Ctx) -> None

No-op. Tools are query-only as memory; a side-effecting tool belongs in the ToolRegistry, not as a memory write sink.

VectorMemory dataclass

VectorMemory(
    vector: Any,
    where: dict[str, Any] | None = None,
    name: str = "vector",
)

A MemorySource backed by a VectorPort.

where is the default metadata filter applied to every query and (informationally) attached at write time only if the caller embeds it in MemoryItem.metadata. At query time, a call-time where= merges over the constructor default — call-time wins on key collisions, so a wired-in narrow filter can be widened or overridden per-query without rebuilding the source.

Tenant isolation is delegated to the underlying VectorPort.upsert / VectorPort.search, which both take ctx.scope as their bucket key. Wrap with ScopedMemory for fail-loud enforcement at the framework boundary.

query async

query(
    query: str,
    *,
    k: int,
    ctx: Ctx,
    where: dict[str, Any] | None = None,
) -> list[Any]

Top-k scope-restricted lookup. The call-time where merges over the constructor default (call-time wins). The underlying VectorPort.search returns (score, Chunk); we adapt each pair to a MemoryItem and stamp source with this instance's name so downstream consumers can attribute the result.

Chunk.id rides along on MemoryItem.id. It used to be the one field the adapter dropped, which was an odd asymmetry — write has always keyed chunks by it, so the store knew the identity of every row it handed back and the item did not. Two consequences followed: CompositeMemory could not tell that the journal and the vector store had returned the same fact (the normal case, since the journal is usually what the store was built from), and writing a recalled item back inserted a SECOND copy under a fresh uuid instead of updating the row.

write async

write(items: Iterable[Any], *, ctx: Ctx) -> None

Index a batch of MemoryItems. Each item becomes a Chunk keyed by item.id, else metadata["id"] when the caller wrote one, else a fresh uuid. content becomes Chunk.text; the rest of metadata is preserved so callers can later where=-filter on whatever tags they wrote.

item.id takes precedence because read-modify-write is the loop this closes: an item recalled from this store carries its row's id, so writing it back must UPDATE that row. Under the uuid fallback it inserted a second copy of the same fact instead, which then had to be cleaned up downstream by CompositeMemory's dedupe — a duplicate manufactured by the store rather than found in it.

It is trusted ONLY when the item came from this instance, though, and that guard is load-bearing rather than defensive. item.id is a PROVENANCE record — the key of the row a backend handed back — and ids are unique only within a backend. CompositeMemory.write broadcasts every item to every source, so without the guard an item recalled from a journal whose row key happened to be "3" upserted over vector chunk "3": an unrelated fact silently destroyed, on the default write path. query stamps source with this instance's name, so the round trip above still matches and still updates in place.

metadata["id"] stays supported underneath, and outranks the missing provenance match, because it is the caller INSTRUCTING this store which row to write rather than reporting where an item came from. It is popped either way so the key never survives into Chunk.metadata — two places claiming to be the id would make the next round trip depend on which one won.

score_sort_rerank async

score_sort_rerank(
    query: str, items: list[MemoryItem], *, k: int
) -> list[MemoryItem]

Default: stable sort by score descending. None scores sink to the bottom but aren't dropped — a tool-wrapped source that doesn't rank still gets to surface in the top-k.

as_grounder

as_grounder(
    memory: MemorySource,
    *,
    k: int = 5,
    where: dict[str, Any] | None = None,
    format: Callable[[list[MemoryItem]], str] | None = None,
) -> Callable[[Ctx, str], Awaitable[str]]

Adapt a MemorySource into a RequestBuilder.Grounder.

k and where are baked into the returned closure. format overrides the default [source] content rendering — pass a custom formatter when the prompt expects a specific shape (Markdown bullets, numbered citations, JSON, etc.).

The query passed to the grounder is the user task — the same string the agent is about to answer. For "query rewriting" before retrieval (HyDE, query decomposition) wrap the memory itself rather than transforming inside the grounder.

as_grounding_source

as_grounding_source(
    memory: MemorySource,
    *,
    k: int = 5,
    where: dict[str, Any] | None = None,
) -> Callable[[Ctx, str], Awaitable[Sequence[MemoryItem]]]

Adapt a MemorySource into a RequestBuilder.GroundingSource.

The same closure as_grounder builds, minus the flattening step — which is the whole implementation and also the whole point. There is no format= here because rendering has moved to where it can see what it is rendering: RequestBuilder(render=...), downstream of RequestBuilder(admit=...). Keeping a formatter on this side too would give two places to decide the same thing, and the one on this side runs BEFORE admission, so a rule like "drop model-authored memories" would be applied to text that had already been joined.

k and where are baked in at wiring time exactly as before, and the query is still the user task verbatim — for query rewriting (HyDE, decomposition) wrap the MemorySource rather than reaching inside this closure.

Note the return annotation is the structural Callable[...] rather than GroundingSource: agentkit.state.memory must not import agentkit.capabilities (the capabilities side imports MemoryItem from here, and the pair would cycle). as_grounder above spells its return type out for the same reason.

default_parse

default_parse(result: Any, query: str) -> list[MemoryItem]

Best-effort parser for tool-call return values.

Handles the four shapes a search-ish tool commonly returns:

• dict with a "results" key → recurse on the results list
• list[dict] → each dict is treated as {content, score?, metadata?} (anything else in the dict becomes metadata; "text" and "snippet" are accepted as content aliases for convenience)
• list[str] → each string becomes a MemoryItem.content
• str → split on blank lines ("\n\n") into multiple items; a single-paragraph string yields one item

Anything else is wrapped as a single item with content=str(result) — enough to be visible in traces, not enough to be useful: callers with richer return shapes should pass result_to_items=.

The query argument is accepted for parity with custom parsers (which may want to highlight matches) but the default parser ignores it.

render_grounding

render_grounding(items: Sequence[MemoryItem]) -> str

Render records as attributed evidence; callers decide admission first.