API reference¶

Autogenerated from the source docstrings via mkdocstrings. The signatures and behaviour shown here are the ground truth; the narrative pages link back here when they reference a specific class or method.

Core types¶

MemoryRecord `dataclass` ¶

MemoryRecord(
    *,
    agent_id: str,
    content: str,
    tier: MemoryTier,
    id: str = _new_id(),
    embedding: list[float] | None = None,
    created_at: datetime = _utcnow(),
    last_accessed: datetime | None = None,
    access_count: int = 0,
    expires_at: datetime | None = None,
    metadata: dict[str, Any] = dict(),
)

Common shape for anything Mneme stores or ranks.

kw_only means every field is passed by keyword, which frees subclasses to add required fields without fighting dataclass default-ordering rules.

Attributes:

Name	Type	Description
`agent_id`	`str`	Namespaces a record to one agent. Memory is never shared across agents in v1.0.
`content`	`str`	The natural-language payload that gets embedded and shown.
`tier`	`MemoryTier`	Which tier this record belongs to. Concrete subclasses default it.
`id`	`str`	Opaque unique id, generated if not supplied.
`embedding`	`list[float] \| None`	The content's vector, or `None` until it has been embedded.
`created_at`	`datetime`	When the record was first written (UTC, aware).
`last_accessed`	`datetime \| None`	When it was last returned by retrieval, or `None` if never. Drives access-frequency decay.
`access_count`	`int`	How many times retrieval has surfaced it. Also drives decay.
`expires_at`	`datetime \| None`	When the forgetting pass should delete this record. `None` (default) means "never". Set this when you have a known TTL — the forgetting pass deletes anything with `expires_at <= now`. :meth:`MemoryManager.retrieve` also defensively filters out expired records even between forgetting passes.
`metadata`	`dict[str, Any]`	Free-form caller-supplied tags (source, conversation id, etc.).

touch ¶

touch(*, now: datetime | None = None) -> None

Record that retrieval just surfaced this memory.

Bumps access_count and stamps last_accessed. The forgetting pass reads both to decide what has gone cold.

WorkingMemory `dataclass` ¶

WorkingMemory(
    *,
    agent_id: str,
    content: str,
    tier: MemoryTier = MemoryTier.WORKING,
    id: str = _new_id(),
    embedding: list[float] | None = None,
    created_at: datetime = _utcnow(),
    last_accessed: datetime | None = None,
    access_count: int = 0,
    expires_at: datetime | None = None,
    metadata: dict[str, Any] = dict(),
    role: str,
)

Bases: MemoryRecord

A single turn in the current conversation.

Lives in process memory (FIFO), not the storage backend. Adds a role so a turn can be replayed as a chat message.

EpisodicMemory `dataclass` ¶

EpisodicMemory(
    *,
    agent_id: str,
    content: str,
    tier: MemoryTier = MemoryTier.EPISODIC,
    id: str = _new_id(),
    embedding: list[float] | None = None,
    created_at: datetime = _utcnow(),
    last_accessed: datetime | None = None,
    access_count: int = 0,
    expires_at: datetime | None = None,
    metadata: dict[str, Any] = dict(),
)

Bases: MemoryRecord

A specific past interaction, embedded and persisted to the backend.

Retrieved by similarity to the current query, optionally filtered by recency or metadata.

SemanticFact `dataclass` ¶

SemanticFact(
    *,
    agent_id: str,
    content: str,
    tier: MemoryTier = MemoryTier.SEMANTIC,
    id: str = _new_id(),
    embedding: list[float] | None = None,
    created_at: datetime = _utcnow(),
    last_accessed: datetime | None = None,
    access_count: int = 0,
    expires_at: datetime | None = None,
    metadata: dict[str, Any] = dict(),
    confidence: float = 1.0,
    provenance: list[str] = list(),
)

Bases: MemoryRecord

A consolidated fact extracted from episodic memories.

Attributes:

Name	Type	Description
`confidence`	`float`	How sure consolidation is of this fact, in `[0.0, 1.0]`. Weights the record during retrieval.
`provenance`	`list[str]`	Ids of the episodic memories that contributed to this fact, so a fact can be traced back to its evidence.

RetrievalResult `dataclass` ¶

RetrievalResult(
    *,
    record: MemoryRecord,
    score: float,
    similarity: float | None = None,
    recency: float | None = None,
    authority: float | None = None,
)

One ranked hit returned by mneme.retrieve.

Wraps the underlying record with the final score plus the component scores that produced it, so callers can see why something ranked where it did instead of getting an opaque number.

Attributes:

