calute.memory.compat

Backward-compatible memory API for legacy Calute callers.

Provides MemoryType, MemoryEntry, and MemoryStore which mirror the older public surface so that existing code continues to work without modification.

class calute.memory.compat.MemoryEntry(content: str, memory_type: str | calute.memory.compat.MemoryType = MemoryType.SHORT_TERM, timestamp: ~datetime.datetime = <factory>, metadata: dict[str, typing.Any] = <factory>, agent_id: str | None = None, task_id: str | None = None, conversation_id: str | None = None, user_id: str | None = None, relevance_score: float = 1.0, access_count: int = 0, last_accessed: datetime.datetime | None = None, embedding: list[float] | None = None, memory_id: str = <factory>, context: dict[str, typing.Any] = <factory>, importance_score: float = 0.5, tags: list[str] = <factory>)

Bases: MemoryItem

Legacy memory entry shape expected by the older API/tests.

Extends MemoryItem with additional fields that the legacy public API exposes: context, importance_score, and tags. These are synced into metadata during __post_init__ so that the underlying storage machinery sees them.

memory_type

Memory category (default: SHORT_TERM).

Type

str | calute.memory.compat.MemoryType

context

Arbitrary context dictionary attached to this entry.

Type

dict[str, Any]

importance_score

Importance weight used for promotion decisions (0-1).

Type

float

tags

Searchable string labels for this entry.

Type

list[str]

context: dict[str, Any]

importance_score: float = 0.5

memory_type: str | calute.memory.compat.MemoryType = 'short_term'

tags: list[str]

to_dict() → dict[str, Any]

Serialise entry to a dictionary including legacy fields.

Returns

Dictionary representation extending the base MemoryItem.to_dict() output with context, importance_score, and tags keys.

class calute.memory.compat.MemoryStore(max_short_term: int = 100, max_working: int = 10, max_long_term: int = 10000, enable_vector_search: bool = False, embedding_dimension: int = 768, enable_persistence: bool = False, persistence_path: str | None = None, cache_size: int = 100, memory_type: calute.memory.compat.MemoryType | None = None)

Bases: ContextualMemory

Compatibility wrapper preserving the legacy memory store surface.

Maintains per-type buckets (self.memories) alongside the underlying ContextualMemory stores so that legacy callers using add_memory() / retrieve_memories() etc. continue to work correctly.

add_memory(content: str, memory_type: MemoryType, agent_id: str, context: dict | None = None, importance_score: float = 0.5, tags: list | None = None, **kwargs) → MemoryEntry

Create and store a new memory entry.

Adds the entry to the appropriate per-type bucket, syncs it into the underlying ContextualMemory stores, and enforces capacity limits.

Parameters

• content – Text content to store.

• memory_type – Which bucket to place the entry in.

• agent_id – Identifier of the creating agent.

• context – Optional context dictionary attached to the entry.

• importance_score – Importance weight (0-1) influencing promotion.

• tags – Optional list of string tags for filtering.

• **kwargs – Additional fields passed to MemoryEntry.

Returns

The newly created MemoryEntry.

clear_memories(memory_type: calute.memory.compat.MemoryType | None = None, agent_id: str | None = None) → None

Clear stored entries, optionally scoped to a type and/or agent.

Parameters

• memory_type – If provided, only clear this type’s bucket.

• agent_id – If provided, only remove entries belonging to this agent.

consolidate_memories(agent_id: str | None = None, merge_similar: bool = True, threshold: float = 0.7) → str

Promote high-importance memories to long-term and return a summary.

Iterates over short-term, working, and episodic buckets and moves any entry with importance_score >= threshold into the long-term bucket. Then generates a text summary of the most important and recent entries.

Parameters

• agent_id – If provided, only consider entries for this agent.

• merge_similar – Accepted for API compatibility; merging is not performed.

• threshold – Minimum importance score for promotion and inclusion in summary.

Returns

Formatted text summary, or an empty string if no entries match.

get_statistics() → dict

Return aggregate statistics including the legacy total_memories count.

Returns

Dictionary extending the base statistics with total_memories (sum across all type buckets) and cache_hit_rate (always 0.0).

retrieve_memories(memory_types: list[calute.memory.compat.MemoryType] | None = None, agent_id: str | None = None, tags: list | None = None, limit: int = 10, min_importance: float = 0.0, query_embedding: object = None, memory_type: calute.memory.compat.MemoryType | None = None) → list[calute.memory.compat.MemoryEntry]

Retrieve entries matching the given criteria across memory types.

Parameters

• memory_types – Explicit list of types to search; defaults to all.

• agent_id – If provided, only return entries for this agent.

• tags – If provided, only return entries containing at least one tag.

• limit – Maximum number of entries to return.

• min_importance – Minimum importance_score threshold (inclusive).

• query_embedding – Accepted for API compatibility but not used.

• memory_type – Single type shortcut (used when memory_types is None).

Returns

List of matching MemoryEntry instances, newest first.

retrieve_recent(minutes_ago: int = 60) → list[calute.memory.compat.MemoryEntry]

Return all entries created within the last minutes_ago minutes.

Parameters

minutes_ago – Look-back window in minutes.

Returns

List of MemoryEntry instances, newest first.

class calute.memory.compat.MemoryType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: StrEnum

Legacy memory type enum.

Each value names a category of memory that the MemoryStore maintains separately.

EPISODIC = 'episodic'

LONG_TERM = 'long_term'

PROCEDURAL = 'procedural'

SEMANTIC = 'semantic'

SHORT_TERM = 'short_term'

WORKING = 'working'