Skip to main content

​
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

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