Name	Type	Description
`record`	`MemoryRecord`	The memory that matched.
`score`	`float`	Final fused score used for ranking (higher is better).
`similarity`	`float \| None`	Raw query/record vector similarity, if computed.
`recency`	`float \| None`	Freshness-decay contribution, if computed.
`authority`	`float \| None`	Tier-authority contribution (semantic > episodic > working), if computed.

MemoryTier ¶

Bases: StrEnum

The three tiers a memory can belong to.

A StrEnum so a tier serialises to its plain value ("episodic") in JSON and storage backends without a custom encoder.

Manager¶

MemoryManager ¶

MemoryManager(
    *,
    agent_id: str,
    backend: MnemeBackend,
    embedder: EmbeddingProvider,
    working_size: int = 20,
    llm_judge: LLMJudge | None = None,
)

Owns one agent's three tiers and the shared backend + embedder.

Parameters:

Name	Type	Description	Default
`agent_id`	`str`	The namespace for everything this manager touches. Passed through to each tier so backend writes are isolated.	required
`backend`	`MnemeBackend`	Storage + vector-search for episodic + semantic tiers. Must be configured for `embedder.dimensions`.	required
`embedder`	`EmbeddingProvider`	Embedding provider used for both write-time (add) and query-time (search) vectors.	required
`working_size`	`int`	Max turns kept in working memory. Default 20.	`20`
`llm_judge`	`LLMJudge \| None`	Optional :class:`LLMJudge` used by :meth:`consolidate`. `None` (default) is fine if you never call `consolidate()`; it raises `RuntimeError` when invoked without a judge.	`None`

working `instance-attribute` ¶

working = WorkingMemoryTier(
    agent_id=agent_id, max_size=working_size
)

episodic `instance-attribute` ¶

episodic = EpisodicMemoryTier(
    agent_id=agent_id, backend=backend, embedder=embedder
)

semantic `instance-attribute` ¶

semantic = SemanticMemoryTier(
    agent_id=agent_id, backend=backend, embedder=embedder
)

retrieve ¶

retrieve(
    query: str,
    *,
    k: int = 5,
    tiers: list[MemoryTier] | None = None,
    authority_weights: dict[MemoryTier, float]
    | None = None,
    half_life_days: float = DEFAULT_HALF_LIFE_DAYS,
    use_confidence: bool = True,
    now: datetime | None = None,
) -> list[RetrievalResult]

Return the top-k memories most relevant to query.

Embeds the query once, asks the backend for an over-fetched candidate set across the requested tiers, then re-ranks each candidate by

.. math::

score = similarity \times authority \times recency \times confidence?

where:

similarity is the raw cosine similarity returned by the backend (clamped to a non-negative floor so negative-similarity records never beat positive-similarity ones on tier or recency boosts).
authority is :data:DEFAULT_AUTHORITY_WEIGHTS for the record's tier, or the override in authority_weights.
recency is 0.5 ** (age_days / half_life_days) — exponential decay with a configurable half-life.
confidence? is the :attr:SemanticFact.confidence value if the record is a fact and use_confidence is True; otherwise 1.

Parameters:

Name	Type	Description	Default
`query`	`str`	The natural-language query to embed and match against.	required
`k`	`int`	Maximum number of results. Default 5.	`5`
`tiers`	`list[MemoryTier] \| None`	Which tiers to search. Default `[EPISODIC, SEMANTIC]`. Passing `MemoryTier.WORKING` raises `ValueError` — working memory has its own access path (`manager.working.turns()`) and is not embedded.	`None`
`authority_weights`	`dict[MemoryTier, float] \| None`	Override per-tier authority. Missing tiers default to 0.0. Use to express domain-specific policy (e.g., raise EPISODIC weight for an agent where the conversation history matters more than distilled facts).	`None`
`half_life_days`	`float`	Days after which a record's recency factor drops to 0.5. Default 7. Smaller = faster forgetting.	`DEFAULT_HALF_LIFE_DAYS`
`use_confidence`	`bool`	Whether to multiply semantic facts' scores by their :attr:`SemanticFact.confidence`. Default True. Disable to make confidence a metadata-only signal.	`True`
`now`	`datetime \| None`	Reference time for recency computation. Defaults to `datetime.now(UTC)`. Pass explicitly in tests.	`None`

Returns:

Type	Description
`list[RetrievalResult]`	A list of :class:`RetrievalResult`, ordered by `score`
`list[RetrievalResult]`	descending. Each result carries the raw `similarity`,
`list[RetrievalResult]`	computed `recency`, and `authority` so callers can inspect
`list[RetrievalResult]`	why a record ranked where it did.

consolidate ¶

