Memory Storage#

User memory infrastructure with persistent storage, data source integration, and structured memory operations.

Note

For implementation guides and examples, see Memory Storage.

Core Components#

MemoryStorageManager(memory_directory)

Simple file-based memory manager for user memory storage.

UserMemoryProvider()

Core data source provider for comprehensive user memory integration.

MemoryContent(*, timestamp, content)

Structured memory content with timestamp for persistent user context storage.

get_memory_storage_manager()

Get the global memory storage manager instance.

Storage Management#

class framework.services.memory_storage.MemoryStorageManager(memory_directory)[source]#

Bases: object

Simple file-based memory manager for user memory storage.

Provides persistent storage of user memory entries in JSON format with thread-safe file operations and proper error handling. Each user’s memory is stored in a separate JSON file identified by sanitized user ID.

Initialize memory manager with storage directory.

Creates the memory directory if it doesn’t exist and sets up logging for memory operations.

Parameters:

memory_directory (str) – Directory path for storing memory files

Raises:

OSError – If directory cannot be created or accessed

__init__(memory_directory)[source]#

Initialize memory manager with storage directory.

Creates the memory directory if it doesn’t exist and sets up logging for memory operations.

Parameters:

memory_directory (str) – Directory path for storing memory files

Raises:

OSError – If directory cannot be created or accessed

get_user_memory(user_id)[source]#

Get user’s memory content as formatted string.

Parameters:

user_id (str) – User identifier

Returns:

Formatted memory content with timestamps, empty string if no memory

Return type:

str

Examples

Retrieving formatted memory:

>>> manager = MemoryManager("/path/to/memory")
>>> memory_text = manager.get_user_memory("user123")
>>> print(memory_text)
[2025-01-15 14:30] User prefers morning meetings
[2025-01-15 15:45] Working on project Alpha
get_all_memory_entries(user_id)[source]#

Get all user memory entries as list of MemoryContent objects.

Parameters:

user_id (str) – User identifier

Returns:

List of structured memory content objects

Return type:

List[MemoryContent]

Note

Handles timestamp parsing errors by using current time as fallback. Only returns entries with non-empty content.

add_memory_entry(user_id, memory_content)[source]#

Add new memory entry for user.

Parameters:

  • user_id (str) – User identifier

  • memory_content (MemoryContent) – Memory content to add

Returns:

True if entry was added successfully, False otherwise

Return type:

bool

Examples

Adding a memory entry:

>>> from datetime import datetime
>>> manager = MemoryManager("/path/to/memory")
>>> entry = MemoryContent(
...     timestamp=datetime.now(),
...     content="User completed training module"
... )
>>> success = manager.add_memory_entry("user123", entry)
>>> print(f"Entry added: {success}")
get_memories_from_state(state)[source]#

Get user memory from agent state as UserMemories object.

Parameters:

state (AgentState) – Agent state containing user session context

Returns:

UserMemories object with content list

Return type:

UserMemories

Note

Returns empty UserMemories if user_id is not available in config.

Data Source Integration#

class framework.services.memory_storage.UserMemoryProvider[source]#

Bases: DataSourceProvider

Core data source provider for comprehensive user memory integration.

Integrates user memory storage into the framework’s data source management system as a core component that’s universally available across all capabilities and operations. The provider serves as the primary interface between the memory storage backend and the framework’s data source architecture.

This provider implements sophisticated memory retrieval and formatting capabilities, automatically converting stored memory entries into structured context data that enhances LLM interactions with personalized user information. It maintains high priority in the data source hierarchy to ensure memory context is consistently available for conversation continuity and personalized responses.

The provider handles the complete memory integration workflow including user identification, memory retrieval, format conversion, and context creation. It integrates seamlessly with the approval system, session management, and LLM prompt construction systems.

Key architectural features:

  • Core framework data source with universal availability

  • Automatic memory retrieval based on session user ID

  • Structured context creation for LLM prompt integration

  • Health checking and configuration requirement management

  • Error handling with graceful degradation on memory access failures

Note

The provider responds to all requests with valid user IDs, making memory context available for both task extraction and capability execution phases.

