Skip to content

Memory clients

The memory API over HTTP: record, recall, supersede and explain.

memory

Agent memory over a Holographic Memory service: remember, recall, forget, explain.

Facts carry optional validity intervals and an event time that is distinct from the time the store ingested them. A newer fact can supersede an older one; the old one is kept as queryable history. Every policy default is conservative: nothing is superseded, rescored, consolidated or forgotten unless you ask.

Times are Unix milliseconds (int) or timezone-aware datetime values; results carry milliseconds, and :func:to_datetime converts them.

__all__ module-attribute

__all__ = ['FactStatus', 'HistoryMode', 'TimeLike', 'MemoryError', 'to_millis', 'to_datetime', 'Remembered', 'ChainLink', 'Explanation', 'Recalled', 'FactRecord', 'ForgetReport', 'ConsolidationReport', 'MemoryClient', 'AsyncMemoryClient']

FactStatus module-attribute

FactStatus = Literal['current', 'superseded', 'expired', 'not-yet-valid']

HistoryMode module-attribute

HistoryMode = Literal['current', 'all', 'historical']

TimeLike module-attribute

TimeLike = int | _dt.datetime

MemoryError

Bases: Exception

The service rejected a memory request. code and status come from its error body.

code instance-attribute

code: str = code

status instance-attribute

status: int = status

__init__

__init__(message: str, code: str = 'invalid_request', status: int = 422) -> None

Remembered dataclass

id instance-attribute

id: str

created instance-attribute

created: bool

False when the identical fact was already stored (an idempotent retry).

supersedes instance-attribute

supersedes: list[str]

superseded_by instance-attribute

superseded_by: list[str]

ingest_time instance-attribute

ingest_time: int

__init__

__init__(id: str, created: bool, supersedes: list[str], superseded_by: list[str], ingest_time: int) -> None

id instance-attribute

id: str

relation instance-attribute

relation: Literal['predecessor', 'successor']

effective_from instance-attribute

effective_from: int

ingest_time instance-attribute

ingest_time: int

__init__

__init__(id: str, relation: Literal['predecessor', 'successor'], effective_from: int, ingest_time: int) -> None

Explanation dataclass

Why a fact matched and how it was adjusted.

retrieved instance-attribute

retrieved: bool

rank instance-attribute

rank: int | None

lexical_score instance-attribute

lexical_score: float | None

semantic_score instance-attribute

semantic_score: float | None

retrieval_score instance-attribute

retrieval_score: float | None

lexical_weight instance-attribute

lexical_weight: float

semantic_weight instance-attribute

semantic_weight: float

fusion instance-attribute

fusion: str

matched_terms instance-attribute

matched_terms: list[str]

recency_factor instance-attribute

recency_factor: float

historical_factor instance-attribute

historical_factor: float

final_score instance-attribute

final_score: float | None

status instance-attribute

status: FactStatus

as_of instance-attribute

as_of: int

known_at instance-attribute

known_at: int | None

valid_from instance-attribute

valid_from: int | None

valid_to instance-attribute

valid_to: int | None

within_validity instance-attribute

within_validity: bool

chain instance-attribute

chain: list[ChainLink]

__init__

__init__(retrieved: bool, rank: int | None, lexical_score: float | None, semantic_score: float | None, retrieval_score: float | None, lexical_weight: float, semantic_weight: float, fusion: str, matched_terms: list[str], recency_factor: float, historical_factor: float, final_score: float | None, status: FactStatus, as_of: int, known_at: int | None, valid_from: int | None, valid_to: int | None, within_validity: bool, chain: list[ChainLink]) -> None

Recalled dataclass

id instance-attribute

id: str

text instance-attribute

text: str | None

score instance-attribute

score: float

status instance-attribute

status: FactStatus

ingest_time instance-attribute

ingest_time: int | None

event_time instance-attribute

event_time: int | None

valid_from instance-attribute

valid_from: int | None

valid_to instance-attribute

valid_to: int | None

subject instance-attribute

subject: str | None

entities instance-attribute

entities: list[str]

supersedes instance-attribute

