jeevesagent.memory

Memory backends.

The simplest way to pick a backend is to pass a string or URL to Agent(memory=...):

Agent(..., memory="inmemory")            # default; lost on restart
Agent(..., memory="sqlite:./bot.db")     # persistent, no infra
Agent(..., memory="chroma:./vectors")    # local Chroma
Agent(..., memory="postgres://...")      # Postgres + pgvector
Agent(..., memory="redis://...")         # Redis (+ RediSearch)

The resolve_memory() resolver handles the string parsing; it also accepts a config dict ({"backend": "chroma", ...}) and already-constructed Memory instances pass through unchanged.

Backends explicitly:

• InMemoryMemory — dict-backed, no embeddings. Default.

• SqliteMemory — single sqlite file. Persistent, no server.

• VectorMemory — in-process with cosine recall. No persistence.

• ChromaMemory — Chroma client (local persistent or in-process).

• PostgresMemory — Postgres + pgvector + HNSW index.

• RedisMemory — Redis (+ optional RediSearch HNSW vector index).

Embedders in jeevesagent.memory.embedder:

• HashEmbedder — deterministic, zero-key, fine for dev / tests.

• OpenAIEmbedder / VoyageEmbedder / CohereEmbedder — real semantic embeddings.

Submodules

• jeevesagent.memory.auto_extract
• jeevesagent.memory.chroma
• jeevesagent.memory.chroma_facts
• jeevesagent.memory.consolidator
• jeevesagent.memory.embedder
• jeevesagent.memory.facts
• jeevesagent.memory.inmemory
• jeevesagent.memory.lazy
• jeevesagent.memory.postgres
• jeevesagent.memory.postgres_facts
• jeevesagent.memory.redis
• jeevesagent.memory.redis_facts
• jeevesagent.memory.resolver
• jeevesagent.memory.sqlite
• jeevesagent.memory.sqlite_facts
• jeevesagent.memory.vector
• jeevesagent.memory.worker

Classes

AutoExtractMemory	Wraps a Memory and runs auto fact extraction on every
ChromaFactStore	Bi-temporal fact store backed by a Chroma collection.
ChromaMemory	Memory backed by chromadb.
CohereEmbedder	Embeddings via Cohere's cohere SDK.
ConsolidationWorker	Periodic consolidator for any Memory backend.
Consolidator	Wraps a Model to extract Fact rows from episodes.
FactStore	Storage surface for bi-temporal facts.
HashEmbedder	Deterministic SHA256-seeded unit vectors.
InMemoryFactStore	Dict-backed bi-temporal fact store.
InMemoryMemory	Dict-backed implementation of Memory.
LazyMemory	Defer construction of an async-built Memory until first
OpenAIEmbedder	Embeddings via OpenAI's embeddings.create API.
PostgresFactStore	Postgres-backed bi-temporal fact store.
PostgresMemory	Postgres-backed Memory.
RedisFactStore	Bi-temporal fact store over plain Redis hashes.
RedisMemory	Redis-backed Memory. Use connect() to construct.
SqliteFactStore	Durable bi-temporal fact store rooted at a sqlite file.
SqliteMemory	Durable Memory rooted at a single sqlite file.
VectorMemory	Pure-Python embedding-backed Memory.
VoyageEmbedder	Embeddings via Voyage AI's voyageai SDK.

Functions

resolve_memory(→ jeevesagent.core.protocols.Memory)

Resolve a memory= argument into a concrete Memory.

Package Contents

class jeevesagent.memory.AutoExtractMemory(inner: jeevesagent.core.protocols.Memory, consolidator: jeevesagent.memory.consolidator.Consolidator, *, on_extract_error: collections.abc.Callable[[BaseException], collections.abc.Awaitable[None]] | None = None, telemetry: jeevesagent.core.protocols.Telemetry | None = None, auto_picked: bool = False)

Wraps a Memory and runs auto fact extraction on every remember call.

Construct via the Agent auto_extract= kwarg; this class isn’t normally instantiated by user code. The wrapped memory must expose a .facts attribute (a FactStore) for extraction to do anything — when inner.facts is None, the wrapper still installs cleanly but every extraction is a no-op.

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

Persist the episode, then run auto-extraction.

The episode write happens first and is the contract — the function returns its id even when extraction fails. So the consolidator’s fragility never leaks into the agent’s own durability guarantees.

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