Warning

Memory retrieval failures are handled gracefully with logging but do not prevent other data sources from functioning.

See also

framework.data_management.providers.DataSourceProvider : Base provider interface framework.services.memory_storage.MemoryStorageManager : Storage backend used by this provider framework.data_management.request.DataSourceRequest : Request structure for data retrieval framework.data_management.providers.DataSourceContext : Context structure returned by this provider framework.state.UserMemories : State structure for memory collections

Initialize the core user memory provider with storage backend integration.

Sets up the memory storage manager connection for accessing stored user memories and prepares the provider for integration with the framework’s data source management system. The initialization establishes the connection to the global memory storage manager instance for consistent memory access across the framework.

The provider is designed to be instantiated once during framework initialization and reused across all memory retrieval operations.

Note

Uses the global memory storage manager instance for consistent memory access across all framework components.

See also

framework.services.memory_storage.get_memory_storage_manager() : Global manager factory framework.services.memory_storage.MemoryStorageManager : Storage backend

__init__()[source]#

Initialize the core user memory provider with storage backend integration.

Sets up the memory storage manager connection for accessing stored user memories and prepares the provider for integration with the framework’s data source management system. The initialization establishes the connection to the global memory storage manager instance for consistent memory access across the framework.

The provider is designed to be instantiated once during framework initialization and reused across all memory retrieval operations.

Note

Uses the global memory storage manager instance for consistent memory access across all framework components.

See also

framework.services.memory_storage.get_memory_storage_manager() : Global manager factory framework.services.memory_storage.MemoryStorageManager : Storage backend

property name: str#

Unique identifier for this data source provider in the framework registry.

Provides the canonical name used for provider registration, logging, and identification within the data source management system. This name must be unique across all registered data source providers.

Returns:

Provider name identifier used for framework registration

Return type:

str

Note

This name is used in framework configuration, logging, and provider lookup operations.

property context_type: str#

Context type identifier for memory data integration and validation.

Specifies the type of context data this provider creates, enabling the framework to properly validate, route, and format memory context data. This type identifier is used by the context management system for type checking and LLM prompt formatting decisions.

Returns:

Context type identifier for memory data validation and routing

Return type:

str

Note

This context type should match registered context types in the framework’s context registry for proper integration.

See also

framework.data_management.providers.DataSourceContext : Context structure using this type

property description: str#

Human-readable description of this data source for documentation and logging.

Provides a clear, descriptive explanation of what this data source provides and its role in the framework. This description is used in logging, debugging, health check reports, and administrative interfaces.

Returns:

Human-readable description of the memory data source functionality

Return type:

str

async retrieve_data(request)[source]#

Retrieve user memory data and create structured context for framework integration.

Implements the core memory retrieval workflow including user identification, memory storage access, data format conversion, and context creation. The method handles the complete process of converting stored memory entries into structured context data suitable for LLM prompt integration.

The retrieval process follows these steps: 1. Extract user ID from the data source request 2. Retrieve memory entries from the storage manager 3. Convert entries to framework-compatible UserMemories format 4. Create structured DataSourceContext with metadata 5. Return formatted context for prompt integration

Parameters:

request (DataSourceRequest) – Data source request containing user information and session context

Returns:

Structured context with user memory data and metadata, or None if unavailable

Return type:

Optional[DataSourceContext]

Note

Query-based memory retrieval is not currently supported. All memory entries are returned regardless of query parameters.

Warning

Returns None if user ID is unavailable or memory retrieval fails, allowing other data sources to continue functioning.

Examples

Successful memory retrieval:

>>> from framework.data_management.request import DataSourceRequest, DataSourceRequester
>>> provider = UserMemoryProvider()
>>> requester = DataSourceRequester(component_type="capability", component_name="analysis")
>>> request = DataSourceRequest(user_id="user123", requester=requester)
>>> context = await provider.retrieve_data(request)
>>> if context:
...     print(f"Retrieved {context.metadata['entry_count']} memory entries")

See also

