Skip to main content

Chat History

Manage and retrieve conversation history using the Memory system.

Overview

Chat history in the 0G AI SDK is managed through the Memory system, which provides both ephemeral (temporary) and persistent storage for conversation messages. This allows you to maintain context across sessions and build sophisticated conversational AI applications.

Memory Integration

The Chat class works seamlessly with the Memory class to store and retrieve conversation history:
from zg_ai_sdk import Agent, ChatMessage

# Agent automatically manages chat history through memory
agent = await create_agent({
    'name': 'History Assistant',
    'provider_address': '0xf07240Efa67755B5311bc75784a061eDB47165Dd',
    'memory_bucket': 'chat-history',
    'private_key': 'your-private-key'
})

Methods

get_messages()

Retrieve current conversation messages from ephemeral memory.
def get_messages(self) -> List[ChatMessage]
Returns: List[ChatMessage] - List of messages in the current conversation

add_message()

Add a message to the conversation history.
def add_message(self, message: ChatMessage) -> None
Parameters:
  • message (ChatMessage): The message to add to history

clear_ephemeral_messages()

Clear the current conversation from ephemeral memory.
def clear_ephemeral_messages(self) -> None

Examples

import asyncio
from zg_ai_sdk import create_agent, ChatMessage

async def main():
    agent = await create_agent({
        'name': 'History Assistant',
        'provider_address': '0xf07240Efa67755B5311bc75784a061eDB47165Dd',
        'memory_bucket': 'history-demo',
        'private_key': 'your-private-key'
    })
    
    # Have a conversation
    await agent.ask('Hello, my name is Alice')
    await agent.ask('What is my name?')
    
    # Get conversation history
    messages = agent.memory.get_messages()
    print(f"Conversation has {len(messages)} messages:")
    
    for msg in messages:
        print(f"{msg.role}: {msg.content}")

asyncio.run(main())

Agent History Methods

save_conversation()

Save the current conversation to persistent storage.
async def save_conversation(self, conversation_id: Optional[str] = None) -> str
Parameters:
  • conversation_id (Optional[str]): Custom ID for the conversation. If not provided, a timestamp-based ID is generated.
Returns: str - The conversation ID Example:
# Auto-generated ID
conv_id = await agent.save_conversation()

# Custom ID
conv_id = await agent.save_conversation('important_discussion')

load_conversation()

Load a previously saved conversation.
async def load_conversation(self, conversation_id: str) -> None
Parameters:
  • conversation_id (str): The ID of the conversation to load
Example:
await agent.load_conversation('important_discussion')

clear_conversation()

Clear the current conversation from ephemeral memory.
def clear_conversation(self) -> None
Example:
agent.clear_conversation()  # Start fresh

Memory Statistics

Get detailed information about memory usage:
stats = agent.memory.get_stats()
print(f"Ephemeral messages: {stats['ephemeral_messages']}")
print(f"Max ephemeral messages: {stats['max_ephemeral_messages']}")

Advanced History Management

Custom Message Filtering

def get_user_questions(agent):
    messages = agent.memory.get_messages()
    questions = []
    
    for msg in messages:
        if msg.role == 'user' and '?' in msg.content:
            questions.append(msg.content)
    
    return questions

# Get all user questions from the conversation
questions = get_user_questions(agent)
print("User asked:", questions)

Conversation Context Building

def build_context_summary(agent, max_messages=10):
    messages = agent.memory.get_messages()
    recent_messages = messages[-max_messages:] if len(messages) > max_messages else messages
    
    context = "Recent conversation:\n"
    for msg in recent_messages:
        context += f"{msg.role}: {msg.content[:100]}...\n"
    
    return context

# Get conversation summary
summary = build_context_summary(agent)
print(summary)
def search_messages(agent, keyword):
    messages = agent.memory.get_messages()
    matching_messages = []
    
    for msg in messages:
        if keyword.lower() in msg.content.lower():
            matching_messages.append(msg)
    
    return matching_messages

# Find messages containing specific keywords
python_messages = search_messages(agent, 'python')
print(f"Found {len(python_messages)} messages about Python")

Configuration

Max Ephemeral Messages

Control how many messages are kept in ephemeral memory:
agent = await create_agent({
    'name': 'Limited History Assistant',
    'provider_address': '0xf07240Efa67755B5311bc75784a061eDB47165Dd',
    'memory_bucket': 'limited-history',
    'private_key': 'your-private-key',
    'max_ephemeral_messages': 20  # Keep only last 20 messages
})

Memory Bucket Organization

Organize conversations using different memory buckets:
# Different agents for different purposes
support_agent = await create_agent({
    'memory_bucket': 'customer-support',
    # ... other config
})

research_agent = await create_agent({
    'memory_bucket': 'research-assistant', 
    # ... other config
})

Error Handling

from zg_ai_sdk import SDKError

try:
    await agent.load_conversation('non_existent_conversation')
except SDKError as e:
    if e.code == 'STORAGE_ERROR':
        print("Conversation not found")
    else:
        print(f"Error loading conversation: {e.message}")

Best Practices

  1. Regular Cleanup: Clear ephemeral messages periodically for long-running applications
  2. Meaningful IDs: Use descriptive conversation IDs for easier management
  3. Context Limits: Be mindful of token limits when using long conversation histories
  4. Backup Important Conversations: Save critical conversations to persistent storage

Next Steps