jeevesagent.memory
==================

.. py:module:: jeevesagent.memory

.. autoapi-nested-parse::

   Memory backends.

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

   * :class:`InMemoryMemory` — naive dict-backed, no embeddings. The
     default; great for tests and tiny demos.
   * :class:`VectorMemory` — in-memory with embedding-based cosine
     recall. Pure Python, no infrastructure. Scales to a few thousand
     episodes.
   * :class:`ChromaMemory` — backed by Chroma (local persistent or
     in-memory client). Lazy ``chromadb`` import.
   * :class:`PostgresMemory` — Postgres + pgvector with HNSW index.
     Lazy ``asyncpg`` + ``pgvector`` imports. Production-grade.
   * :class:`RedisMemory` — Redis with optional RediSearch HNSW
     vector index, falling back to brute-force when RediSearch isn't
     available. Lazy ``redis`` import.

   Embedders live in :mod:`jeevesagent.memory.embedder`:

   * :class:`HashEmbedder` — deterministic, zero-dep, perfect for tests.
   * :class:`OpenAIEmbedder` — real semantic embeddings via OpenAI.



Submodules
----------

.. toctree::
   :maxdepth: 1

   /api/jeevesagent/memory/chroma/index
   /api/jeevesagent/memory/chroma_facts/index
   /api/jeevesagent/memory/consolidator/index
   /api/jeevesagent/memory/embedder/index
   /api/jeevesagent/memory/facts/index
   /api/jeevesagent/memory/inmemory/index
   /api/jeevesagent/memory/postgres/index
   /api/jeevesagent/memory/postgres_facts/index
   /api/jeevesagent/memory/redis/index
   /api/jeevesagent/memory/redis_facts/index
   /api/jeevesagent/memory/sqlite_facts/index
   /api/jeevesagent/memory/vector/index
   /api/jeevesagent/memory/worker/index


Classes
-------

.. autoapisummary::

   jeevesagent.memory.ChromaFactStore
   jeevesagent.memory.ChromaMemory
   jeevesagent.memory.CohereEmbedder
   jeevesagent.memory.ConsolidationWorker
   jeevesagent.memory.Consolidator
   jeevesagent.memory.FactStore
   jeevesagent.memory.HashEmbedder
   jeevesagent.memory.InMemoryFactStore
   jeevesagent.memory.InMemoryMemory
   jeevesagent.memory.OpenAIEmbedder
   jeevesagent.memory.PostgresFactStore
   jeevesagent.memory.PostgresMemory
   jeevesagent.memory.RedisFactStore
   jeevesagent.memory.RedisMemory
   jeevesagent.memory.SqliteFactStore
   jeevesagent.memory.VectorMemory
   jeevesagent.memory.VoyageEmbedder


Package Contents
----------------

.. py:class:: 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.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:



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



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



   .. py:method:: 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:



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



   .. py:property:: embedder
      :type: jeevesagent.core.protocols.Embedder



.. py:class:: 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 :meth:`local` for an on-disk persistent client or
   :meth:`ephemeral` for a process-local in-memory client.


   .. py:method:: append_block(name: str, content: str) -> None
      :async:



   .. py:method:: consolidate() -> None
      :async:



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


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



   .. py:method:: 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
      :classmethod:


      Persistent on-disk client at ``persist_directory``.

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



   .. py:method:: 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:



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



   .. py:method:: remember(episode: jeevesagent.core.types.Episode) -> str
      :async:



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



   .. py:method:: update_block(name: str, content: str) -> None
      :async:



   .. py:method:: working() -> list[jeevesagent.core.types.MemoryBlock]
      :async:



   .. py:attribute:: facts
      :type:  Any | None
      :value: None



.. py:class:: 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


   .. py:method:: embed(text: str) -> list[float]
      :async:



   .. py:method:: embed_batch(texts: list[str]) -> list[list[float]]
      :async:



   .. py:attribute:: dimensions
      :type:  int


   .. py:attribute:: name
      :type:  str
      :value: 'embed-english-v3.0'



.. py:class:: 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 :class:`Memory` backend.


   .. py:method:: run_forever() -> None
      :async:


      Sleep ``interval_seconds`` then consolidate. Repeat until
      cancelled.

      Spawn this in an :func:`anyio.create_task_group` — the cancel
      scope at scope exit terminates the worker cooperatively.



   .. py:method:: run_once() -> int
      :async:


      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.



   .. py:property:: iterations
      :type: int


      Number of consolidate cycles attempted (test introspection).


   .. py:property:: total_extracted
      :type: int


      Cumulative count of facts extracted across all cycles.


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

   Wraps a :class:`Model` to extract :class:`Fact` rows from episodes.


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


      Process ``episodes``; append extracted facts to ``store``;
      return the new :class:`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``.



.. py:class:: FactStore

   Bases: :py:obj:`Protocol`


   Storage surface for bi-temporal facts.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:



   .. py:method:: 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:



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



.. py:class:: 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.


   .. py:method:: embed(text: str) -> list[float]
      :async:



   .. py:method:: embed_batch(texts: list[str]) -> list[list[float]]
      :async:



   .. py:attribute:: dimensions
      :type:  int
      :value: 384



   .. py:attribute:: name
      :type:  str
      :value: 'hash-embedder-384'



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

   Dict-backed bi-temporal fact store.

   All operations are coordinated by an :class:`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 :meth:`recall_text` ranks by cosine similarity
   against the query's embedding. When no embedder is given,
   :meth:`recall_text` falls back to token-overlap matching.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:


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



   .. py:method:: append_many(facts: collections.abc.Iterable[jeevesagent.core.types.Fact]) -> list[str]
      :async:


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



   .. py:method:: 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:



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


      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 :class:`Fact` for semantics.



   .. py:method:: snapshot() -> dict[str, jeevesagent.core.types.Fact]


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



