Store

Store Memory

Store data persistently using the 0G decentralized storage network.

Overview

The Memory class provides both ephemeral (temporary) and persistent storage capabilities. Persistent storage uses the 0G network to ensure your data is decentralized, secure, and always available.

Constructor

Memory(storage_client: ZGStorageClient, bucket: str, max_ephemeral_messages: int = 50)

Parameters

storage_client · ZGStorageClient · required

The storage client for connecting to the 0G network

bucket · str · required

The storage bucket/stream ID for organizing data

max_ephemeral_messages · int · default: "50"

Maximum number of ephemeral messages to keep in memory

Methods

remember()

Store data persistently on the 0G network.

async def remember(self, key: str, value: Any) -> None

Parameters:

• key (str): Unique identifier for the stored data
• value (Any): The data to store (will be JSON serialized)

Example:

await memory.remember('user_preferences', {
    'language': 'English',
    'theme': 'dark',
    'notifications': True
})

recall()

Retrieve previously stored data from the 0G network.

async def recall(self, key: str) -> Any

Parameters:

• key (str): The key of the data to retrieve

Returns: Any - The stored data, or None if not found

Example:

preferences = await memory.recall('user_preferences')
if preferences:
    print(f"User language: {preferences['language']}")

forget()

Remove data from persistent storage.

async def forget(self, key: str) -> None

Parameters:

• key (str): The key of the data to remove

Example:

await memory.forget('temporary_data')

Examples

Basic Storage

import asyncio
from zg_ai_sdk import create_agent

async def main():
    agent = await create_agent({
        'name': 'Storage Assistant',
        'provider_address': '0xf07240Efa67755B5311bc75784a061eDB47165Dd',
        'memory_bucket': 'storage-demo',
        'private_key': 'your-private-key'
    })
    
    # Store user information
    await agent.remember('user_profile', {
        'name': 'Alice',
        'age': 30,
        'interests': ['AI', 'Python', 'Blockchain']
    })
    
    # Store application settings
    await agent.remember('app_settings', {
        'version': '1.0.0',
        'debug_mode': False,
        'api_timeout': 30
    })
    
    # Retrieve and use stored data
    profile = await agent.recall('user_profile')
    settings = await agent.recall('app_settings')
    
    print(f"User: {profile['name']}, Interests: {profile['interests']}")
    print(f"App Version: {settings['version']}")

asyncio.run(main())

Ephemeral Storage

For temporary data that doesn’t need persistence:

set_ephemeral()

Store data in temporary memory.

def set_ephemeral(self, key: str, value: Any) -> None

get_ephemeral()

Retrieve data from temporary memory.

def get_ephemeral(self, key: str) -> Any

delete_ephemeral()

Remove data from temporary memory.

def delete_ephemeral(self, key: str) -> bool

Example:

# Temporary session data
memory.set_ephemeral('session_id', 'abc123')
memory.set_ephemeral('temp_calculation', {'result': 42, 'steps': [1, 2, 3]})

# Retrieve
session_id = memory.get_ephemeral('session_id')
calculation = memory.get_ephemeral('temp_calculation')

# Clean up
memory.delete_ephemeral('temp_calculation')

Conversation Storage

remember_conversation()

Save the current conversation to persistent storage.

async def remember_conversation(self, conversation_id: str) -> None

recall_conversation()

Load a previously saved conversation.

async def recall_conversation(self, conversation_id: str) -> List[ChatMessage]

Example:

# Save current conversation
await memory.remember_conversation('important_chat_001')

# Load conversation later
messages = await memory.recall_conversation('important_chat_001')
for msg in messages:
    print(f"{msg.role}: {msg.content}")

Data Types and Serialization

The Memory system automatically handles JSON serialization for common Python types:

Supported Types

• Basic types: str, int, float, bool, None
• Collections: list, dict, tuple
• Datetime objects (converted to ISO strings)
• Custom objects (if JSON serializable)

Custom Serialization

import json
from datetime import datetime

class CustomData:
    def __init__(self, name, created_at):
        self.name = name
        self.created_at = created_at
    
    def to_dict(self):
        return {
            'name': self.name,
            'created_at': self.created_at.isoformat()
        }
    
    @classmethod
    def from_dict(cls, data):
        return cls(
            name=data['name'],
            created_at=datetime.fromisoformat(data['created_at'])
        )

# Store custom object
custom_obj = CustomData('Test', datetime.now())
await memory.remember('custom_data', custom_obj.to_dict())

# Retrieve custom object
data = await memory.recall('custom_data')
restored_obj = CustomData.from_dict(data)

Error Handling

from zg_ai_sdk import SDKError

try:
    await memory.remember('test_key', {'data': 'value'})
    result = await memory.recall('test_key')
except SDKError as e:
    if e.code == 'STORAGE_ERROR':
        print("Failed to access storage")
    elif e.code == 'NETWORK_ERROR':
        print("Network connection failed")
    else:
        print(f"Storage error: {e.message}")

Best Practices

1. Key Naming: Use descriptive, hierarchical keys like user_123_preferences
2. Data Size: Keep individual items under 1MB for optimal performance
3. Batch Operations: Group related data to minimize storage calls
4. Error Handling: Always handle potential storage failures gracefully
5. Data Validation: Validate data before storing to prevent corruption

Performance Considerations

• Caching: Frequently accessed data is cached locally
• Compression: Large data is automatically compressed
• Batching: Multiple operations are batched when possible
• Retry Logic: Automatic retry for transient network failures

Next Steps

• Retrieve Memory
• Search Memory
• Agent Integration