consolidate(
    *,
    since: datetime | None = None,
    batch_size: int = DEFAULT_CONSOLIDATION_BATCH_SIZE,
    dedup_threshold: float = DEFAULT_DEDUP_THRESHOLD,
    max_episodes: int = DEFAULT_CONSOLIDATION_MAX_EPISODES,
) -> list[SemanticFact]

Promote recent episodic memories into semantic facts.

Reads up to max_episodes episodes (filtered by since when given), batches them, calls :attr:llm_judge per batch with the fact-extraction prompt, then for each extracted fact:

Embeds the fact content.
Searches existing semantic facts for the most similar one.
If best-similarity ≥ dedup_threshold → merge into the existing fact (newer content wins; provenance is the union; confidence is the max).
Otherwise → write the fact as a new :class:SemanticFact.

Parameters:

Name	Type	Description	Default
`since`	`datetime \| None`	If given, only episodes with `created_at >= since` are considered. `None` reads from the very first episode.	`None`
`batch_size`	`int`	Episodes per LLM call. Default 20.	`DEFAULT_CONSOLIDATION_BATCH_SIZE`
`dedup_threshold`	`float`	Cosine similarity above which two facts are considered duplicates. Default 0.85.	`DEFAULT_DEDUP_THRESHOLD`
`max_episodes`	`int`	Hard cap on how many episodes a single `consolidate()` run will process. Default 200.	`DEFAULT_CONSOLIDATION_MAX_EPISODES`

Returns:

Name	Type	Description
`The`	`list[SemanticFact]`	class:`SemanticFact` records that were either created or
	`list[SemanticFact]`	merged-into during this run.

Raises:

Type	Description
`RuntimeError`	if `self.llm_judge is None`.

forget ¶

forget(
    *,
    now: datetime | None = None,
    ttl_only: bool = False,
    access_floor: int = DEFAULT_ACCESS_FLOOR,
    cold_age_days: float = DEFAULT_COLD_AGE_DAYS,
    max_per_tier: int = DEFAULT_FORGET_MAX_PER_TIER,
) -> dict[str, int]

Delete expired and cold records from the persisted tiers.

Two phases, in order:

TTL: every record whose expires_at is set and <= now is deleted unconditionally. This is the only thing that runs when ttl_only=True.
Access-frequency decay: records older than cold_age_days with access_count <= access_floor are deleted. The defaults (30 days, 0 accesses) mean "never retrieved in 30 days → gone."

Working memory is untouched — it lives in process and evicts via FIFO already.

Parameters:

Name	Type	Description	Default
`now`	`datetime \| None`	Reference time. Defaults to `datetime.now(UTC)`. Pass explicitly in tests.	`None`
`ttl_only`	`bool`	If `True`, skip phase 2 and only delete expired records. Useful when you want hard-deadline semantics without the heuristic cold-eviction.	`False`
`access_floor`	`int`	Records with `access_count <= access_floor` are eligible for cold eviction. Default 0 — anything ever retrieved survives.	`DEFAULT_ACCESS_FLOOR`
`cold_age_days`	`float`	A record must be at least this old (by `created_at`) to be considered cold.	`DEFAULT_COLD_AGE_DAYS`
`max_per_tier`	`int`	Hard cap on records scanned per tier per call. Forgetting is O(records scanned). On a backend with millions of records, raise the cap or run forget() more often with a smaller window.	`DEFAULT_FORGET_MAX_PER_TIER`

Returns:

Type	Description
`dict[str, int]`	`{"expired": N, "cold": M}` so callers can audit and log.

start_scheduler ¶

start_scheduler(
    *,
    consolidate_every: timedelta | None = timedelta(
        minutes=30
    ),
    forget_every: timedelta | None = timedelta(days=1),
    forget_kwargs: dict[str, Any] | None = None,
    consolidate_kwargs: dict[str, Any] | None = None,
) -> None

Start a background scheduler running consolidation + forgetting.

Requires the [scheduler] extra (pip install 'mneme[scheduler]'). Spins up an APScheduler BackgroundScheduler in a daemon thread and registers up to two periodic jobs.

Idempotent: calling :meth:start_scheduler while one is already running raises RuntimeError — stop the existing scheduler first.

Parameters:

Name	Type	Description	Default
`consolidate_every`	`timedelta \| None`	Period for :meth:`consolidate`. `None` skips the consolidation job (useful when you handle it yourself). Default 30 minutes.	`timedelta(minutes=30)`
`forget_every`	`timedelta \| None`	Period for :meth:`forget`. `None` skips the forgetting job. Default daily.	`timedelta(days=1)`
`consolidate_kwargs`	`dict[str, Any] \| None`	Keyword args forwarded to each `consolidate()` call.	`None`
`forget_kwargs`	`dict[str, Any] \| None`	Keyword args forwarded to each `forget()` call.	`None`