supersedes: list[str]

superseded_by instance-attribute

superseded_by: list[str]

metadata instance-attribute

metadata: dict[str, Any]

source_uri instance-attribute

source_uri: str | None

explanation class-attribute instance-attribute

explanation: Explanation | None = None

__init__

__init__(id: str, text: str | None, score: float, status: FactStatus, ingest_time: int | None, event_time: int | None, valid_from: int | None, valid_to: int | None, subject: str | None, entities: list[str], supersedes: list[str], superseded_by: list[str], metadata: dict[str, Any], source_uri: str | None, explanation: Explanation | None = None) -> None

FactRecord dataclass

A stored fact with its history links, as get and history return it.

id instance-attribute

id: str

text instance-attribute

text: str | None

status instance-attribute

status: FactStatus

ingest_time instance-attribute

ingest_time: int | None

event_time instance-attribute

event_time: int | None

valid_from instance-attribute

valid_from: int | None

valid_to instance-attribute

valid_to: int | None

effective_from instance-attribute

effective_from: int

subject instance-attribute

subject: str | None

entities instance-attribute

entities: list[str]

supersedes instance-attribute

supersedes: list[str]

superseded_by instance-attribute

superseded_by: list[str]

metadata instance-attribute

metadata: dict[str, Any]

__init__

__init__(id: str, text: str | None, status: FactStatus, ingest_time: int | None, event_time: int | None, valid_from: int | None, valid_to: int | None, effective_from: int, subject: str | None, entities: list[str], supersedes: list[str], superseded_by: list[str], metadata: dict[str, Any]) -> None

ForgetReport dataclass

forgotten instance-attribute

forgotten: list[str]

dry_run instance-attribute

dry_run: bool

purged instance-attribute

purged: bool

__init__

__init__(forgotten: list[str], dry_run: bool, purged: bool) -> None

ConsolidationReport dataclass

created instance-attribute

created: list[Remembered]

forgotten instance-attribute

forgotten: list[str]

dry_run instance-attribute

dry_run: bool

__init__

__init__(created: list[Remembered], forgotten: list[str], dry_run: bool) -> None

MemoryClient

Memory operations over a :class:HolographicClient.

memory = MemoryClient(HolographicClient("http://localhost:8080")) memory.remember("Alice lives in Paris", subject="alice/city", valid_from=2020_01_01)

__init__

__init__(client: HolographicClient) -> None

connect classmethod

connect(url: str = 'http://localhost:8080', **options: Any) -> MemoryClient

A client over a new :class:HolographicClient; options are its arguments.

close

close() -> None

__enter__

__enter__() -> MemoryClient

__exit__

__exit__(*exc: Any) -> None

remember

remember(text: str, *, id: str | None = None, event_time: TimeLike | None = None, valid_from: TimeLike | None = None, valid_to: TimeLike | None = None, subject: str | None = None, supersedes: Sequence[str] | None = None, entities: Sequence[str] | None = None, metadata: Mapping[str, Any] | None = None, embedding: Sequence[float] | None = None, source_uri: str | None = None) -> Remembered

Store one fact. Retrying the same call is safe: it reports created=False.

remember_many

remember_many(facts: Iterable[Mapping[str, Any]]) -> list[Remembered]

Store facts atomically. Each mapping takes remember's keyword names plus text.

recall

recall(query: str, *, k: int | None = None, as_of: TimeLike | None = None, known_at: TimeLike | None = None, history: HistoryMode | None = None, subject: str | None = None, entities: Sequence[str] | None = None, filter: Mapping[str, Any] | None = None, embedding: Sequence[float] | None = None, lexical_weight: float | None = None, semantic_weight: float | None = None, min_semantic_score: float | None = None, fusion: Literal['blend', 'rrf'] | None = None, candidate_limit: int | None = None, explain: bool = False) -> list[Recalled]

Facts relevant to query, best first. Only facts current now unless history says otherwise.

explain

