calute.memory.short_term_memory#
Short-term memory implementation.
- class calute.memory.short_term_memory.ShortTermMemory(capacity: int = 20, storage: Any | None = None, enable_embeddings: bool = False)[source]#
Bases:
MemoryShort-term memory with FIFO eviction and recent-context tracking.
Uses a bounded
collections.dequeto enforce a fixed capacity. When the capacity is reached, the oldest item is silently discarded as new items are appended. Ideal for maintaining conversation context and recent interaction history.- storage#
Optional
MemoryStoragebackend for persistence.
- max_items#
Maximum capacity (mirrors the deque
maxlen).
- enable_embeddings#
Whether dense vector embeddings are enabled.
- _items#
Bounded deque holding
MemoryIteminstances.
- _index#
Dictionary mapping memory IDs to items for O(1) lookup.
Example
>>> from calute.memory import ShortTermMemory >>> stm = ShortTermMemory(capacity=10) >>> item = stm.save("User asked about weather") >>> stm.get_recent(3) [MemoryItem(...)]
- clear() None[source]#
Clear all short-term memories.
Removes all items from memory and storage backend if configured.
- delete(memory_id: str | None = None, filters: dict[str, Any] | None = None) int[source]#
Delete memory items by ID or filter criteria.
- Parameters
memory_id – Specific memory ID to delete
filters – Filter criteria to match items for deletion
- Returns
Number of items deleted
- get_recent(n: int = 5) list[calute.memory.base.MemoryItem][source]#
Get the most recent memory items.
- Parameters
n – Number of recent items to retrieve
- Returns
List of the n most recent MemoryItem instances
- 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 specific memories by ID or filter criteria.
When memory_id is provided, returns the specific item. Otherwise, filters through memories and returns matching items (most recent first).
- Parameters
memory_id – Specific memory ID to retrieve
filters – Filter criteria to match against memory attributes
limit – Maximum number of items to return when using filters
- Returns
Single MemoryItem if memory_id provided, list of MemoryItem if filters used, or None if memory_id not found
- save(content: str, metadata: dict[str, Any] | None = None, agent_id: str | None = None, user_id: str | None = None, conversation_id: str | None = None, **kwargs) MemoryItem[source]#
Save to short-term memory.
Oldest items are automatically removed when capacity is reached (FIFO behavior).
- Parameters
content – The content to store in memory
metadata – Additional metadata to attach to the memory item
agent_id – Identifier of the agent creating this memory
user_id – Identifier of the user associated with this memory
conversation_id – Identifier of the conversation context
**kwargs – Additional fields to include in metadata
- Returns
The created MemoryItem instance
- search(query: str, limit: int = 10, filters: dict[str, Any] | None = None, min_relevance: float = 0.0, **kwargs) list[calute.memory.base.MemoryItem][source]#
Search short-term memory using keyword matching and filters.
Performs case-insensitive keyword matching and returns most recent matches first. Supports filtering by agent_id, user_id, and conversation_id.
- Parameters
query – Search query string for keyword matching
limit – Maximum number of results to return
filters – Filter criteria (supports agent_id, user_id, conversation_id)
min_relevance – Minimum relevance score threshold (0.0 to 1.0)
**kwargs – Additional search parameters (unused)
- Returns
List of matching MemoryItem instances sorted by relevance and timestamp