Reference

Library API reference

Every public class and method in outhad-contextkitai 0.1.4, generated from the release on PyPI.

Generated from outhad-contextkitai 0.1.4, the latest release on PyPI. This page is rebuilt from the published wheel after every release, so it always matches what pip install gives you.

Memory

from outhad_contextkit.memory.main import Memory

`init`

def __init__(config: MemoryConfig = MemoryConfig())

No description.

`from_config` classmethod

def from_config(config_dict: Dict[str, Any])

No description.

`decay_score`

def decay_score(memory_id: str) -> Optional[float]

Phase D2 — return the computed decay score for one memory.

Returns None when decay-v2 is disabled or the memory is unknown.

`archive_low`

def archive_low(
    *,
    threshold: Optional[float] = None,
    dry_run: bool = False,
) -> Dict[str, Any]

Phase D2 — archive every node whose decay_score is below threshold.

threshold defaults to decay_v2.archive_threshold. dry_run=True returns the candidate ids without mutation so operators can audit before committing.

`start_decay_scheduler`

def start_decay_scheduler() -> None

Phase D3 — start the periodic decay scheduler (idempotent).

`stop_decay_scheduler`

def stop_decay_scheduler(*, timeout: float = 5.0) -> None

Phase D3 — stop the scheduler thread (idempotent).

`close`

def close() -> None

Idempotent cleanup. Stops the decay scheduler if running.

`record_reference`

def record_reference(
    memory_id: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    strength: float = 1.0,
) -> Optional[int]

Phase D6 — bump access_count + last_accessed_at for a memory the agent actually used downstream.

Returns the new access count, or None when the lifecycle subsystem is disabled.

`versions`

def versions(memory_id: str) -> List[Any]

Phase D5 — return the version chain (oldest → newest).

`latest_version`

def latest_version(memory_id: str) -> Optional[Any]

Phase D5 — return the newest non-superseded version, or None.

`diff_versions`

def diff_versions(old_id: str, new_id: str, *, mode: Optional[str] = None) -> str

Phase D5 — diff two memory versions.

mode defaults to decay_v2.versioning.diff_provider — "unified" (text), "json" (JSON-Patch), or "none".

`rollback`

def rollback(
    memory_id: str,
    *,
    to_version: int,
    actor_id: Optional[str] = None,
) -> Optional[str]

Phase D5 — append a new version whose payload equals version=to_version's payload.

Returns the new memory_id or None when versioning is off / the requested version cannot be found.

`demote_to_cold`

def demote_to_cold(memory_id: str) -> bool

Phase D7 — move memory_id to cold storage; remove hot copies.

`promote_from_cold`

def promote_from_cold(memory_id: str) -> Optional[Dict[str, Any]]

Phase D7 — pull payload back from cold storage and re-insert.

`backfill_decay_v2`

def backfill_decay_v2() -> Dict[str, int]

Phase D8 — recompute decay scores + run archive_low once.

Idempotent. Useful right after enabling decay_v2 on an existing deployment to bring archived state in line with the new thresholds.

`record_feedback`

def record_feedback(
    memory_id: str,
    *,
    helpful: bool,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    query: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]

Record user feedback on a retrieved memory.

Returns {'memory_id', 'delta_applied', 'new_relevance', 'event_id', 'helpful'} on success. Returns None when MSPR / CGL / feedback is disabled or the memory is unknown.

Side effects:

Persists a FeedbackEvent to feedback_events.db.
Bumps MemoryNode.relevance via :meth:ContextGraph.bump_relevance (floor pinned on helpful when FeedbackConfig.pin_floor_on_boost=True).
Clamps the per-memory cumulative boost at FeedbackConfig.max_boost_per_memory so repeated helpful verdicts cannot snowball past the cap.
Increments helpful_count / unhelpful_count + writes last_feedback_at on the node.
Emits a memory_feedback_recorded change-bus event.

`reset_feedback`

