Adapt Memory¶

Adapt Memory — mvp.adapt_memory

Cluster: Knowledge & Grounding | Type: component | MCP Tools: 54

Overview¶

adapt_memory has two surfaces. The lightweight AdaptMemoryBlock is an in-process key-value memory store that persists arbitrary values only across infer() calls within the same block instance. It supports store, retrieve, list (with optional limit), delete, clear, and metadata ops on named records that carry a timestamp and optional tag list. The state dict is held on the block dataclass, making it suitable for short-lived agent sessions where a lightweight working memory is needed without a database dependency.

For launch-facing MCP usage, prefer the SQLite-backed memory_mcp server. It persists memory across process restarts via MEMORY_DB_PATH (default ~/.memory/memory.db) and adds episodic, semantic, procedural, relational, STM/LTM, working-memory, search operations, and an explicit advanced backend-adapter layer. Treat the simple in-process block as a basic/internal interface; use memory_mcp for user workflows where non-technical users expect recall to survive installation, restart, and multi-session use.

The stable curated API remains the default. Advanced backend-native operations are opt-in through memory_backend_capabilities, memory_backend_call, memory_extract_entities, memory_backend_search, and memory_pack_context_native. These advanced calls are allowlisted, payload-bounded, audited with redacted metadata, and normalized into completion_state, warning_card, evidence, request_id, and run_id fields.

Reliability mapping (CRUD, three-way): durable SQLite operations return verified ONLY when the evidence carries a store confirmation (a row id or row count); an ephemeral db_path=":memory:" store returns qualified-draft with degradation_reason="ephemeral_store" and a G6_E_MEMORY_EPHEMERAL_STORE warning; a store exception fails closed to blocked-escalated with G6_E_MEMORY_STORE_UNAVAILABLE rather than silently succeeding. There is no path where an in-:memory: or store-failed operation reads as verified. The simple in-process AdaptMemoryBlock is non-durable by construction and is always qualified-draft, never verified; its degraded/degradation_reason channel is reserved for the silent-overwrite signal (key_already_exists). The advanced backend lane was already honest and is unchanged: quality-changing fallbacks return qualified-draft with backend_unavailable:*/fallback_used evidence, and unavailable/disallowed/invalid backend requests return blocked-escalated. The component stays beta (awaiting GDPval calibration).

When to use:

• Accumulating intermediate results across multiple agent steps within a session
• Caching retrieved documents or computed values to avoid redundant processing
• Passing named artefacts between pipeline stages without serialising to disk
• Building a simple blackboard architecture where multiple blocks read and write shared state

Example:

from mvp.adapt_memory import AdaptMemoryBlock, MemoryInput

block = AdaptMemoryBlock(name="memory")
block.infer(MemoryInput(op="store", key="plan", value={"steps": ["research", "draft", "review"]}))
result = block.infer(MemoryInput(op="retrieve", key="plan"))
# result.value.value → {"steps": ["research", "draft", "review"]}
# result.value.found → True

Works well with: adapt_instructor, goal_engine, ctx_rag

Public API¶

AdaptMemoryBlock(AIBlock[MemoryInput, MemoryOutput, dict])¶

In-process key-value memory store.

Methods:

infer(data: MemoryInput) -> Result[MemoryOutput]¶

MemoryRecord(BaseModel)¶

A single entry in the memory store.

MemoryInput(BaseModel)¶

Input for AdaptMemoryBlock.

MemoryOutput(BaseModel)¶

Output from AdaptMemoryBlock.

AdaptMemoryBlock(AIBlock[MCPMemoryInput, MCPMemoryOutput, dict])¶

Unified memory block with SQLite persistence and 7 memory types.

Methods:

infer(data: MCPMemoryInput) -> Result[MCPMemoryOutput]¶

MCPMemoryRecord(BaseModel)¶

A single memory record stored in SQLite.

MCPMemoryInput(BaseModel)¶

Input for the unified AdaptMemoryBlock (30 operations).

MCPMemoryOutput(BaseModel)¶

Output from the unified AdaptMemoryBlock.

MemoryStore¶

Sync SQLite memory store with 6 tables covering all memory types.

Constructor:

Methods:

store_record(key: str, value: str, tier: str = 'any', memory_type: str = 'general', tags: list[str] | None = None, provenance: str = '') -> str¶

Upsert a record; returns the record id.

retrieve(key: str, tier: str = 'any', memory_type: str = 'any') -> dict[str, Any] | None¶

list_records(tier: str = 'any', memory_type: str = 'any', tags: list[str] | None = None, limit: int = 100) -> list[dict[str, Any]]¶

delete(key: str, tier: str = 'any', memory_type: str = 'any') -> bool¶

clear(tier: str = 'any') -> int¶

Clear memory records.

record_event(event_type: str, description: str, tags: list[str] | None = None) -> str¶

query_timeline(start_time: str = '', end_time: str = '', event_type: str = '', limit: int = 100) -> list[dict[str, Any]]¶

store_fact(subject: str, predicate: str, object_: str, tags: list[str] | None = None) -> str¶

query_graph(subject: str = '', predicate: str = '', object_: str = '', limit: int = 50) -> list[dict[str, Any]]¶

store_procedure(key: str, steps: list[str], tags: list[str] | None = None) -> str¶

retrieve_procedure(key: str) -> dict[str, Any] | None¶

list_procedures(limit: int = 100) -> list[dict[str, Any]]¶

record_run(key: str, success: bool = True) -> bool¶

set_state(key: str, value: str) -> None¶

get_state(key: str) -> str | None¶

log_interaction(agent: str, content: str) -> str¶

list_interactions(limit: int = 100) -> list[dict[str, Any]]¶

text_search(query: str, tier: str = 'any', memory_type: str = 'any', top_k: int = 5) -> tuple[list[str], list[float]]¶

Return (texts, scores) ranked by TF-IDF similarity (cached, invalidated on writes).

tag_search(tags: list[str], tier: str = 'any', limit: int = 100) -> list[dict[str, Any]]¶

Return records whose tags intersect with the given tag set.

temporal_search(start_time: str = '', end_time: str = '', limit: int = 100) -> list[dict[str, Any]]¶

Return records in an ISO timestamp range.

count_records() -> int¶

close() -> None¶

record_backend_audit(operation: str, backend: str, action: str, payload_size: int, timeout_ms: int, result_status: str, degradation: bool, fallback: str, block_reason: str) -> str¶

list_backend_audits(limit: int = 100) -> list[dict[str, Any]]¶

count_by_tier(tier: str) -> int¶

COUNT(*) of records for a given tier. Fast — no row scan.

promote_tier(key: str, new_tier: str) -> bool¶

Update the tier of a record by key. Returns True if a row was updated.

MCP Tools¶