Retrieve

Retrieve Memory

Retrieve stored data from the 0G decentralized storage network.

Overview

The Memory system provides efficient retrieval of both ephemeral and persistent data. Data is automatically cached for better performance and supports various retrieval patterns for different use cases.

Methods

recall()

Retrieve data from persistent storage.

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

Parameters:

• key (str): The unique identifier for the stored data

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

get_ephemeral()

Retrieve data from ephemeral (temporary) storage.

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

Parameters:

• key (str): The key for the ephemeral data

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

get_messages()

Retrieve current conversation messages.

def get_messages(self) -> List[ChatMessage]

Returns: List[ChatMessage] - List of messages in the current conversation

Examples

Basic Retrieval

import asyncio
from zg_ai_sdk import create_agent

async def main():
    agent = await create_agent({
        'name': 'Retrieval Assistant',
        'provider_address': '0xf07240Efa67755B5311bc75784a061eDB47165Dd',
        'memory_bucket': 'retrieval-demo',
        'private_key': 'your-private-key'
    })
    
    # Store some data first
    await agent.remember('user_name', 'Alice')
    await agent.remember('user_age', 25)
    await agent.remember('user_skills', ['Python', 'JavaScript', 'AI'])
    
    # Retrieve individual items
    name = await agent.recall('user_name')
    age = await agent.recall('user_age')
    skills = await agent.recall('user_skills')
    
    print(f"Name: {name}")
    print(f"Age: {age}")
    print(f"Skills: {', '.join(skills)}")
    
    # Try to retrieve non-existent data
    missing = await agent.recall('non_existent_key')
    print(f"Missing data: {missing}")  # Will print: None

asyncio.run(main())

Conversation Retrieval

recall_conversation()

Retrieve a previously saved conversation.

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

Parameters:

• conversation_id (str): The ID of the conversation to retrieve

Returns: List[ChatMessage] - List of messages in the conversation

Example:

# Retrieve saved conversation
messages = await memory.recall_conversation('important_meeting_001')

print(f"Conversation has {len(messages)} messages:")
for msg in messages:
    timestamp = msg.timestamp.strftime('%H:%M:%S') if msg.timestamp else 'N/A'
    print(f"[{timestamp}] {msg.role}: {msg.content}")

get_conversation_context()

Get conversation context as formatted string.

def get_conversation_context(self) -> str

Returns: str - Formatted conversation context

Example:

context = memory.get_conversation_context()
print("Current conversation context:")
print(context)

Advanced Retrieval Patterns

Hierarchical Data Retrieval

async def get_nested_data(agent, base_key):
    """Retrieve hierarchical data structure"""
    # Get the main object
    main_data = await agent.recall(base_key)
    
    if not main_data:
        return None
    
    # Get related objects
    if 'related_keys' in main_data:
        for related_key in main_data['related_keys']:
            related_data = await agent.recall(related_key)
            main_data[f'related_{related_key}'] = related_data
    
    return main_data

# Usage
project_data = await get_nested_data(agent, 'project_001')

Fallback Retrieval

async def get_with_fallback(agent, primary_key, fallback_key, default_value=None):
    """Try primary key, then fallback, then default"""
    # Try primary key
    data = await agent.recall(primary_key)
    if data is not None:
        return data
    
    # Try fallback key
    data = await agent.recall(fallback_key)
    if data is not None:
        return data
    
    # Return default
    return default_value

# Usage
user_theme = await get_with_fallback(
    agent, 
    'user_123_theme', 
    'default_theme', 
    'light'
)

Versioned Retrieval

async def get_latest_version(agent, base_key):
    """Get the latest version of versioned data"""
    # Get version index
    version_index = await agent.recall(f'{base_key}_versions') or []
    
    if not version_index:
        return None
    
    # Get latest version
    latest_version = max(version_index, key=lambda v: v['timestamp'])
    return await agent.recall(f"{base_key}_v{latest_version['version']}")

# Usage
latest_document = await get_latest_version(agent, 'document_001')

Error Handling and Validation

from zg_ai_sdk import SDKError

async def safe_recall(agent, key, expected_type=None, validator=None):
    """Safely retrieve and validate data"""
    try:
        data = await agent.recall(key)
        
        if data is None:
            return None
        
        # Type checking
        if expected_type and not isinstance(data, expected_type):
            print(f"Warning: Expected {expected_type}, got {type(data)}")
            return None
        
        # Custom validation
        if validator and not validator(data):
            print(f"Warning: Data validation failed for key {key}")
            return None
        
        return data
        
    except SDKError as e:
        print(f"Storage error retrieving {key}: {e.message}")
        return None
    except Exception as e:
        print(f"Unexpected error retrieving {key}: {e}")
        return None

# Usage with validation
def validate_user_data(data):
    required_fields = ['name', 'email']
    return all(field in data for field in required_fields)

user_data = await safe_recall(
    agent, 
    'user_123', 
    expected_type=dict, 
    validator=validate_user_data
)

Performance Optimization

Bulk Retrieval

import asyncio

async def bulk_recall(agent, keys):
    """Retrieve multiple keys concurrently"""
    tasks = [agent.recall(key) for key in keys]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return {
        key: result if not isinstance(result, Exception) else None
        for key, result in zip(keys, results)
    }

# Usage
keys = ['user_1', 'user_2', 'user_3', 'settings', 'config']
data = await bulk_recall(agent, keys)

Lazy Loading

class LazyData:
    def __init__(self, agent, key):
        self.agent = agent
        self.key = key
        self._data = None
        self._loaded = False
    
    async def get(self):
        if not self._loaded:
            self._data = await self.agent.recall(self.key)
            self._loaded = True
        return self._data
    
    def is_loaded(self):
        return self._loaded

# Usage
lazy_config = LazyData(agent, 'large_config')
# Data is only loaded when needed
config = await lazy_config.get()

Memory Statistics

Get information about memory usage:

stats = memory.get_stats()
print(f"Ephemeral messages: {stats['ephemeral_messages']}")
print(f"Ephemeral data items: {stats['ephemeral_data']}")
print(f"Max ephemeral messages: {stats['max_ephemeral_messages']}")

Best Practices

1. Check for None: Always check if retrieved data is None
2. Use Type Hints: Specify expected return types for better code clarity
3. Implement Caching: Cache frequently accessed data locally
4. Handle Errors: Always handle potential storage errors gracefully
5. Validate Data: Validate retrieved data before using it
6. Use Batch Operations: Retrieve multiple items concurrently when possible

Next Steps

• Search Memory
• Store Memory
• Agent Integration