property facts: Any

Forward the inner backend’s fact store. Reading this gives callers the same access to the bi-temporal store the consolidator writes into.

property inner: jeevesagent.core.protocols.Memory

The wrapped backend. Power-user introspection — most call sites just use the protocol methods.

class jeevesagent.memory.ChromaFactStore(client: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_FACTS_COLLECTION)

Bi-temporal fact store backed by a Chroma collection.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

classmethod ephemeral(*, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_FACTS_COLLECTION) → ChromaFactStore

classmethod local(persist_directory: str, *, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_FACTS_COLLECTION) → ChromaFactStore

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

property embedder: jeevesagent.core.protocols.Embedder

class jeevesagent.memory.ChromaMemory(client: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_COLLECTION, fact_store: Any | None = None)

Memory backed by chromadb.

Construct via local() for an on-disk persistent client or ephemeral() for a process-local in-memory client.

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

classmethod ephemeral(*, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_COLLECTION, with_facts: bool = False, facts_collection_name: str = 'jeeves_facts') → ChromaMemory

In-memory client (lost on process exit). Great for tests.

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

classmethod local(persist_directory: str, *, embedder: jeevesagent.core.protocols.Embedder | None = None, collection_name: str = DEFAULT_COLLECTION, with_facts: bool = False, facts_collection_name: str = 'jeeves_facts') → ChromaMemory

Persistent on-disk client at persist_directory.

with_facts=True attaches a ChromaFactStore rooted at the same client so facts persist alongside episodes in the same on-disk store.

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

facts: Any | None = None

class jeevesagent.memory.CohereEmbedder(model: str = 'embed-english-v3.0', *, client: Any | None = None, api_key: str | None = None, input_type: str = 'search_document')

Embeddings via Cohere’s cohere SDK.

Models and dimensions:

• embed-english-v3.0 / embed-multilingual-v3.0 -> 1024

• embed-english-light-v3.0 / embed-multilingual-light-v3.0 -> 384

input_type is required by Cohere v3 models:

• "search_document" (default) — corpus / fact-store entries

• "search_query" — retrieval queries

• "classification" / "clustering" for non-retrieval uses

async embed(text: str) → list[float]

async embed_batch(texts: list[str]) → list[list[float]]

dimensions: int

name: str = 'embed-english-v3.0'

class jeevesagent.memory.ConsolidationWorker(memory: jeevesagent.core.protocols.Memory, *, interval_seconds: float = 60.0, on_consolidated: OnConsolidatedCb | None = None, on_error: OnErrorCb | None = None)

Periodic consolidator for any Memory backend.

async run_forever() → None

Sleep interval_seconds then consolidate. Repeat until cancelled.

Spawn this in an anyio.create_task_group() — the cancel scope at scope exit terminates the worker cooperatively.

async run_once() → int

Run a single consolidation pass. Returns the number of new facts extracted (0 when no fact store / nothing changed).

Errors in memory.consolidate() are routed to on_error and not re-raised, so callers can use this in a polling loop without wrapping it in their own try/except.

property iterations: int

Number of consolidate cycles attempted (test introspection).

property total_extracted: int

Cumulative count of facts extracted across all cycles.

class jeevesagent.memory.Consolidator(*, model: jeevesagent.core.protocols.Model, system_prompt: str = DEFAULT_SYSTEM_PROMPT, max_facts_per_episode: int = 20)

Wraps a Model to extract Fact rows from episodes.

async consolidate(episodes: collections.abc.Iterable[jeevesagent.core.types.Episode], *, store: jeevesagent.memory.facts.FactStore) → list[jeevesagent.core.types.Fact]

Process episodes; append extracted facts to store; return the new Fact instances in extraction order.

Uses store.append_many when available so the underlying store can batch the embedder calls (one embed_batch API round-trip instead of N individual embed calls). Falls back to per-fact append for stores that haven’t implemented append_many.

class jeevesagent.memory.FactStore

Bases: Protocol

Storage surface for bi-temporal facts.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

class jeevesagent.memory.HashEmbedder(dimensions: int = DEFAULT_HASH_DIMENSIONS)

Deterministic SHA256-seeded unit vectors.

Each text gets a fresh random.Random seeded by the SHA256 of its UTF-8 bytes, then samples dimensions Gaussian values and L2-normalises the result. Same text always produces the same vector; different texts produce well-distributed vectors with cosine distances that correlate with literal text equality (not semantic similarity).