framework.data_management.request.DataSourceRequest : Request structure framework.data_management.providers.DataSourceContext : Returned context structure framework.state.UserMemories : Memory data format framework.services.memory_storage.MemoryStorageManager.get_all_memory_entries() : Storage backend method

should_respond(request)[source]#

Determine if memory provider should respond to this request based on user context.

Evaluates whether this provider can meaningfully contribute to the current request by checking for the presence of user identification information. The memory provider is designed to respond to all requests where user context is available, as memory information is universally relevant for personalized interactions.

This method performs a fast, non-I/O check to determine provider applicability before expensive memory retrieval operations are attempted. The provider responds to requests from both task extraction (for context-aware task understanding) and capability execution (for personalized operation enhancement).

Parameters:

request (DataSourceRequest) – Data source request to evaluate for memory provider applicability

Returns:

True if user ID is available and memory retrieval should be attempted

Return type:

bool

Note

This is a lightweight check that doesn’t perform actual memory retrieval. Memory access failures are handled in the retrieve_data method.

Examples

Request with user ID:

>>> from framework.data_management.request import DataSourceRequest, DataSourceRequester
>>> provider = UserMemoryProvider()
>>> requester = DataSourceRequester(component_type="capability", component_name="analysis")
>>> request = DataSourceRequest(user_id="user123", requester=requester)
>>> should_respond = provider.should_respond(request)
>>> print(f"Should respond: {should_respond}")  # True

Request without user ID:

>>> request_no_user = DataSourceRequest(user_id=None, requester=requester)
>>> should_respond = provider.should_respond(request_no_user)
>>> print(f"Should respond: {should_respond}")  # False

See also

framework.data_management.request.DataSourceRequest : Request structure evaluated retrieve_data() : Method called if this returns True

get_config_requirements()[source]#

Get configuration requirements for core user memory system operation.

Specifies the configuration parameters required for proper memory provider operation, including storage directory paths and access permissions. This information is used by the framework’s configuration validation system to ensure all necessary settings are available before provider initialization.

The returned configuration specification includes parameter descriptions, types, requirements status, and configuration path mappings for integration with the configuration system.

Returns:

Dictionary of required configuration parameters with specifications

Return type:

dict

Note

Configuration requirements are validated during framework initialization to ensure memory provider can operate correctly.

Examples

Configuration requirements structure:

>>> provider = UserMemoryProvider()
>>> requirements = provider.get_config_requirements()
>>> print(requirements['memory_directory']['description'])
Directory where core user memory files are stored

See also

configs.config.get_agent_dir() : Configuration path resolution framework.services.memory_storage.MemoryStorageManager : Uses configured directory

async health_check()[source]#

Perform comprehensive health check for the core memory system functionality.

Validates that the memory provider and its dependencies are operational and accessible for memory retrieval operations. This includes checking storage manager availability, configuration validity, and basic system functionality.

The health check is designed to be lightweight and non-intrusive, focusing on system availability rather than comprehensive functionality testing. Health check results are used by the framework’s monitoring and diagnostic systems to ensure data source reliability.

Returns:

True if the memory system is accessible and functional for operations

Return type:

bool

Note

Health check failures are logged but don’t prevent framework operation. Other data sources continue to function if memory provider is unavailable.

Examples

Basic health check:

>>> provider = UserMemoryProvider()
>>> is_healthy = await provider.health_check()
>>> if is_healthy:
...     print("Memory provider is operational")
... else:
...     print("Memory provider has issues")

See also

framework.services.memory_storage.MemoryStorageManager : Backend system checked framework.data_management.manager.DataSourceManager.health_check_all() : Framework health checking

format_for_prompt(context)[source]#

Format core user memory data for optimal LLM prompt integration and readability.

Transforms structured memory context data into a visually appealing, well-organized format that enhances LLM understanding and provides clear user context information. The formatting uses visual markers, structured organization, and clear hierarchy to make memory information easily consumable by language models.

The formatted output includes entry counts, visual indicators, and organized presentation that helps LLMs understand the significance and context of stored user information. The format is designed to be both human-readable and LLM-optimized for maximum effectiveness in conversational contexts.

Parameters:

