jeevesagent.memory

Memory backends.

Pick one of these and pass it to Agent(..., memory=...):

• InMemoryMemory — naive dict-backed, no embeddings. The default; great for tests and tiny demos.

• VectorMemory — in-memory with embedding-based cosine recall. Pure Python, no infrastructure. Scales to a few thousand episodes.

• ChromaMemory — backed by Chroma (local persistent or in-memory client). Lazy chromadb import.

• PostgresMemory — Postgres + pgvector with HNSW index. Lazy asyncpg + pgvector imports. Production-grade.

• RedisMemory — Redis with optional RediSearch HNSW vector index, falling back to brute-force when RediSearch isn’t available. Lazy redis import.

Embedders live in jeevesagent.memory.embedder:

• HashEmbedder — deterministic, zero-dep, perfect for tests.

• OpenAIEmbedder — real semantic embeddings via OpenAI.

Submodules

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

Classes

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.
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.
VectorMemory	Pure-Python embedding-backed Memory.
VoyageEmbedder	Embeddings via Voyage AI's voyageai SDK.

Package Contents

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

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 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) → None

async working() → 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)

Dict-backed implementation of Memory.

async append_block(name: str, content: str) → None

async consolidate() → None

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

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) → None

async working() → list[jeevesagent.core.types.MemoryBlock]

facts: jeevesagent.memory.facts.FactStore

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) → 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 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 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) → None

async working() → 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) → 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 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) → None

async working() → 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.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) → None

async consolidate() → None

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

No-op when no consolidator is configured.

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) → None

async working() → 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'