Use this in tests (fast, no network) and as a default for in-memory backends that need some vector but don’t need real semantic recall.

async embed(text: str) → list[float]

async embed_batch(texts: list[str]) → list[list[float]]

dimensions: int = 384

name: str = 'hash-embedder-384'

class jeevesagent.memory.InMemoryFactStore(*, embedder: jeevesagent.core.protocols.Embedder | None = None)

Dict-backed bi-temporal fact store.

All operations are coordinated by an anyio.Lock so concurrent appends from the consolidator and reads from the agent loop don’t tear the index.

When an embedder is supplied, every appended fact’s triple ("subject predicate object") is embedded and stored alongside the fact, and recall_text() ranks by cosine similarity against the query’s embedding. When no embedder is given, recall_text() falls back to token-overlap matching.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

Append a fact, invalidating any superseded predecessors.

Supersession rule: any existing fact with matching subject + predicate, currently valid (valid_until is None), and a different object gets its valid_until set to the new fact’s valid_from.

async append_many(facts: collections.abc.Iterable[jeevesagent.core.types.Fact]) → list[str]

Append a batch of facts. Embedder calls are coalesced via Embedder.embed_batch() when an embedder is configured — one network round-trip for the batch instead of N.

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

Rank facts against query.

With an embedder configured: cosine-similarity over the query’s embedding vs each fact triple’s stored embedding. Without one: token-overlap with a small stop-word list (longer overlaps win, ties break by shorter haystack = more specific match).

user_id partitions the candidate set as a hard namespace boundary — see Fact for semantics.

snapshot() → dict[str, jeevesagent.core.types.Fact]

property embedder: jeevesagent.core.protocols.Embedder | None

class jeevesagent.memory.InMemoryMemory(*, consolidator: jeevesagent.memory.consolidator.Consolidator | None = None, fact_store: jeevesagent.memory.facts.FactStore | None = None, max_users: int | None = _DEFAULT_MAX_USERS, user_idle_ttl_seconds: float | None = _DEFAULT_USER_TTL_SECONDS)

Dict-backed implementation of Memory.

Multi-tenant accounting (M10): per-user working-block state is held in a bounded LRU + TTL container so a runaway tenant or one-shot user_id explosion can’t grow the in-process dict without limit. Defaults: max_users=100_000 and user_idle_ttl_seconds=86_400 (24h). Pass None to disable bounding for single-tenant or fixed-tenant deployments. Eviction drops a user’s working blocks; callers needing durable spill-to-disk should use SqliteMemory or a SQL-backed memory instead.

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

Process unconsolidated episodes through the configured Consolidator, appending facts to self.facts.

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

Return user/assistant pairs from prior runs of this session.

Materialises each persisted Episode for the given session_id (within the user_id partition) into a [USER input, ASSISTANT output] pair, ordered oldest-first and capped at limit turns total — i.e. up to limit / 2 Q/A exchanges. Tool-call traces are not replayed; the final assistant text per turn is sufficient context for follow-ups.

snapshot() → dict[str, Any]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

facts: jeevesagent.memory.facts.FactStore

class jeevesagent.memory.LazyMemory(builder: collections.abc.Callable[[], collections.abc.Awaitable[Any]], *, description: str = 'memory')

Defer construction of an async-built Memory until first use.

Users rarely instantiate this directly — it’s what the _resolve_memory() resolver returns when given a postgres:// or redis:// URL. Pass a zero-arg async callable that builds the real backend; everything else (working / remember / recall / facts / session_messages / consolidate) is forwarded once that callable resolves.

async aclose() → None

Close the inner backend if it was constructed.

Safe to call when the backend was never resolved (no-op).

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

property description: str

Human-readable label (e.g. "postgres://prod-db/agent") — used in error messages so users can tell which Memory failed to connect.

property facts: Any | None

Direct access to the inner backend’s fact store (if any).

Reading this BEFORE the backend has connected returns None — the connection deliberately hasn’t happened yet. Once the backend is resolved (after the first agent.run or an explicit await mem._resolve()), this returns the live FactStore. Power-user escape hatch; most callers go through recall_facts().

property is_ready: bool

True once the backend has been constructed and cached.