Raises:

Type	Description
`ImportError`	if the `[scheduler]` extra isn't installed.
`RuntimeError`	if a scheduler is already running, or if `consolidate_every` is set but `self.llm_judge` is None.

stop_scheduler ¶

stop_scheduler(*, wait: bool = True) -> None

Stop the background scheduler if one is running. Idempotent.

Parameters:

Name	Type	Description	Default
`wait`	`bool`	Forwarded to APScheduler's `shutdown(wait=...)`. If `True` (default), blocks until in-flight jobs finish.	`True`

clear_all ¶

clear_all() -> dict[str, int]

Wipe every memory this agent has, across every tier.

Returns a dict {tier_name: count_deleted} so callers can audit. Useful for tests and for explicit "forget this agent" operations.

Tiers¶

WorkingMemoryTier ¶

WorkingMemoryTier(*, agent_id: str, max_size: int = 20)

A FIFO buffer of recent conversation turns for one agent.

Parameters:

Name	Type	Description	Default
`agent_id`	`str`	Namespaces this buffer to a single agent. Matches the `agent_id` on the persisted tiers' records.	required
`max_size`	`int`	Maximum number of turns to retain. Adding a turn when the buffer is full evicts the oldest. Default 20 — enough for most multi-turn conversations without bloating prompts.	`20`

add ¶

add(
    *,
    role: str,
    content: str,
    metadata: dict[str, Any] | None = None,
) -> WorkingMemory

Append a turn. Evicts the oldest if the buffer is full.

Returns the created :class:WorkingMemory so callers can hold on to it (typically only the manager does).

turns ¶

turns() -> list[WorkingMemory]

Return the current turns in chronological order (oldest first).

clear ¶

clear() -> int

Remove every turn. Returns the number removed.

EpisodicMemoryTier ¶

EpisodicMemoryTier(
    *,
    agent_id: str,
    backend: MnemeBackend,
    embedder: EmbeddingProvider,
)

Persistent, semantically-searchable conversation history for one agent.

Parameters:

Name	Type	Description	Default
`agent_id`	`str`	Namespaces this tier's records.	required
`backend`	`MnemeBackend`	The :class:`MnemeBackend` used for persistence and search.	required
`embedder`	`EmbeddingProvider`	The :class:`EmbeddingProvider` used to compute vectors at write time and at query time.	required

add ¶

add(
    content: str, *, metadata: dict[str, Any] | None = None
) -> EpisodicMemory

Embed content and persist it as an episodic memory.

search ¶

search(
    query: str,
    *,
    k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[EpisodicMemory, float]]

Return the top-k episodes most similar to query.

Embeds the query, calls :meth:MnemeBackend.search scoped to the episodic tier, and returns (record, similarity) pairs sorted by similarity descending. Records are typed as :class:EpisodicMemory — the tier filter guarantees the backend returns only those.

clear ¶

clear() -> int

Delete every episodic record for this agent. Returns count deleted.

SemanticMemoryTier ¶

SemanticMemoryTier(
    *,
    agent_id: str,
    backend: MnemeBackend,
    embedder: EmbeddingProvider,
)

Persistent, semantically-searchable consolidated facts for one agent.

Shape is parallel to :class:EpisodicMemoryTier with one difference:

:meth:add accepts confidence and provenance so callers (and the consolidator) can record where a fact came from and how sure we are about it.

add ¶

add(
    content: str,
    *,
    confidence: float = 1.0,
    provenance: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
) -> SemanticFact

Embed content and persist it as a semantic fact.

Parameters:

Name	Type	Description	Default
`content`	`str`	The fact, in natural language. E.g., "user prefers TypeScript examples".	required
`confidence`	`float`	How sure we are, in `[0.0, 1.0]`. Retrieval ranking (Step 6) weights by this. Default 1.0 (set by hand → trust it).	`1.0`
`provenance`	`list[str] \| None`	Optional list of episodic-memory ids that produced this fact. Populated by the consolidator; manual `add` callers typically leave it empty.	`None`
`metadata`	`dict[str, Any] \| None`	Free-form caller tags.	`None`

search ¶

search(
    query: str,
    *,
    k: int = 5,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[SemanticFact, float]]

Return the top-k facts most similar to query.

Backends¶

MnemeBackend ¶

Bases: Protocol

Pluggable storage + vector-search layer for Mneme's persisted tiers.

Implementations are usually classes but can be any object that satisfies this structural type. @runtime_checkable enables isinstance checks in tests; method calls are still resolved by structural typing, not by inheritance.

Method contract notes:

