Skip to main content
Async conversation stores provide non-blocking persistence for conversation sessions and messages, enabling high-performance multi-agent systems.

​
Quick Start

1

Use Built-in Async Store

Use a built-in async conversation store with automatic resource management:
from praisonaiagents import Agent
from praisonai.persistence.conversation import AsyncPostgresConversationStore

async def main():
    async with AsyncPostgresConversationStore(
        url="postgresql://user:pass@localhost/db"
    ) as store:
        agent = Agent(
            name="Assistant",
            instructions="Help users with their tasks",
            conversation_store=store
        )
        
        result = await agent.start("Hello, how are you?")
        print(result)

# Run the async agent
import asyncio
asyncio.run(main())
2

Implement Custom Async Store

Create a custom async store by inheriting from AsyncConversationStore:
from praisonai.persistence.conversation import AsyncConversationStore, ConversationSession, ConversationMessage
from typing import List, Optional

class CustomAsyncStore(AsyncConversationStore):
    async def create_session(self, session: ConversationSession) -> ConversationSession:
        # Your async implementation
        return session
    
    async def get_session(self, session_id: str) -> Optional[ConversationSession]:
        # Your async implementation
        return None
    
    async def add_message(self, session_id: str, message: ConversationMessage) -> ConversationMessage:
        # Your async implementation
        return message
    
    # ... implement other required methods
    
    async def close(self) -> None:
        # Clean up async resources
        pass

# Use your custom store
store = CustomAsyncStore()
agent = Agent(name="Assistant", conversation_store=store)

​
How It Works

ComponentRole
AgentInitiates conversation requests
OrchestratorRoutes requests based on store type using isinstance()
AsyncConversationStoreHandles async persistence operations
DatabaseStores session and message data

​
Configuration Options

AsyncConversationStore API Reference

Complete method signatures and configuration options

​
Common Patterns

​
Context Manager Usage

Always use async with for automatic resource management: 
async with AsyncPostgresConversationStore(url="postgresql://...") as store:
    session = await store.get_session("session_123")
    if session:
        message = ConversationMessage(
            session_id="session_123",
            role="user",
            content="Hello"
        )
        await store.add_message("session_123", message)

​
Upsert Session Helper

Use the built-in upsert_session() method for create-or-update operations: 
session = ConversationSession(session_id="new_session", user_id="user_123")
updated_session = await store.upsert_session(session)  # Creates or updates

​
Custom Async Store Implementation

When implementing a custom async store, inherit from AsyncConversationStore: 
from praisonai.persistence.conversation import AsyncConversationStore

class MyAsyncStore(AsyncConversationStore):
    # MUST inherit AsyncConversationStore for proper orchestrator dispatch
    async def create_session(self, session: ConversationSession) -> ConversationSession:
        # Implementation here
        pass

​
Best Practices

The async context manager ensures proper resource cleanup:
# ✅ Good - automatic cleanup
async with store:
    await store.get_session("id")

# ❌ Bad - manual cleanup required  
try:
    await store.get_session("id")
finally:
    await store.close()
The orchestrator uses isinstance() checks to dispatch correctly:
# ✅ Good - proper inheritance
class MyStore(AsyncConversationStore):
    pass

# ❌ Bad - will be wrapped in asyncio.to_thread()
class MyStore(ConversationStore):
    async def get_session(self):  # This won't work as expected
        pass
Use either sync or async consistently:
# ✅ Good - pure async
async with AsyncPostgresConversationStore() as store:
    session = await store.get_session("id")

# ❌ Bad - mixing patterns
store = AsyncPostgresConversationStore()
session = store.get_session("id")  # This is not the async method
Breaking Change in PR #1829: The async_* prefixed methods have been removed from async stores:
# Before (removed)
session = await store.async_get_session(session_id)

# After (current)
session = await store.get_session(session_id)
Update your code to use the non-prefixed async methods.

Async DB Hooks

Event-driven persistence hooks for async stores

Persistence Overview

Architecture and backend options