agents.memory.core.stores

Memory store management system integrating with existing Haive store tools.

This module provides enhanced memory storage and retrieval capabilities that build on the existing store tools with intelligent classification, self-query retrieval, and memory lifecycle management.

Classes

MemoryStoreConfig

Configuration for enhanced memory store management.

MemoryStoreManager

Enhanced memory store manager with intelligent classification and retrieval.

Module Contents

class agents.memory.core.stores.MemoryStoreConfig(/, **data)

Bases: pydantic.BaseModel

Configuration for enhanced memory store management.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.memory.core.stores.MemoryStoreManager(config)

Enhanced memory store manager with intelligent classification and retrieval.

This manager builds on the existing store tools to provide: - Automatic memory classification and metadata extraction - Self-query retrieval with memory context - Memory lifecycle management and consolidation - Multi-type memory retrieval strategies

Initialize memory store manager with configuration.

Parameters:

config (MemoryStoreConfig)

async consolidate_memories(namespace=None, max_age_hours=None, dry_run=False)

Consolidate memories by removing duplicates, summarizing old memories, and cleaning up.

Parameters:

  • namespace (tuple[str, Ellipsis] | None) – Namespace to consolidate (if None, consolidate all)

  • max_age_hours (int | None) – Maximum age of memories to keep (if None, use decay calculation)

  • dry_run (bool) – If True, only analyze without making changes

Returns:

MemoryConsolidationResult with consolidation statistics

Return type:

haive.agents.memory.core.types.MemoryConsolidationResult

async delete_memory(memory_id)

Delete a memory by ID.

Parameters:

memory_id (str) – Memory identifier to delete

Returns:

True if successful, False otherwise

Return type:

bool

async get_memory_by_id(memory_id)

Retrieve a specific memory by ID and update access metadata.

Parameters:

memory_id (str) – Unique memory identifier

Returns:

Memory data with metadata or None if not found

Return type:

dict[str, Any] | None

async get_memory_statistics(namespace=None)

Get statistics about stored memories.

Parameters:

namespace (tuple[str, Ellipsis] | None) – Namespace to analyze (if None, analyze all)

Returns:

Dictionary with memory statistics

Return type:

dict[str, Any]

async retrieve_memories(query, namespace=None, memory_types=None, limit=None, time_range=None, importance_threshold=None)

Retrieve memories using intelligent query analysis and ranking.

Parameters:

  • query (str) – Search query (natural language)

  • namespace (tuple[str, Ellipsis] | None) – Memory namespace to search

  • memory_types (list[haive.agents.memory.core.types.MemoryType] | None) – Specific memory types to search (if None, auto-detect)

  • limit (int | None) – Maximum number of results

  • time_range (tuple[datetime.datetime, datetime.datetime] | None) – Optional time range filter (start, end)

  • importance_threshold (float | None) – Minimum importance score

Returns:

List of retrieved memories with metadata

Return type:

list[dict[str, Any]]

async store_memory(content, namespace=None, user_context=None, conversation_context=None, force_classification=None, importance_override=None)

Store a memory with automatic classification and metadata extraction.

Parameters:

  • content (str) – Memory content to store

  • namespace (tuple[str, Ellipsis] | None) – Memory namespace (defaults to configured default)

  • user_context (dict[str, Any] | None) – User-specific context for classification

  • conversation_context (dict[str, Any] | None) – Conversation context for classification

  • force_classification (haive.agents.memory.core.types.MemoryType | None) – Override automatic classification

  • importance_override (float | None) – Override automatic importance scoring

Returns:

Memory ID for later retrieval

Return type:

str

async update_memory(memory_id, content=None, additional_metadata=None, reclassify=False)

Update an existing memory with new content or metadata.

Parameters:

  • memory_id (str) – Memory identifier

  • content (str | None) – New content (if updating content)

  • additional_metadata (dict[str, Any] | None) – Additional metadata to merge

  • reclassify (bool) – Whether to reclassify memory types

Returns:

True if successful, False otherwise

Return type:

bool