def reset_feedback(*, user_id: Optional[str] = None) -> int

Delete feedback rows (optionally scoped to one user).

Returns the number of rows deleted. No-op if MSPR is disabled.

`set_role_policy`

def set_role_policy(policy: Optional[Any]) -> None

Install a :class:RolePolicy for role/tenant-aware retrieval.

Pass None to disable role-based filtering. The policy is consulted by _graph_first_rerank when mspr.role.enabled is True. Pydantic config stays JSON-serialisable because the callable lives off-config on the Memory instance.

`get_role_policy`

def get_role_policy() -> Optional[Any]

Return the currently installed RolePolicy or None.

`backfill_mspr`

def backfill_mspr() -> Dict[str, int]

Phase F8 — idempotent backfill of MSPR-derived tables.

Today this rebuilds query_success from feedback_events when an operator upgrades from a pre-F7 build. Future phases can hang additional backfills off the same entrypoint.

Returns a counts dict {'success_rows': N, 'feedback_rows': M} for logging / progress display. Raises RuntimeError when MSPR is disabled — by design, so silent no-ops on a misconfigured Memory don't mask a typo.

`backfill_tenant`

def backfill_tenant(
    *,
    default_tenant_id: Optional[str] = None,
    batch_size: int = 500,
) -> Dict[str, int]

Phase T8 — stamp legacy memory + CGL nodes with the default tenant.

Idempotent. Returns {"history": N, "nodes": M, "edges": K}. Raises RuntimeError when tenant.enabled=False.

`export_tenant`

def export_tenant(tenant_id: str, dest_dir: str) -> Dict[str, int]

Phase T7 — write every storage row for tenant_id to disk.

`import_tenant`

def import_tenant(src_dir: str) -> Dict[str, int]

Phase T7 — reverse of :meth:export_tenant.

`migrate_tenant`

def migrate_tenant(src_id: str, dst_id: str) -> Dict[str, int]

Phase T7 — rewrite every storage row from src_id to dst_id.

`add`

def add(
    messages,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    infer: bool = True,
    memory_type: Optional[str] = None,
    prompt: Optional[str] = None,
)

Create a new memory.

Adds new memories scoped to a single session id (e.g. user_id, agent_id, or run_id). One of those ids is required.

Args

messages str or List[Dict[str, str]]: The message content or list of messages (e.g., [{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi"}]) to be processed and stored.
user_id str, optional: ID of the user creating the memory. Defaults to None.
agent_id str, optional: ID of the agent creating the memory. Defaults to None.
run_id str, optional: ID of the run creating the memory. Defaults to None.
metadata dict, optional: Metadata to store with the memory. Defaults to None.
infer bool, optional: If True (default), an LLM is used to extract key facts from 'messages' and decide whether to add, update, or delete related memories. If False, 'messages' are added as raw memories directly.
memory_type str, optional: Specifies the type of memory. Currently, only MemoryType.PROCEDURAL.value ("procedural_memory") is explicitly handled for creating procedural memories (typically requires 'agent_id'). Otherwise, memories are treated as general conversational/factual memories.memory_type (str, optional): Type of memory to create. Defaults to None. By default, it creates the short term memories and long term (semantic and episodic) memories. Pass "procedural_memory" to create procedural memories.
prompt str, optional: Prompt to use for the memory creation. Defaults to None.

Returns

dict: A dictionary containing the result of the memory addition operation, typically

including a list of memory items affected (added, updated) under a "results" key,
and potentially "relations" if graph store is enabled.
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "event": "ADD"}]}`

`get`

def get(memory_id)

Retrieve a memory by ID.

Args

memory_id str: ID of the memory to retrieve.

Returns

dict: Retrieved memory.

`get_all`

def get_all(
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    filters: Optional[Dict[str, Any]] = None,
    limit: int = 100,
)

List all memories.

Args

user_id str, optional: user id
agent_id str, optional: agent id
run_id str, optional: run id
filters dict, optional: Additional custom key-value filters to apply to the search. These are merged with the ID-based scoping filters. For example, filters={"actor_id": "some_user"}.
limit int, optional: The maximum number of memories to return. Defaults to 100.

Returns

dict: A dictionary containing a list of memories under the "results" key,

and potentially "relations" if graph store is enabled. For API v1.0,
it might return a direct list (see deprecation warning).
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", ...}]}`

