jeevesagent.memory.facts

Bi-temporal fact store.

The store holds Fact instances — semantic (subject, predicate, object) claims extracted from episodes by a Consolidator.

Bi-temporal contract:

• valid_from / valid_until are when the fact was true in the world. valid_until = None means “still valid now”.

• recorded_at is when we learned the fact (when the consolidator ran).

On InMemoryFactStore.append(), conflicts are resolved by supersession: if there’s an existing currently-valid fact with the same (subject, predicate) but different object, its valid_until is set to the new fact’s valid_from. This is the Zep-style temporal graph behaviour — old beliefs aren’t deleted, they get “closed off” so we can still reason about what was true at any historical moment.

Today’s only backend is InMemoryFactStore. Postgres / sqlite fact stores are a follow-up — the protocol is stable.

Classes

FactStore	Storage surface for bi-temporal facts.
InMemoryFactStore	Dict-backed bi-temporal fact store.

Module Contents

class jeevesagent.memory.facts.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.facts.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