explain(query: str, fact_id: str, *, as_of: TimeLike | None = None, known_at: TimeLike | None = None, embedding: Sequence[float] | None = None, lexical_weight: float | None = None, semantic_weight: float | None = None, fusion: Literal['blend', 'rrf'] | None = None, candidate_limit: int | None = None) -> Explanation

Why fact_id does or does not come back for query.

forget

forget(*, ids: Iterable[str] | None = None, subject: str | None = None, entity: str | None = None, ingested_before: TimeLike | None = None, event_before: TimeLike | None = None, matching: Mapping[str, Any] | None = None, include_chain: bool = False, dry_run: bool = False, purge: bool = False) -> ForgetReport

Remove facts by exactly one scope. Repeating a call removes nothing more.

include_chain also removes every version linked by supersession; purge compacts the store afterwards so deleted text leaves the journal; dry_run only reports.

get

get(fact_id: str) -> FactRecord | None

history

history(fact_id: str) -> list[FactRecord]

Every version of the fact's supersession chain, in effect order.

apply_forgetting

apply_forgetting(*, include_chain: bool = False, dry_run: bool = True, purge: bool = False) -> ForgetReport

Run the service's configured forgetting policy. Defaults to a dry run; the default policy selects nothing.

consolidate

consolidate(*, dry_run: bool = True, allow_destructive: bool = False) -> ConsolidationReport

Run the service's configured consolidation policy. Defaults to a dry run; the default policy plans nothing.

AsyncMemoryClient

Async counterpart of :class:MemoryClient over an :class:AsyncHolographicClient.

__init__

__init__(client: AsyncHolographicClient) -> None

connect classmethod

connect(url: str = 'http://localhost:8080', **options: Any) -> AsyncMemoryClient

aclose async

aclose() -> None

__aenter__ async

__aenter__() -> AsyncMemoryClient

__aexit__ async

__aexit__(*exc: Any) -> None

remember async

remember(text: str, *, id: str | None = None, event_time: TimeLike | None = None, valid_from: TimeLike | None = None, valid_to: TimeLike | None = None, subject: str | None = None, supersedes: Sequence[str] | None = None, entities: Sequence[str] | None = None, metadata: Mapping[str, Any] | None = None, embedding: Sequence[float] | None = None, source_uri: str | None = None) -> Remembered

remember_many async

remember_many(facts: Iterable[Mapping[str, Any]]) -> list[Remembered]

recall async

recall(query: str, *, k: int | None = None, as_of: TimeLike | None = None, known_at: TimeLike | None = None, history: HistoryMode | None = None, subject: str | None = None, entities: Sequence[str] | None = None, filter: Mapping[str, Any] | None = None, embedding: Sequence[float] | None = None, lexical_weight: float | None = None, semantic_weight: float | None = None, min_semantic_score: float | None = None, fusion: Literal['blend', 'rrf'] | None = None, candidate_limit: int | None = None, explain: bool = False) -> list[Recalled]

explain async

explain(query: str, fact_id: str, *, as_of: TimeLike | None = None, known_at: TimeLike | None = None, embedding: Sequence[float] | None = None, lexical_weight: float | None = None, semantic_weight: float | None = None, fusion: Literal['blend', 'rrf'] | None = None, candidate_limit: int | None = None) -> Explanation

forget async

forget(*, ids: Iterable[str] | None = None, subject: str | None = None, entity: str | None = None, ingested_before: TimeLike | None = None, event_before: TimeLike | None = None, matching: Mapping[str, Any] | None = None, include_chain: bool = False, dry_run: bool = False, purge: bool = False) -> ForgetReport

get async

get(fact_id: str) -> FactRecord | None

history async

history(fact_id: str) -> list[FactRecord]

apply_forgetting async

apply_forgetting(*, include_chain: bool = False, dry_run: bool = True, purge: bool = False) -> ForgetReport

consolidate async

consolidate(*, dry_run: bool = True, allow_destructive: bool = False) -> ConsolidationReport

to_millis

to_millis(value: TimeLike | None) -> int | None

Unix milliseconds for an int (returned unchanged) or a datetime (naive means UTC).

to_datetime

to_datetime(millis: int) -> datetime

The UTC datetime of a Unix-millisecond time.