`search`

def search(
    query: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    limit: int = 100,
    filters: Optional[Dict[str, Any]] = None,
    threshold: Optional[float] = None,
    use_tcmgm: bool = False,
    time_window: Optional[Dict] = None,
    include_causal: bool = True,
    merge_chunks: Optional[bool] = None,
    chunk_context_window: Optional[int] = None,
    use_context_graph: Optional[bool] = None,
    include_archived: bool = False,
)

Searches for memories based on a query

Args

query str: Query to search for.
user_id str, optional: ID of the user to search for. Defaults to None.
agent_id str, optional: ID of the agent to search for. Defaults to None.
run_id str, optional: ID of the run to search for. Defaults to None.
limit int, optional: Limit the number of results. Defaults to 100.
filters dict, optional: Filters to apply to the search. Defaults to None..
threshold float, optional: Minimum score for a memory to be included in the results. Defaults to None.
use_tcmgm bool, optional: Use TCMGM fused retrieval (combines vector, graph, and timeline). Defaults to False.
time_window dict, optional: Time window for TCMGM temporal filtering. Dict with 'start' and 'end' datetime strings (ISO format). Defaults to None.
include_causal bool, optional: Include causal chain exploration when using TCMGM. Defaults to True.
merge_chunks bool, optional: Merge chunks from same document (default: config.chunking.merge_chunks_on_retrieval). Defaults to None.
chunk_context_window int, optional: Number of neighboring chunks to include (default: config.chunking.chunk_context_window). Defaults to None.

Returns

dict: A dictionary containing the search results, typically under a "results" key,

