Async Conversation Store

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

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    instructions="Help users with their tasks.",
)
agent.start("Continue our conversation from yesterday.")

The user resumes a long chat; the async store loads session history without blocking other agents on the same event loop.

Quick Start

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())

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

Component	Role
Agent	Initiates conversation requests
Orchestrator	Routes requests based on store type using `isinstance()`
AsyncConversationStore	Handles async persistence operations
Database	Stores 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

Always Use async with Context Manager

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()

Inherit AsyncConversationStore for Custom Stores

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

Don't Mix Sync and Async APIs

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

Session Resume

MySQL Conversation Store

​Quick Start

​How It Works

​Configuration Options

AsyncConversationStore API Reference

​Common Patterns

​Context Manager Usage

​Upsert Session Helper

​Custom Async Store Implementation

​Best Practices

​Related