class jeevesagent.memory.OpenAIEmbedder(model: str = 'text-embedding-3-small', *, dimensions: int | None = None, client: Any | None = None, api_key: str | None = None)

Embeddings via OpenAI’s embeddings.create API.

Dimensions are fixed by the model:

• text-embedding-3-small -> 1536

• text-embedding-3-large -> 3072

• text-embedding-ada-002 -> 1536

Pass dimensions= only for text-embedding-3-* models, which support the dimensions parameter for projection.

async embed(text: str) → list[float]

async embed_batch(texts: list[str]) → list[list[float]]

dimensions: int

name: str = 'text-embedding-3-small'

class jeevesagent.memory.PostgresFactStore(pool: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None)

Postgres-backed bi-temporal fact store.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

async append_many(facts: collections.abc.Iterable[jeevesagent.core.types.Fact]) → list[str]

classmethod connect(dsn: str, *, embedder: jeevesagent.core.protocols.Embedder | None = None, min_size: int = 1, max_size: int = 10) → PostgresFactStore

Async:

async init_schema() → None

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

schema_sql() → list[str]

Return the DDL for this fact store’s schema.

Exposed so tests can assert on the SQL strings, and so migration scripts can apply the schema in their own transaction.

property embedder: jeevesagent.core.protocols.Embedder | None

class jeevesagent.memory.PostgresMemory(pool: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None, namespace: str = DEFAULT_NAMESPACE, fact_store: Any | None = None)

Postgres-backed Memory.

pool is an asyncpg.Pool (or anything with the same API). Tests can pass a fake pool whose acquire() returns a fake connection.

async aclose() → None

async append_block(name: str, content: str, *, user_id: str | None = None) → None

classmethod connect(dsn: str, *, embedder: jeevesagent.core.protocols.Embedder | None = None, namespace: str = DEFAULT_NAMESPACE, min_size: int = 1, max_size: int = 10, with_facts: bool = False) → PostgresMemory

Async:

Open an asyncpg pool and register the pgvector codec.

When with_facts=True a PostgresFactStore rooted at the same pool is attached as self.facts so the agent loop’s memory.facts integration just works.

async consolidate() → None

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async init_schema() → None

Apply schema_sql() against the connected pool.

When a PostgresFactStore is attached as self.facts, its schema is initialised in the same call.

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

schema_sql() → list[str]

Return the DDL needed to bootstrap this backend’s schema.

Exposed so tests can assert on the SQL without running it; also usable from migration scripts that want to apply the schema in their own transaction.

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

property embedding_dimensions: int

facts: Any | None = None

property namespace: str

class jeevesagent.memory.RedisFactStore(client: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None, key_prefix: str = DEFAULT_KEY_PREFIX)

Bi-temporal fact store over plain Redis hashes.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

classmethod connect(url: str = 'redis://localhost:6379/0', *, embedder: jeevesagent.core.protocols.Embedder | None = None, key_prefix: str = DEFAULT_KEY_PREFIX) → RedisFactStore

Async:

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

property embedder: jeevesagent.core.protocols.Embedder

class jeevesagent.memory.RedisMemory(client: Any, *, embedder: jeevesagent.core.protocols.Embedder | None = None, key_prefix: str = DEFAULT_KEY_PREFIX, index_name: str = DEFAULT_INDEX_NAME, use_vector_index: bool = True, fact_store: Any | None = None)

Redis-backed Memory. Use connect() to construct.

async aclose() → None

async append_block(name: str, content: str, *, user_id: str | None = None) → None

classmethod connect(url: str = 'redis://localhost:6379/0', *, embedder: jeevesagent.core.protocols.Embedder | None = None, key_prefix: str = DEFAULT_KEY_PREFIX, index_name: str = DEFAULT_INDEX_NAME, use_vector_index: bool = True, with_facts: bool = False, fact_key_prefix: str = 'jeeves:fact:') → RedisMemory

Async:

Open an async Redis connection.

with_facts=True attaches a RedisFactStore sharing the same client; facts go to {fact_key_prefix}* keys so they don’t collide with episode keys.

async consolidate() → None

async ensure_index() → None

Create the RediSearch HNSW index, if not already present.

Skipped silently when use_vector_index=False or when RediSearch isn’t available on the server.

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

facts: Any | None = None

class jeevesagent.memory.SqliteFactStore(path: str | pathlib.Path, *, embedder: jeevesagent.core.protocols.Embedder | None = None)