and potentially "relations" if graph store is enabled.
When use_tcmgm=True, also includes "timeline_results", "causal_chains", and "fused_ranking".
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "score": 0.8, ...}]}`

`build_timeline`

def build_timeline(
    messages,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    session_start: Optional[datetime] = None,
) -> Dict[str, Any]

Turn a conversation into dated events and cause → effect links (TCMGM).

The events and links are stored in the graph, where get_timeline and search(use_tcmgm=True) find them. Set tcmgm.build_timeline_on_add to do this on every add() instead.

Makes LLM calls: one to pull out the events, one for the causal links (unless tcmgm.causal_extraction_mode is "rules"), and the usual entity extraction for each event stored in the graph.

Args

messages str or List[Dict[str, str]]: The conversation, as text or as messages with role and text content.
user_id str, optional: ID the events belong to. TCMGM keys events by the first of user_id, agent_id, run_id.
agent_id str, optional: Used when user_id is not given.
run_id str, optional: Used when user_id and agent_id are not given.
session_start datetime, optional: Time of the first event. Defaults to now; later events are spaced two minutes apart.

Returns

dict: {"events": [...], "causal_links": [...], "timeline_span": {"start", "end"}}.

Raises

RuntimeError: TCMGM is disabled, or the graph store is not neo4j.
ValueError: None of user_id, agent_id, run_id is given.

`get_timeline`

def get_timeline(
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    time_window: Optional[Dict] = None,
    limit: int = 50,
) -> List[Dict[str, Any]]

Return the events stored by build_timeline for one id, oldest first (TCMGM).

Args

user_id str, optional: ID to read. TCMGM keys events by the first of user_id, agent_id, run_id.
agent_id str, optional: Used when user_id is not given.
run_id str, optional: Used when user_id and agent_id are not given.
time_window dict, optional: {"start": ..., "end": ...} as ISO strings or datetimes; either bound may be left out.
limit int, optional: Maximum number of nodes. Defaults to 50.

Returns

list: Events with event_id, content, timestamp, confidence,

``modality`` and ``user_id``.

Raises

RuntimeError: TCMGM is disabled, or the graph store is not neo4j.
ValueError: None of user_id, agent_id, run_id is given.

`get_causal_chain`

def get_causal_chain(
    event_id: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    direction: str = 'backward',
    max_depth: int = 3,
) -> List[Dict[str, Any]]

Follow cause → effect links from one event (TCMGM).

direction="backward" answers "why did this happen?" (its causes); "forward" answers "what did this lead to?" (its effects).

Args

event_id str: An event_id from get_timeline or build_timeline.
user_id str, optional: ID the events belong to. TCMGM keys events by the first of user_id, agent_id, run_id.
agent_id str, optional: Used when user_id is not given.
run_id str, optional: Used when user_id and agent_id are not given.
direction str, optional: "backward" (default) or "forward".
max_depth int, optional: Longest chain to follow. Defaults to 3.

Returns

list: Chains, shortest first, each ``{"events": [...], "causal_links": [...],

"depth": n}`` with events in cause → effect order.

Raises

RuntimeError: TCMGM is disabled, or the graph store is not neo4j.
ValueError: No id is given, or direction is not backward/forward.

`context_graph` property

def context_graph()

Public handle to the Context-Graph Layer facade.

Returns None when MemoryConfig.context_graph.enabled is False. Examples and tests should prefer this public attribute over the private _context_graph field.

`tick_decay`

def tick_decay() -> Dict[str, int]

Apply one round of CGL decay. No-op when the layer is disabled.

`reset_context_graph`

def reset_context_graph() -> None

Drop every CGL node / edge while leaving vector and entity stores alone.

Emits a single synthetic reset changelog event with target_id="*" so subscribers can detect full resets.

`subscribe_context_graph_changes`

def subscribe_context_graph_changes(
    callback: Callable[[Any], None],
    *,
    from_event_id: Optional[int] = None,
    event_types: Optional[Iterable[str]] = None,
) -> Optional[int]

The callback is invoked in a background daemon thread; it must not raise. Returns None when the Context-Graph Layer or its changelog is disabled.

Args

callback: (ChangeEvent) -> None — called once per matching event.
from_event_id: Replay all events with id > this value before streaming live events. Pass 0 to replay the entire history.
event_types: Optional whitelist of event-type strings (e.g. ["node_added", "edge_added"]). None accepts all types.

`unsubscribe_context_graph_changes`

def unsubscribe_context_graph_changes(token: int) -> bool

Deregister a change-bus subscription by *token*.

Returns True if the token was found and removed. Returns False when the CGL is disabled or the token is unknown.

`mark_relevant`

def mark_relevant(memory_id: str, delta: float = 0.25) -> Optional[float]

Boost a memory's CGL relevance and pin a floor against decay.

Returns the new relevance in [0, 1], or None when the Context-Graph Layer is disabled or the memory is unknown.

`snapshot_context_graph`

def snapshot_context_graph(path: str) -> str

Write a portable JSON-Lines snapshot of the Context-Graph Layer.

Works for both the networkx and Neo4j backends. Returns the path. Raises RuntimeError if the CGL is disabled.

`replay_context_graph`

def replay_context_graph(
    *,
    until: Optional[datetime] = None,
    target_backend: Optional[Any] = None,
) -> Any

Rebuild a fresh :class:ContextGraph from the changelog.

Returns a *new* :class:ContextGraph facade (the production graph on self is left untouched). When target_backend is omitted an in-process :class:NetworkXBackend is used, which is a sensible default for audit / diff workflows regardless of the live backend. until caps the replay at a point-in-time (inclusive).

`mark_irrelevant`

def mark_irrelevant(memory_id: str, delta: float = 0.25) -> Optional[float]

Attenuate a memory's CGL relevance without pinning a floor.

`backfill_context_graph`

def backfill_context_graph(
    *,
    batch_size: int = 500,
    embed: bool = True,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
) -> Dict[str, int]

Back-fill the Context-Graph Layer from existing vector-store memories.

Idempotent: nodes are upserted, not duplicated. Safe to run multiple times. Use after enabling CGL on a Memory instance that already has historical data.

Args

batch_size: Memories processed per embedding batch (controls concurrency / rate-limiting with the embedding provider).
embed: Re-compute embeddings so TOPIC_SIMILAR edges can be synthesised. Set to False for a fast structural-only backfill. user_id / agent_id / run_id: Scope the backfill to a subset of memories. When all three are None every memory in the store is backfilled.

Returns

{"nodes": N, "edges": E} where *N* is the number of nodes upserted and *E* is the net new edges created.

Raises

RuntimeError: If the Context-Graph Layer is not enabled.

`update`

def update(memory_id, data)

Update a memory by ID.

Args

memory_id str: ID of the memory to update.
data dict: Data to update the memory with.

Returns

dict: Updated memory.

`delete`

def delete(memory_id)

Delete a memory by ID.

Args

memory_id str: ID of the memory to delete.

`delete_all`

def delete_all(
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
)

Delete all memories.

Args

user_id str, optional: ID of the user to delete memories for. Defaults to None.
agent_id str, optional: ID of the agent to delete memories for. Defaults to None.
run_id str, optional: ID of the run to delete memories for. Defaults to None.

`history`

def history(
    memory_id,
    *,
    tenant_id: Optional[str] = None,
    sub_tenant_id: Optional[str] = None,
)

Get the history of changes for a memory by ID.

Args

memory_id str: ID of the memory to get history for.
tenant_id Optional[str]: Phase T4 — when supplied, scopes the read to rows tagged with this tenant (or legacy NULL rows). When omitted and MemoryConfig.tenant.enabled is True, the default tenant id is used.
sub_tenant_id Optional[str]: Phase T4 — sub-tenant scope filter. Same NULL-tolerance as tenant_id.

Returns

list: List of changes for the memory.

`reset`

def reset()

Reset the memory store by:

Deletes the vector store collection
Resets the database
Recreates the vector store with a new client

`chat`

def chat(query)

No description.

AsyncMemory

from outhad_contextkit.memory.main import AsyncMemory

`init`

def __init__(config: MemoryConfig = MemoryConfig())

No description.

`from_config` classmethod

async def from_config(config_dict: Dict[str, Any])

No description.

`add` async

async def add(
    messages,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    infer: bool = True,
    memory_type: Optional[str] = None,
    prompt: Optional[str] = None,
    llm=None,
)

Create a new memory asynchronously.

Args

messages str or List[Dict[str, str]]: Messages to store in the memory.
user_id str, optional: ID of the user creating the memory.
agent_id str, optional: ID of the agent creating the memory. Defaults to None.
run_id str, optional: ID of the run creating the memory. Defaults to None.
metadata dict, optional: Metadata to store with the memory. Defaults to None.
infer bool, optional: Whether to infer the memories. Defaults to True.
memory_type str, optional: Type of memory to create. Defaults to None. Pass "procedural_memory" to create procedural memories.
prompt str, optional: Prompt to use for the memory creation. Defaults to None.
llm BaseChatModel, optional: LLM class to use for generating procedural memories. Defaults to None. Useful when user is using LangChain ChatModel.

Returns

dict: A dictionary containing the result of the memory addition operation.

`get` async

async def get(memory_id)

Retrieve a memory by ID asynchronously.

Args

memory_id str: ID of the memory to retrieve.

Returns

dict: Retrieved memory.

`get_all` async

async def get_all(
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    filters: Optional[Dict[str, Any]] = None,
    limit: int = 100,
)

List all memories.

Args:

user_id (str, optional): user id
agent_id (str, optional): agent id
run_id (str, optional): run id
filters (dict, optional): Additional custom key-value filters to apply to the search.
    These are merged with the ID-based scoping filters. For example,
    `filters={"actor_id": "some_user"}`.
limit (int, optional): The maximum number of memories to return. Defaults to 100.

Returns:

dict: A dictionary containing a list of memories under the "results" key,
      and potentially "relations" if graph store is enabled. For API v1.0,
      it might return a direct list (see deprecation warning).
      Example for v1.1+: `{"results": [{"id": "...", "memory": "...", ...}]}`

`search` async

async def search(
    query: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    limit: int = 100,
    filters: Optional[Dict[str, Any]] = None,
    threshold: Optional[float] = None,
    merge_chunks: Optional[bool] = None,
    chunk_context_window: Optional[int] = None,
    use_context_graph: Optional[bool] = None,
    include_archived: bool = False,
    use_tcmgm: bool = False,
    time_window: Optional[Dict] = None,
    include_causal: bool = True,
)

Searches for memories based on a query

Args

query str: Query to search for.
user_id str, optional: ID of the user to search for. Defaults to None.
agent_id str, optional: ID of the agent to search for. Defaults to None.
run_id str, optional: ID of the run to search for. Defaults to None.
limit int, optional: Limit the number of results. Defaults to 100.
filters dict, optional: Filters to apply to the search. Defaults to None.
threshold float, optional: Minimum score for a memory to be included in the results. Defaults to None.
use_tcmgm bool, optional: Use TCMGM fused retrieval (combines vector, graph, and timeline). Defaults to False.
time_window dict, optional: Time window for TCMGM temporal filtering. Dict with 'start' and 'end' datetime strings (ISO format). Defaults to None.
include_causal bool, optional: Include causal chain exploration when using TCMGM. Defaults to True.
merge_chunks bool, optional: Whether to merge chunks from same document. Defaults to config.chunking.merge_chunks_on_retrieval.
chunk_context_window int, optional: Number of neighboring chunks to include (±N). Defaults to config.chunking.chunk_context_window.

Returns

dict: A dictionary containing the search results, typically under a "results" key,

and potentially "relations" if graph store is enabled.
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "score": 0.8, ...}]}`