Every record passed to :meth:upsert must have embedding set — backends are not expected to embed text. Records without embeddings are not searchable and should be rejected with ValueError.
:meth:search returns results sorted by similarity, descending. Ties should be broken deterministically (implementations may use created_at descending as a tiebreaker so newer wins on equal score).
Backends are multi-tenant: every read/write is scoped by agent_id. There is no cross-agent leakage.

upsert ¶

upsert(record: MemoryRecord) -> None

Insert a new record or replace an existing one by record.id.

Raises:

Type	Description
`ValueError`	if `record.embedding` is `None`.

get ¶

get(record_id: str) -> MemoryRecord | None

Return the record with the given id, or None if not found.

delete ¶

delete(record_id: str) -> bool

Delete a record by id. Returns True if the id existed, else False.

search ¶

search(
    *,
    query_embedding: list[float],
    agent_id: str,
    k: int = 5,
    tiers: Iterable[MemoryTier] | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[tuple[MemoryRecord, float]]

Return up to k records most similar to query_embedding.

Parameters:

Name	Type	Description	Default
`query_embedding`	`list[float]`	The query vector. Must be the same dimension as stored records' embeddings; backends may raise if not.	required
`agent_id`	`str`	Namespace filter — only records belonging to this agent are searched.	required
`k`	`int`	Maximum number of results. Backends should return fewer if fewer matching records exist.	`5`
`tiers`	`Iterable[MemoryTier] \| None`	If given, restrict search to records in these tiers. If `None` (default), all tiers are searched.	`None`
`metadata_filter`	`dict[str, Any] \| None`	If given, only records whose `metadata` dict contains every key/value pair are returned. Equality only; no operators.	`None`

Returns:

Type	Description
`list[tuple[MemoryRecord, float]]`	`[(record, similarity), ...]` sorted by similarity descending.
`list[tuple[MemoryRecord, float]]`	Similarity is in `[-1.0, 1.0]` for cosine; backends should
`list[tuple[MemoryRecord, float]]`	document any other metric they use.

list_recent ¶

list_recent(
    *,
    agent_id: str,
    tier: MemoryTier,
    since: datetime | None = None,
    limit: int = 1000,
) -> list[MemoryRecord]

Return records in a tier ordered by created_at descending.

Added in v0.2 to support consolidation, which needs to enumerate episodes by time rather than similarity. Distinct from :meth:search in that there is no query vector and no ranking — just a windowed time-ordered listing.

Parameters:

Name	Type	Description	Default
`agent_id`	`str`	Namespace filter (mandatory, like every other method).	required
`tier`	`MemoryTier`	Which tier to list. Single tier only — consolidation doesn't have a use case for cross-tier listings, and a single tier per call keeps the contract simple.	required
`since`	`datetime \| None`	If given, only records with `created_at >= since` are returned. `None` means "all of them" subject to `limit`.	`None`
`limit`	`int`	Maximum number of records to return. Default 1000.	`1000`

Returns:

Type	Description
`list[MemoryRecord]`	`[record, ...]` newest first. Length is at most `limit`.

touch ¶

touch(record_ids: Iterable[str], *, now: datetime) -> int

Mark records as just-accessed: bump access_count, set last_accessed.

Added in v0.3. Called by :meth:MemoryManager.retrieve for the top-k records it actually returns so access-frequency decay (used by :meth:MemoryManager.forget) reflects real usage across restarts.

Parameters:

Name	Type	Description	Default
`record_ids`	`Iterable[str]`	Ids to touch. Unknown ids are silently skipped.	required
`now`	`datetime`	Timestamp to write into `last_accessed`.	required

Returns:

Type	Description
`int`	The number of records actually touched (i.e. found in storage).
`int`	Useful for observability; safe to ignore.

count ¶

count(
    *, agent_id: str, tier: MemoryTier | None = None
) -> int

Number of records for an agent, optionally scoped to one tier.

clear ¶

clear(
    *, agent_id: str, tier: MemoryTier | None = None
) -> int

Delete records for an agent, optionally tier-scoped. Returns count deleted.

Useful in tests and for explicit "forget this agent" operations.

InMemoryBackend ¶

InMemoryBackend()

Process-local backend. Satisfies :class:mneme.backends.MnemeBackend.

Holds every record in a single dict[str, MemoryRecord] keyed by id. Search scans the dict, filters, computes cosine similarity for each candidate, and returns the top-k. This is O(n) and fine for a few thousand records — past that, switch to a real vector index.

SQLiteBackend ¶

SQLiteBackend(*, path: str | Path, dimensions: int)

SQLite + sqlite-vec implementation of :class:mneme.MnemeBackend.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	File path for the database, or `":memory:"` for an in-process database (useful for tests). Accepts `str` or :class:`Path`.	required
`dimensions`	`int`	Embedding dimensionality. Locked at construction time — `vec0` virtual tables require a fixed dimension. Records whose embedding length differs raise :class:`ValueError`.	required

Not thread-safe in v0.1. Open one backend instance per process and route all calls through it. For multi-process write workloads, switch to a backend that supports concurrent writes (Qdrant, pgvector).

close ¶

close() -> None

Close the underlying sqlite connection. Idempotent.

sqlite3.Connection.close() raises ProgrammingError if called on an already-closed connection; we suppress that so callers can call close() defensively without guarding it.

QdrantBackend ¶

QdrantBackend(
    *,
    client: QdrantClient,
    collection: str = "mneme",
    dimensions: int,
)

Qdrant-backed implementation of :class:mneme.MnemeBackend.

Parameters:

Name	Type	Description	Default
`client`	`QdrantClient`	A constructed `qdrant_client.QdrantClient`. Pass the client the application already has if Qdrant is shared.	required
`collection`	`str`	Collection name. Created on first construction if it doesn't exist already. Default `"mneme"`.	`'mneme'`
`dimensions`	`int`	Embedding dimensionality. Locked at collection creation; mismatches raise :class:`ValueError`.	required

Not thread-safe at the Python layer; Qdrant's HTTP/gRPC client itself is generally safe for concurrent calls, but Mneme's own writers should coordinate.

PgVectorBackend ¶

PgVectorBackend(
    *,
    dsn: str,
    dimensions: int,
    table: str = "mneme_records",
)

Postgres + pgvector implementation of :class:mneme.MnemeBackend.

Parameters:

Name	Type	Description	Default
`dsn`	`str`	A psycopg connection string (`"postgresql://user:pw@host/db"`). One connection per backend instance; not a connection pool.	required
`dimensions`	`int`	Embedding dimensionality. Locked at table creation; mismatches raise :class:`ValueError`.	required
`table`	`str`	Table name. Default `"mneme_records"`. Created on first construction if it doesn't exist already.	`'mneme_records'`

Not thread-safe; psycopg3 sync connections aren't safe to share across threads. Open one backend per thread (or use a connection pool above this layer) if you need concurrency.

close ¶

close() -> None

Close the underlying psycopg connection. Idempotent.

Embedders¶

EmbeddingProvider ¶

Bases: Protocol

Turns text into vectors.

Implementations declare a fixed :attr:dimensions and expose a single :meth:embed call that takes a list of strings and returns one vector per string in the same order.

Single-text use is just provider.embed([text])[0]. We deliberately do not add a convenience method for that case — every implementation would then have to provide both, and the one-line caller is just as clear.

embed ¶

embed(texts: Sequence[str]) -> list[list[float]]

Embed texts and return one vector per input in order.

Implementations should accept any Sequence[str] and return a plain list[list[float]]. Vectors should have length :attr:dimensions; callers may assume this and pass them straight to a backend constructed with matching dimensions.

OpenAIEmbeddings ¶

OpenAIEmbeddings(
    *,
    model: str = "text-embedding-3-small",
    dimensions: int | None = None,
    api_key: str | None = None,
)

Embedding provider backed by OpenAI's /v1/embeddings endpoint.

Parameters:

Name	Type	Description	Default
`model`	`str`	Embedding model name. Defaults to `text-embedding-3-small` (1536 dims, cheap, the right call for almost everything).	`'text-embedding-3-small'`
`dimensions`	`int \| None`	Optional override. Most callers leave this `None` and let the model's default dimension apply. If you pass a custom value, OpenAI returns a truncated/projected vector of that size (only supported on `-3-small` and `-3-large`).	`None`
`api_key`	`str \| None`	Optional explicit key. If `None`, the OpenAI SDK falls back to `OPENAI_API_KEY` from the environment.	`None`

Reads OPENAI_API_KEY from the environment when api_key is not provided. Mneme itself never auto-loads .env files; do that in your application code (or via the test conftest in this repo).

HashEmbedder ¶

HashEmbedder(*, dimensions: int = 16, seed: int = 0)

Deterministic embedder for tests. No API calls, no model.

Given the same input, returns the same vector. Different inputs return different vectors. There is no semantic structure — "cat" and "kitten" are no more similar in this space than "cat" and "quantum chromodynamics". Use this to test plumbing (does an embedding flow from caller to backend correctly?) — never to test retrieval quality.

Vectors are length-1 (unit-norm) so cosine similarity behaves sanely.

Parameters:

Name	Type	Description	Default
`dimensions`	`int`	Length of each output vector. Default 16 — small enough to keep tests fast, large enough that hash collisions in any single coordinate don't dominate similarity.	`16`
`seed`	`int`	Hashed in with each input so two HashEmbedders with different seeds produce different vector spaces for the same strings.	`0`

LLM judges¶

LLMJudge ¶

Bases: Protocol

A model that returns structured JSON for a given chat prompt.

Implementations declare:

:attr:model — informational name used in logs (and exposed so callers can record which model produced a fact).
:meth:complete — single method. Takes OpenAI-style chat messages and a JSON schema; returns the parsed JSON as a dict.

The judge does not own the prompt. Callers compose the messages, hand them in, and parse the response from the dict shape they requested.

complete ¶

complete(
    *,
    messages: Sequence[Mapping[str, str]],
    response_schema: Mapping[str, Any],
    temperature: float = 0.0,
    schema_name: str = "Response",
) -> dict[str, Any]

Run a chat completion with structured output.

Parameters:

Name	Type	Description	Default
`messages`	`Sequence[Mapping[str, str]]`	OpenAI-style messages — each `{"role": "...", "content": "..."}`. Mneme uses `"system"` and `"user"` roles for v0.2; future tasks may add `"assistant"` for few-shot prompts.	required
`response_schema`	`Mapping[str, Any]`	A JSON schema (the inner schema object, not the OpenAI `response_format` wrapper) describing the expected response shape. Implementations should enforce it strictly — use `additionalProperties: false` and explicit `required`.	required
`temperature`	`float`	Sampling temperature. Default 0.0 — fact extraction wants deterministic, low-variance output.	`0.0`
`schema_name`	`str`	Human-readable name for the schema. Surfaced in OpenAI's API and useful for logs / debugging.	`'Response'`

Returns:

Type	Description
`dict[str, Any]`	The parsed JSON response as a dict. Validation against the schema
`dict[str, Any]`	is the implementation's responsibility (most APIs that accept JSON
`dict[str, Any]`	schemas enforce it server-side; the dict you get back conforms).