.. py:class:: InMemoryMemory(*, consolidator: jeevesagent.memory.consolidator.Consolidator | None = None, fact_store: jeevesagent.memory.facts.FactStore | None = None)

   Dict-backed implementation of :class:`Memory`.


   .. py:method:: append_block(name: str, content: str) -> None
      :async:



   .. py:method:: consolidate() -> None
      :async:


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



   .. py:method:: 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:



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



   .. py:method:: remember(episode: jeevesagent.core.types.Episode) -> str
      :async:



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


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

      Materialises each persisted :class:`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.



   .. py:method:: snapshot() -> dict[str, Any]


   .. py:method:: update_block(name: str, content: str) -> None
      :async:



   .. py:method:: working() -> list[jeevesagent.core.types.MemoryBlock]
      :async:



   .. py:attribute:: facts
      :type:  jeevesagent.memory.facts.FactStore


.. py:class:: 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.


   .. py:method:: embed(text: str) -> list[float]
      :async:



   .. py:method:: embed_batch(texts: list[str]) -> list[list[float]]
      :async:



   .. py:attribute:: dimensions
      :type:  int


   .. py:attribute:: name
      :type:  str
      :value: 'text-embedding-3-small'



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

   Postgres-backed bi-temporal fact store.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:



   .. py:method:: append_many(facts: collections.abc.Iterable[jeevesagent.core.types.Fact]) -> list[str]
      :async:



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

      :async:



   .. py:method:: init_schema() -> None
      :async:



   .. py:method:: 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:



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



   .. py:method:: 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.



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



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

   Postgres-backed :class:`Memory`.

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


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: append_block(name: str, content: str) -> None
      :async:



   .. py:method:: 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
      :classmethod:

      :async:


      Open an asyncpg pool and register the pgvector codec.

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



   .. py:method:: consolidate() -> None
      :async:



   .. py:method:: init_schema() -> None
      :async:


      Apply :meth:`schema_sql` against the connected pool.

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



   .. py:method:: 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:



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



   .. py:method:: remember(episode: jeevesagent.core.types.Episode) -> str
      :async:



   .. py:method:: 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.



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



   .. py:method:: update_block(name: str, content: str) -> None
      :async:



   .. py:method:: working() -> list[jeevesagent.core.types.MemoryBlock]
      :async:



   .. py:property:: embedding_dimensions
      :type: int



   .. py:attribute:: facts
      :type:  Any | None
      :value: None



   .. py:property:: namespace
      :type: str



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

   Bi-temporal fact store over plain Redis hashes.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:



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

      :async:



   .. py:method:: 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:



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



   .. py:property:: embedder
      :type: jeevesagent.core.protocols.Embedder



.. py:class:: 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 :class:`Memory`. Use :meth:`connect` to construct.


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: append_block(name: str, content: str) -> None
      :async:



   .. py:method:: 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
      :classmethod:

      :async:


      Open an async Redis connection.

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



   .. py:method:: consolidate() -> None
      :async:



   .. py:method:: ensure_index() -> None
      :async:


      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.



   .. py:method:: 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:



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



   .. py:method:: remember(episode: jeevesagent.core.types.Episode) -> str
      :async:



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



   .. py:method:: update_block(name: str, content: str) -> None
      :async:



   .. py:method:: working() -> list[jeevesagent.core.types.MemoryBlock]
      :async:



   .. py:attribute:: facts
      :type:  Any | None
      :value: None



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

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


   .. py:method:: aclose() -> None
      :async:



   .. py:method:: all_facts() -> list[jeevesagent.core.types.Fact]
      :async:



   .. py:method:: append(fact: jeevesagent.core.types.Fact) -> str
      :async:


      Append a fact, invalidating any superseded predecessors.

      Same supersession rule as :class:`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``.



   .. py:method:: 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:



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



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



   .. py:property:: path
      :type: pathlib.Path



.. py:class:: 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 :class:`Memory`.


   .. py:method:: append_block(name: str, content: str) -> None
      :async:



   .. py:method:: consolidate() -> None
      :async:


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

      No-op when no consolidator is configured.



   .. py:method:: 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:



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



   .. py:method:: remember(episode: jeevesagent.core.types.Episode) -> str
      :async:



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



   .. py:method:: snapshot() -> dict[str, Any]


   .. py:method:: update_block(name: str, content: str) -> None
      :async:



   .. py:method:: working() -> list[jeevesagent.core.types.MemoryBlock]
      :async:



   .. py:property:: embedder
      :type: jeevesagent.core.protocols.Embedder



   .. py:attribute:: facts
      :type:  jeevesagent.memory.facts.FactStore


.. py:class:: 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.


   .. py:method:: embed(text: str) -> list[float]
      :async:



   .. py:method:: embed_batch(texts: list[str]) -> list[list[float]]
      :async:



   .. py:attribute:: dimensions
      :type:  int


   .. py:attribute:: name
      :type:  str
      :value: 'voyage-3'