context (DataSourceContext) – The DataSourceContext containing memory data and metadata

Returns:

Formatted string optimized for LLM prompt inclusion with visual organization

Return type:

str

Examples

Formatted memory with multiple entries:

>>> # Assume context contains memory data
>>> provider = UserMemoryProvider()
>>> formatted = provider.format_for_prompt(context)
>>> print(formatted)
**🧠 User Memory** (3 saved entries):
  **Personal Notes & Insights:**
    • User prefers morning meetings
    • Working on project Alpha
    • Completed training module

Empty memory formatting:

>>> # Context with no memory entries
>>> formatted = provider.format_for_prompt(empty_context)
>>> print(formatted)
**🧠 User Memory** (0 saved entries):
  (No memory entries available)

Note

Returns empty string if context is None or contains no data, allowing graceful handling of memory retrieval failures.

See also

framework.data_management.providers.DataSourceContext : Input context structure framework.state.UserMemories : Memory data format within context retrieve_data() : Method that creates the context formatted by this method

Data Models#

class framework.services.memory_storage.MemoryContent(*, timestamp, content)[source]#

Bases: BaseModel

Structured memory content with timestamp for persistent user context storage.

Represents a single memory entry containing user-specified content along with creation timestamp for temporal context. This model provides the fundamental data structure for all memory storage operations in the framework, ensuring consistent format and metadata across the memory storage system.

The model integrates with Pydantic for automatic validation and serialization, supporting both JSON persistence and LLM prompt formatting. Each memory entry maintains temporal context through precise timestamps, enabling chronological memory retrieval and context-aware memory management.

Parameters:

  • timestamp (datetime) – Creation timestamp for the memory entry

  • content (str) – User content to be stored in memory

Note

All memory entries are timestamped automatically during creation to maintain temporal context and support chronological memory organization.

See also

framework.services.memory_storage.MemoryStorageManager : Storage backend that manages these entries framework.services.memory_storage.UserMemoryProvider : Data source provider that retrieves these entries framework.state.UserMemories : State container for memory collections format_for_llm() : LLM-optimized formatting method

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.

timestamp: datetime#
content: str#
format_for_llm()[source]#

Format memory content for optimal LLM consumption and prompt integration.

Creates a standardized, human-readable format that combines timestamp context with memory content in a way that’s optimized for LLM understanding and processing. The format provides temporal context while maintaining readability and consistent structure across all memory entries.

The formatted output uses a bracketed timestamp format followed by the content, providing clear temporal markers that help LLMs understand the chronological context of stored information.

Returns:

Formatted string with timestamp and content optimized for LLM prompts

Return type:

str

Examples

Basic memory formatting:

>>> from datetime import datetime
>>> entry = MemoryContent(
...     timestamp=datetime(2025, 1, 15, 14, 30),
...     content="User prefers morning meetings"
... )
>>> formatted = entry.format_for_llm()
>>> print(formatted)
[2025-01-15 14:30] User prefers morning meetings

Complex content formatting:

>>> entry = MemoryContent(
...     timestamp=datetime.now(),
...     content="Working on project Alpha with team lead Sarah"
... )
>>> print(entry.format_for_llm())
[2025-01-15 15:45] Working on project Alpha with team lead Sarah

Note

The timestamp format (YYYY-MM-DD HH:MM) provides sufficient temporal resolution while maintaining readability in LLM prompts.

See also

framework.services.memory_storage.UserMemoryProvider : Uses this method for prompt formatting framework.services.memory_storage.MemoryStorageManager.get_user_memory() : Alternative formatting approach

model_config: ClassVar[ConfigDict] = {}#

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

Utility Functions#

framework.services.memory_storage.get_memory_storage_manager()[source]#

Get the global memory storage manager instance.

Creates and caches a global MemoryStorageManager instance using the configured memory directory from global configuration.

Returns:

Global memory storage manager instance

Return type:

MemoryStorageManager

Note

Uses lazy initialization and global caching for efficiency.

See also

Memory Storage

Complete implementation guide and examples

framework.data_management.DataSourceProvider

Base provider interface implemented by UserMemoryProvider