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

.. py:module:: jeevesagent.memory

.. autoapi-nested-parse::

   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 :func:`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:

   * :class:`InMemoryMemory` — dict-backed, no embeddings. Default.
   * :class:`SqliteMemory` — single sqlite file. Persistent, no server.
   * :class:`VectorMemory` — in-process with cosine recall. No persistence.
   * :class:`ChromaMemory` — Chroma client (local persistent or in-process).
   * :class:`PostgresMemory` — Postgres + pgvector + HNSW index.
   * :class:`RedisMemory` — Redis (+ optional RediSearch HNSW vector index).

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

   * :class:`HashEmbedder` — deterministic, zero-key, fine for dev / tests.
   * :class:`OpenAIEmbedder` / :class:`VoyageEmbedder` /
     :class:`CohereEmbedder` — real semantic embeddings.



Submodules
----------

.. toctree::
   :maxdepth: 1

   /api/jeevesagent/memory/auto_extract/index
   /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/lazy/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/resolver/index
   /api/jeevesagent/memory/sqlite/index
   /api/jeevesagent/memory/sqlite_facts/index
   /api/jeevesagent/memory/vector/index
   /api/jeevesagent/memory/worker/index


Classes
-------

.. autoapisummary::

   jeevesagent.memory.AutoExtractMemory
   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.LazyMemory
   jeevesagent.memory.OpenAIEmbedder
   jeevesagent.memory.PostgresFactStore
   jeevesagent.memory.PostgresMemory
   jeevesagent.memory.RedisFactStore
   jeevesagent.memory.RedisMemory
   jeevesagent.memory.SqliteFactStore
   jeevesagent.memory.SqliteMemory
   jeevesagent.memory.VectorMemory
   jeevesagent.memory.VoyageEmbedder


Functions
---------

.. autoapisummary::

   jeevesagent.memory.resolve_memory


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

.. py:class:: 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 :class:`Memory` and runs auto fact extraction on every
   ``remember`` call.

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


   .. py:method:: append_block(name: str, content: str, *, user_id: str | None = None) -> None
      :async:



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



   .. py:method:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



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


      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.



   .. 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, *, user_id: str | None = None) -> None
      :async:



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



   .. py:property:: facts
      :type: Any


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


   .. py:property:: inner
      :type: jeevesagent.core.protocols.Memory


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


.. 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, *, user_id: str | None = None) -> 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:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. 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:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



   .. py:method:: working(*, user_id: str | None = None) -> 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, max_users: int | None = _DEFAULT_MAX_USERS, user_idle_ttl_seconds: float | None = _DEFAULT_USER_TTL_SECONDS)

   Dict-backed implementation of :class:`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 :class:`SqliteMemory` or a
   SQL-backed memory instead.


   .. py:method:: append_block(name: str, content: str, *, user_id: str | None = None) -> None
      :async:



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


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



   .. py:method:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



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



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


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

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

   Users rarely instantiate this directly — it's what the
   :func:`_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.


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


      Close the inner backend if it was constructed.

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



   .. py:method:: append_block(name: str, content: str, *, user_id: str | None = None) -> None
      :async:



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



   .. py:method:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



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



   .. py:property:: description
      :type: str


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


   .. py:property:: facts
      :type: 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 :meth:`recall_facts`.


   .. py:property:: is_ready
      :type: bool


      ``True`` once the backend has been constructed and cached.


.. 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, *, user_id: str | None = None) -> 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:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :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:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



   .. py:method:: working(*, user_id: str | None = None) -> 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, *, user_id: str | None = None) -> 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:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



   .. py:method:: working(*, user_id: str | None = None) -> 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:: 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 :class:`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 :class:`SqliteFactStore`. Pass
   ``with_facts=False`` to skip it; pass an explicit
   ``fact_store=`` to override (e.g. point facts at a different
   sqlite file).


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



   .. py:method:: append_block(name: str, content: str, *, user_id: str | None = None) -> None
      :async:



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



   .. py:method:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



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



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



   .. 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, *, user_id: str | None = None) -> 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:: export(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryExport
      :async:



   .. py:method:: forget(*, user_id: str | None = None, session_id: str | None = None, before: datetime.datetime | None = None) -> int
      :async:



   .. py:method:: profile(*, user_id: str | None = None) -> jeevesagent.core.types.MemoryProfile
      :async:



   .. 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, *, user_id: str | None = None) -> None
      :async:



   .. py:method:: working(*, user_id: str | None = None) -> 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'



.. py:function:: resolve_memory(spec: Any) -> jeevesagent.core.protocols.Memory

   Resolve a ``memory=`` argument into a concrete :class:`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 :class:`Memory` and passed through
   unchanged.