OpenAILLMJudge ¶

OpenAILLMJudge(
    *,
    model: str = "gpt-4o-mini",
    api_key: str | None = None,
)

LLM judge backed by OpenAI's chat completions with JSON schema response.

Parameters:

Name	Type	Description	Default
`model`	`str`	Chat model name. Defaults to `gpt-4o-mini` — cheap, fast, good enough for consolidation. Switch to `gpt-4o` when the quality differential matters for your eval.	`'gpt-4o-mini'`
`api_key`	`str \| None`	Optional explicit key. If `None`, the OpenAI SDK falls back to `OPENAI_API_KEY` from the environment.	`None`

Uses response_format={"type": "json_schema", ...} with strict: True so the model is constrained at decode time to produce schema-valid output.

MockLLMJudge ¶

MockLLMJudge(
    *, handler: MockJudgeHandler, model: str = "mock"
)

Deterministic LLM judge for tests. No API calls.

Construct with a handler callable that receives the same kwargs :meth:complete was called with and returns the dict the judge should return. This lets tests inspect prompts and inject whatever structured response the test needs without any real model in the loop.

Parameters:

Name	Type	Description	Default
`handler`	`MockJudgeHandler`	Callable taking `(*, messages, response_schema, temperature, schema_name)` and returning a dict.	required
`model`	`str`	Informational model name. Defaults to `"mock"`.	`'mock'`

