calute.memory.contextual_memory

calute.memory.contextual_memory#

Contextual memory for maintaining conversation and task context.

Provides a hybrid memory system that combines short-term and long-term storage with context-aware retrieval and automatic memory promotion based on access patterns and importance scores.

class calute.memory.contextual_memory.ContextualMemory(short_term_capacity: int = 20, long_term_storage: Any | None = None, promotion_threshold: int = 3, importance_threshold: float = 0.7)[source]#

Bases: Memory

Hybrid memory combining short-term and long-term memories.

Provides context-aware retrieval and automatic promotion of memories from short-term to long-term based on access patterns and importance scores. Maintains a context stack for tracking conversation state.

The context stack allows callers to push and pop situational contexts (e.g. task, conversation) that influence how search results are ranked. Items whose metadata context matches the current stack context receive a relevance boost during search.

short_term#

The underlying ShortTermMemory instance used for recent, transient memories.

long_term#

The underlying LongTermMemory instance used for persisted, important memories.

promotion_threshold#

Number of accesses required before a short-term memory item is promoted to long-term storage.

importance_threshold#

Minimum importance score (0.0–1.0) at which a new memory is saved directly to long-term storage.

context_stack#

Ordered list of context dictionaries representing the current situational context, used for re-ranking search results.

Example

>>> from calute.memory import ContextualMemory
>>> mem = ContextualMemory(short_term_capacity=50)
>>> mem.push_context("task", {"name": "research"})
>>> item = mem.save("Important finding", importance=0.9)
>>> results = mem.search("finding")
clear() → None[source]#

Clear all memories from both stores and reset the context stack.

delete(memory_id: str | None = None, filters: dict[str, Any] | None = None) → int[source]#

Delete memory items from both short-term and long-term stores.

The deletion is applied to both stores regardless of where the item actually resides, and the counts are summed.

Parameters

  • memory_id – UUID of a specific memory item to delete. When provided, the item is removed from whichever store holds it.

  • filters – Optional key-value criteria for bulk deletion across both stores.

Returns

Total number of items deleted across both stores.

get_context_summary() → str[source]#

Build a human-readable summary of current context and recent memories.

Combines up to three sections:

  1. Current context – the last three entries on the context stack.

  2. Recent activity – the five most recent short-term memory items.

  3. Important memories – long-term items with importance >= 0.8.

Returns

Multi-line formatted summary string. Returns "No context available." when all sections are empty.

get_current_context() → dict[str, Any] | None[source]#

Peek at the topmost context frame without removing it.

Returns

The current context dictionary containing "type", "data", and "timestamp" keys, or None if the stack is empty.

pop_context() → dict[str, Any] | None[source]#

Pop and return the most recent context frame from the stack.

Removes the topmost context so that subsequent search calls no longer consider it for re-ranking.

Returns

The popped context dictionary containing "type", "data", and "timestamp" keys, or None if the stack is empty.

push_context(context_type: str, context_data: dict[str, Any]) → None[source]#

Push a new context frame onto the context stack.

The pushed context influences subsequent search() calls by boosting the relevance of items whose stored context matches the current stack top.

Parameters

  • context_type – Category label for the context frame (e.g. "task", "conversation", "tool_use").

  • context_data – Arbitrary dictionary of context information that will be used for re-ranking during search.

retrieve(memory_id: str | None = None, filters: dict[str, Any] | None = None, limit: int = 10) → calute.memory.base.MemoryItem | list[calute.memory.base.MemoryItem] | None[source]#

Retrieve memory items from both short-term and long-term stores.

When a specific memory_id is provided, looks in short-term memory first, then falls back to long-term memory. Accessing an item in short-term memory may trigger automatic promotion to long-term memory if it has been accessed enough times.

When memory_id is None, performs a filter-based retrieval across both stores, filling up to limit items from short-term first and then from long-term.

Parameters

  • memory_id – UUID of a specific memory item to retrieve. When provided, other parameters are ignored.

  • filters – Optional key-value criteria for bulk retrieval. Applied to both short-term and long-term stores.

  • limit – Maximum total number of items to return when using filter-based retrieval.

Returns

A single MemoryItem when memory_id is provided (or None if not found), or a list of MemoryItem instances when performing filter-based retrieval.

save(content: str, metadata: dict[str, Any] | None = None, importance: float = 0.5, to_long_term: bool = False, **kwargs) → MemoryItem[source]#

Save a memory item with automatic context attachment.

The current context stack (if non-empty) is attached to the item’s metadata under the "context" key. Items with high importance or the to_long_term flag are stored directly in long-term memory; otherwise they go to short-term memory and are checked for promotion.

Parameters

  • content – Text content to store.

  • metadata – Optional key-value metadata to attach to the item.

  • importance – Importance score (0.0–1.0). Items at or above importance_threshold are routed to long-term memory.

  • to_long_term – When True, forces storage in long-term memory regardless of the importance score.

  • **kwargs – Additional keyword arguments forwarded to the underlying memory store’s save method (e.g. agent_id, user_id).

Returns

The newly created MemoryItem.

search(query: str, limit: int = 10, filters: dict[str, Any] | None = None, search_long_term: bool = True, **kwargs) → list[calute.memory.base.MemoryItem][source]#

Search both short-term and long-term memories.

Searches short-term memory first, then (optionally) long-term memory. Each result is annotated with a "source" metadata key indicating which store it came from. If a context stack is active, results are re-ranked to boost items that match the current context before the final relevance sort.

Parameters

  • query – Natural-language or keyword search query string.

  • limit – Maximum number of results to return after merging and ranking results from both stores.

  • filters – Optional key-value filter criteria forwarded to the underlying memory stores.

  • search_long_term – When True (default), includes long-term memory in the search. Set to False to restrict results to short-term memory only.

  • **kwargs – Additional keyword arguments forwarded to the underlying stores’ search methods.

Returns

List of MemoryItem instances sorted by relevance score in descending order, with at most limit entries.

update(memory_id: str, updates: dict[str, Any]) → bool[source]#

Update a memory item in the appropriate sub-store.

Attempts the update in short-term memory first. If the item is not found there, falls back to long-term memory.

Parameters

  • memory_id – UUID of the memory item to update.

  • updates – Dictionary mapping attribute names to their new values. Supported keys include any MemoryItem attribute (e.g. "content", "metadata").

Returns

True if the item was found and updated in either store, False if the memory_id was not found in either store.

Contents