Durable bi-temporal fact store rooted at a sqlite file.

async aclose() → None

async all_facts() → list[jeevesagent.core.types.Fact]

async append(fact: jeevesagent.core.types.Fact) → str

Append a fact, invalidating any superseded predecessors.

Same supersession rule as InMemoryFactStore: if there’s an existing currently-valid fact with matching subject + predicate but different object, set its valid_until to the new fact’s valid_from.

async query(*, subject: str | None = None, predicate: str | None = None, object_: str | None = None, valid_at: datetime.datetime | None = None, limit: int = 10, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async recall_text(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

property embedder: jeevesagent.core.protocols.Embedder | None

property path: pathlib.Path

class jeevesagent.memory.SqliteMemory(path: str | pathlib.Path, *, embedder: jeevesagent.core.protocols.Embedder | None = None, with_facts: bool = True, fact_store: jeevesagent.memory.facts.FactStore | None = None)

Durable Memory rooted at a single sqlite file.

Construct directly from a path:

memory = SqliteMemory("./bot.db")
agent = Agent("...", model="gpt-4.1-mini", memory=memory)

Or via the resolver:

agent = Agent("...", model="gpt-4.1-mini", memory="sqlite:./bot.db")

Pass path=":memory:" for an ephemeral in-process database (lost on close — useful for tests).

The fact store is auto-attached: the same .db file holds a facts table managed by SqliteFactStore. Pass with_facts=False to skip it; pass an explicit fact_store= to override (e.g. point facts at a different sqlite file).

async aclose() → None

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

property embedder: jeevesagent.core.protocols.Embedder

property path: pathlib.Path

class jeevesagent.memory.VectorMemory(*, embedder: jeevesagent.core.protocols.Embedder | None = None, max_episodes: int | None = None, consolidator: jeevesagent.memory.consolidator.Consolidator | None = None, fact_store: jeevesagent.memory.facts.FactStore | None = None)

Pure-Python embedding-backed Memory.

async append_block(name: str, content: str, *, user_id: str | None = None) → None

async consolidate() → None

Process unconsolidated episodes through the configured Consolidator, appending facts to self.facts.

No-op when no consolidator is configured.

async export(*, user_id: str | None = None) → jeevesagent.core.types.MemoryExport

async forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) → int

async profile(*, user_id: str | None = None) → jeevesagent.core.types.MemoryProfile

async recall(query: str, *, kind: str = 'episodic', limit: int = 5, time_range: tuple[datetime.datetime, datetime.datetime] | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Episode]

async recall_facts(query: str, *, limit: int = 5, valid_at: datetime.datetime | None = None, user_id: str | None = None) → list[jeevesagent.core.types.Fact]

async remember(episode: jeevesagent.core.types.Episode) → str

async session_messages(session_id: str, *, user_id: str | None = None, limit: int = 20) → list[jeevesagent.core.types.Message]

snapshot() → dict[str, Any]

async update_block(name: str, content: str, *, user_id: str | None = None) → None

async working(*, user_id: str | None = None) → list[jeevesagent.core.types.MemoryBlock]

property embedder: jeevesagent.core.protocols.Embedder

facts: jeevesagent.memory.facts.FactStore

class jeevesagent.memory.VoyageEmbedder(model: str = 'voyage-3', *, client: Any | None = None, api_key: str | None = None, input_type: str = 'document')

Embeddings via Voyage AI’s voyageai SDK.

Models and dimensions:

• voyage-3 / voyage-3-large / voyage-code-3 -> 1024

• voyage-3-lite -> 512

input_type controls how Voyage encodes the text:

• "document" (default) — for corpus / fact-store entries

• "query" — for retrieval queries

Pass an explicit input_type= if your embedder is dedicated to one role; for the agent loop’s mixed use (we embed both stored triples and recall queries through the same embedder), the "document" default is the safer choice.

async embed(text: str) → list[float]

async embed_batch(texts: list[str]) → list[list[float]]

dimensions: int

name: str = 'voyage-3'

jeevesagent.memory.resolve_memory(spec: Any) → jeevesagent.core.protocols.Memory

Resolve a memory= argument into a concrete Memory.

None returns the default in-memory backend; a string parses by URL scheme; a dict parses by backend key; anything else is assumed to already be a Memory and passed through unchanged.