Adapters¶

The adapter classes live in mneme.adapters rather than the package root — they each require an optional extra to install. Import them explicitly:

from mneme.adapters import MnemeChatMessageHistory  # needs [langchain]
from mneme.adapters import MnemeLlamaIndexMemory    # needs [llamaindex]
from mneme.adapters import context_for              # always available

MnemeChatMessageHistory ¶

MnemeChatMessageHistory(
    manager: MemoryManager,
    *,
    also_persist_episodic: bool = True,
)

Bases: BaseChatMessageHistory

LangChain BaseChatMessageHistory backed by a Mneme MemoryManager.

Parameters:

Name	Type	Description	Default
`manager`	`MemoryManager`	A constructed :class:`mneme.MemoryManager`. The adapter owns nothing — pass the same manager into other adapters or use it directly alongside.	required
`also_persist_episodic`	`bool`	If `True` (default), every message also gets written to the episodic tier so it becomes searchable via `manager.retrieve()` and consolidatable. Disable if you want the LangChain history to be ephemeral (working-memory only, cleared on session end).	`True`

messages `property` ¶

messages: list[BaseMessage]

Return current working-memory turns as LangChain message objects.

add_message ¶

add_message(message: BaseMessage) -> None

Append a message to Mneme.

Writes to working memory (for .messages reads) and to episodic memory (for retrieval/consolidation), unless also_persist_episodic was set False at construction.