`build_timeline` async

async def build_timeline(
    messages,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    session_start: Optional[datetime] = None,
) -> Dict[str, Any]

Async twin of :meth:Memory.build_timeline.

`get_timeline` async

async def get_timeline(
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    time_window: Optional[Dict] = None,
    limit: int = 50,
) -> List[Dict[str, Any]]

Async twin of :meth:Memory.get_timeline.

`get_causal_chain` async

async def get_causal_chain(
    event_id: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    direction: str = 'backward',
    max_depth: int = 3,
) -> List[Dict[str, Any]]

Async twin of :meth:Memory.get_causal_chain.

`tick_decay` async

async def tick_decay() -> Dict[str, int]

Async wrapper around the CGL decay primitive.

`reset_context_graph` async

async def reset_context_graph() -> None

No description.

`mark_relevant` async

async def mark_relevant(memory_id: str, delta: float = 0.25) -> Optional[float]

Async twin of :meth:Memory.mark_relevant.

`mark_irrelevant` async

async def mark_irrelevant(memory_id: str, delta: float = 0.25) -> Optional[float]

Async twin of :meth:Memory.mark_irrelevant.

`record_feedback` async

async def record_feedback(
    memory_id: str,
    *,
    helpful: bool,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    query: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]

Async twin of :meth:Memory.record_feedback.

