`agentkit.memory`¶

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

agentkit.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.

MemoryItem `dataclass` ¶

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

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. 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.

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",
)

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.

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).

Partial failures surface as :class:CompositeWriteError with a per-source accepted/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.

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).

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.

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.

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.

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.

write `async` ¶

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

Index a batch of MemoryItems. Each item becomes a Chunk keyed by 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.

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.

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.

agentkit.memory¶

MemoryItem dataclass ¶

MemorySource ¶

Reranker ¶

CompositeMemory dataclass ¶

write async ¶

SequentialMemory dataclass ¶

write async ¶

CachedMemory dataclass ¶

CompactedMemory dataclass ¶

ScopedMemory dataclass ¶

FileMemory dataclass ¶

JournalMemory dataclass ¶

write async ¶

ScratchpadMemory dataclass ¶

ToolMemory dataclass ¶

write async ¶

VectorMemory dataclass ¶

query async ¶

write async ¶

score_sort_rerank async ¶

as_grounder ¶

default_parse ¶

`agentkit.memory`¶

MemoryItem `dataclass` ¶

CompositeMemory `dataclass` ¶

write `async` ¶

SequentialMemory `dataclass` ¶

write `async` ¶

CachedMemory `dataclass` ¶

CompactedMemory `dataclass` ¶

ScopedMemory `dataclass` ¶

FileMemory `dataclass` ¶

JournalMemory `dataclass` ¶

write `async` ¶

ScratchpadMemory `dataclass` ¶

ToolMemory `dataclass` ¶

write `async` ¶

VectorMemory `dataclass` ¶

query `async` ¶

write `async` ¶

score_sort_rerank `async` ¶