clear ¶

clear() -> None

Wipe working memory. Episodic survives — call manager.episodic.clear() explicitly if you want to forget past conversations too.

MnemeLlamaIndexMemory ¶

MnemeLlamaIndexMemory(
    manager: MemoryManager,
    *,
    also_persist_episodic: bool = True,
    **kwargs: Any,
)

Bases: BaseMemory

LlamaIndex BaseMemory backed by a Mneme MemoryManager.

Parameters:

Name	Type	Description	Default
`manager`	`MemoryManager`	A constructed :class:`mneme.MemoryManager`. The adapter owns nothing.	required
`also_persist_episodic`	`bool`	If `True` (default), every message also writes to episodic memory so it's searchable via `manager.retrieve()` and visible to consolidation.	`True`

from_defaults `classmethod` ¶

from_defaults(**kwargs: Any) -> MnemeLlamaIndexMemory

LlamaIndex's conventional factory. Requires manager kwarg.

put ¶

put(message: ChatMessage) -> None

Append a chat message to Mneme.

get ¶

get(
    input: str | None = None, **kwargs: Any
) -> list[ChatMessage]

Return recent chat history.

input is accepted for protocol conformance but ignored at v0.5; :func:get returns the same as :func:get_all. A future version may use it as a retrieval query.

set ¶

set(messages: list[ChatMessage]) -> None

Replace working memory with messages.

reset ¶

reset() -> None

Wipe working memory. Episodic survives.

context_for ¶

context_for(
    manager: MemoryManager,
    query: str,
    *,
    k: int = 5,
    token_budget: int | None = None,
    include_working: bool = True,
) -> list[dict[str, str]]

Build a list of OpenAI-style chat messages for an agent prompt.

Parameters:

Name	Type	Description	Default
`manager`	`MemoryManager`	The Mneme manager to draw memory from.	required
`query`	`str`	The user's current question. Used to retrieve relevant episodic + semantic memories.	required
`k`	`int`	How many memories to retrieve. Default 5.	`5`
`token_budget`	`int \| None`	Optional token cap on the returned messages. When given, requires the `tokens` extra (tiktoken). Messages are dropped in priority order: working-memory turns survive last; lower-scored retrieved memories are dropped first.	`None`
`include_working`	`bool`	If `True` (default), append all current working-memory turns as message entries after the retrieved memory.	`True`

Returns:

Type	Description
`list[dict[str, str]]`	A list of `{"role": ..., "content": ...}` dicts ready to pass
`list[dict[str, str]]`	to `client.chat.completions.create(messages=...)`.

API reference¶

Core types¶

MemoryRecord dataclass ¶

touch ¶

WorkingMemory dataclass ¶

EpisodicMemory dataclass ¶

SemanticFact dataclass ¶

RetrievalResult dataclass ¶

MemoryTier ¶

Manager¶

MemoryManager ¶

working instance-attribute ¶

episodic instance-attribute ¶

semantic instance-attribute ¶

retrieve ¶

consolidate ¶

forget ¶

start_scheduler ¶

stop_scheduler ¶

clear_all ¶

Tiers¶

WorkingMemoryTier ¶

add ¶

turns ¶

clear ¶

EpisodicMemoryTier ¶

add ¶

search ¶

clear ¶

SemanticMemoryTier ¶

add ¶

search ¶

Backends¶

MnemeBackend ¶

upsert ¶

get ¶

delete ¶

search ¶

list_recent ¶

touch ¶

count ¶

clear ¶

InMemoryBackend ¶

SQLiteBackend ¶

close ¶

QdrantBackend ¶

PgVectorBackend ¶

close ¶

Embedders¶

EmbeddingProvider ¶

embed ¶

OpenAIEmbeddings ¶

HashEmbedder ¶

LLM judges¶

LLMJudge ¶

complete ¶

OpenAILLMJudge ¶

MockLLMJudge ¶

Adapters¶

MnemeChatMessageHistory ¶

messages property ¶

add_message ¶

clear ¶

MnemeLlamaIndexMemory ¶

from_defaults classmethod ¶

put ¶

get ¶

set ¶

reset ¶

context_for ¶

MemoryRecord `dataclass` ¶

WorkingMemory `dataclass` ¶

EpisodicMemory `dataclass` ¶

SemanticFact `dataclass` ¶

RetrievalResult `dataclass` ¶

working `instance-attribute` ¶

episodic `instance-attribute` ¶

semantic `instance-attribute` ¶

messages `property` ¶

from_defaults `classmethod` ¶