`reset_feedback` async

async def reset_feedback(*, user_id: Optional[str] = None) -> int

Async twin of :meth:Memory.reset_feedback.

`set_role_policy`

def set_role_policy(policy: Optional[Any]) -> None

Async twin of :meth:Memory.set_role_policy (sync — stores a ref).

`get_role_policy`

def get_role_policy() -> Optional[Any]

Async twin of :meth:Memory.get_role_policy.

`backfill_mspr` async

async def backfill_mspr() -> Dict[str, int]

Async twin of :meth:Memory.backfill_mspr.

`backfill_tenant` async

async def backfill_tenant(
    *,
    default_tenant_id: Optional[str] = None,
    batch_size: int = 500,
) -> Dict[str, int]

Async twin of :meth:Memory.backfill_tenant.

`export_tenant` async

async def export_tenant(tenant_id: str, dest_dir: str) -> Dict[str, int]

Async twin of :meth:Memory.export_tenant.

`import_tenant` async

async def import_tenant(src_dir: str) -> Dict[str, int]

Async twin of :meth:Memory.import_tenant.

`migrate_tenant` async

async def migrate_tenant(src_id: str, dst_id: str) -> Dict[str, int]

Async twin of :meth:Memory.migrate_tenant.

`decay_score` async

async def decay_score(memory_id: str) -> Optional[float]

No description.

`archive_low` async

async def archive_low(
    *,
    threshold: Optional[float] = None,
    dry_run: bool = False,
) -> Dict[str, Any]

No description.

`start_decay_scheduler` async

async def start_decay_scheduler() -> None

No description.

`stop_decay_scheduler` async

async def stop_decay_scheduler(*, timeout: float = 5.0) -> None

No description.

`close` async

async def close() -> None

No description.

`record_reference` async

async def record_reference(
    memory_id: str,
    *,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
    strength: float = 1.0,
) -> Optional[int]

No description.

`versions` async

async def versions(memory_id: str) -> List[Any]

No description.

`latest_version` async

async def latest_version(memory_id: str) -> Optional[Any]

No description.

`diff_versions` async

async def diff_versions(old_id: str, new_id: str, *, mode: Optional[str] = None) -> str

No description.

`rollback` async

async def rollback(
    memory_id: str,
    *,
    to_version: int,
    actor_id: Optional[str] = None,
) -> Optional[str]

No description.

`demote_to_cold` async

async def demote_to_cold(memory_id: str) -> bool

No description.

`promote_from_cold` async

async def promote_from_cold(memory_id: str) -> Optional[Dict[str, Any]]

No description.

`backfill_decay_v2` async

async def backfill_decay_v2() -> Dict[str, int]

No description.

`snapshot_context_graph` async

async def snapshot_context_graph(path: str) -> str

Async twin of :meth:Memory.snapshot_context_graph.

`subscribe_context_graph_changes`

def subscribe_context_graph_changes(
    callback: Callable[[Any], None],
    *,
    from_event_id: Optional[int] = None,
    event_types: Optional[Iterable[str]] = None,
) -> Optional[int]

Async-safe twin of :meth:Memory.subscribe_context_graph_changes.

Subscribe is synchronous (thread registration); the callback fires in a background daemon thread regardless of whether the caller is sync or async.

`unsubscribe_context_graph_changes`

def unsubscribe_context_graph_changes(token: int) -> bool

Async-safe twin of :meth:Memory.unsubscribe_context_graph_changes.

`replay_context_graph` async

async def replay_context_graph(
    *,
    until: Optional[datetime] = None,
    target_backend: Optional[Any] = None,
) -> Any

Async twin of :meth:Memory.replay_context_graph.

`backfill_context_graph` async

async def backfill_context_graph(
    *,
    batch_size: int = 500,
    embed: bool = True,
    user_id: Optional[str] = None,
    agent_id: Optional[str] = None,
    run_id: Optional[str] = None,
) -> Dict[str, int]

Async wrapper around :meth:Memory.backfill_context_graph.

`update` async

async def update(memory_id, data)

Update a memory by ID asynchronously.

Args

memory_id str: ID of the memory to update.
data dict: Data to update the memory with.

Returns

dict: Updated memory.

`delete` async

async def delete(memory_id)

Delete a memory by ID asynchronously.

Args

memory_id str: ID of the memory to delete.

`delete_all` async

async def delete_all(user_id=None, agent_id=None, run_id=None)

Delete all memories asynchronously.

Args

user_id str, optional: ID of the user to delete memories for. Defaults to None.
agent_id str, optional: ID of the agent to delete memories for. Defaults to None.
run_id str, optional: ID of the run to delete memories for. Defaults to None.

`history` async

async def history(
    memory_id,
    *,
    tenant_id: Optional[str] = None,
    sub_tenant_id: Optional[str] = None,
)

Get the history of changes for a memory by ID asynchronously.

Args

memory_id str: ID of the memory to get history for.
tenant_id Optional[str]: Phase T4 — tenant scope filter.
sub_tenant_id Optional[str]: Phase T4 — sub-tenant scope filter.

Returns

list: List of changes for the memory.

`reset` async

async def reset()

Reset the memory store asynchronously by:

Deletes the vector store collection
Resets the database
Recreates the vector store with a new client

`chat` async

async def chat(query)

No description.

Library API reference

Memory

__init__

from_config classmethod

decay_score

archive_low

start_decay_scheduler

stop_decay_scheduler

close

record_reference

versions

latest_version

diff_versions

rollback

demote_to_cold

promote_from_cold

backfill_decay_v2

record_feedback

reset_feedback

set_role_policy

get_role_policy

backfill_mspr

backfill_tenant

export_tenant

import_tenant

migrate_tenant

add

get

get_all

search

build_timeline

get_timeline

get_causal_chain

context_graph property

tick_decay

reset_context_graph

subscribe_context_graph_changes

unsubscribe_context_graph_changes

mark_relevant

snapshot_context_graph

replay_context_graph

mark_irrelevant

backfill_context_graph

update

delete

delete_all

history

reset

chat

AsyncMemory

__init__

from_config classmethod

add async

get async

get_all async

search async

build_timeline async

get_timeline async

get_causal_chain async

tick_decay async

reset_context_graph async

mark_relevant async

mark_irrelevant async

record_feedback async

reset_feedback async

set_role_policy

get_role_policy

backfill_mspr async

backfill_tenant async

export_tenant async

import_tenant async

migrate_tenant async

decay_score async

archive_low async

start_decay_scheduler async

stop_decay_scheduler async

close async

record_reference async

versions async

latest_version async

`init`

`from_config` classmethod

`decay_score`

`archive_low`

`start_decay_scheduler`

`stop_decay_scheduler`

`close`

`record_reference`

`versions`

`latest_version`

`diff_versions`

`rollback`

`demote_to_cold`

`promote_from_cold`

`backfill_decay_v2`

`record_feedback`

`reset_feedback`

`set_role_policy`

`get_role_policy`

`backfill_mspr`

`backfill_tenant`

`export_tenant`

`import_tenant`

`migrate_tenant`

`add`

`get`

`get_all`

`search`

`build_timeline`

`get_timeline`

`get_causal_chain`

`context_graph` property

`tick_decay`

`reset_context_graph`

`subscribe_context_graph_changes`

`unsubscribe_context_graph_changes`

`mark_relevant`

`snapshot_context_graph`

`replay_context_graph`

`mark_irrelevant`

`backfill_context_graph`

`update`

`delete`

`delete_all`

`history`

`reset`

`chat`

`init`

`from_config` classmethod

`add` async

`get` async

`get_all` async

`search` async

`build_timeline` async

`get_timeline` async

`get_causal_chain` async

`tick_decay` async

`reset_context_graph` async

`mark_relevant` async

`mark_irrelevant` async

`record_feedback` async

`reset_feedback` async

`set_role_policy`

`get_role_policy`

`backfill_mspr` async

`backfill_tenant` async

`export_tenant` async

`import_tenant` async

`migrate_tenant` async

`decay_score` async

`archive_low` async

`start_decay_scheduler` async

`stop_decay_scheduler` async

`close` async

`record_reference` async

`versions` async

`latest_version` async

`diff_versions` async

`rollback` async

`demote_to_cold` async