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

Fixed in PR #3588 (0ba6829). When an AsyncConversationStore was wired into a sync agent through PersistenceOrchestrator, the very first create_session call was silently dropped — the sync helper was calling the async store.create_session(...) and discarding the returned coroutine. The orchestrator now passes an explicit create_session= dispatcher that routes the coroutine through _sync, so the session row is written on the first turn regardless of whether your store is sync or async. Custom AsyncConversationStore subclasses need no code changes.

Session export / import

PraisonAIDB.aexport_session / aimport_session give async callers a native await path against an async store (PR #5380):

import asyncio
from praisonai.persistence.conversation import AsyncPostgresConversationStore
from praisonai.db import PraisonAIDB

async def main():
    async with AsyncPostgresConversationStore(
        url="postgresql://user:pass@localhost/db"
    ) as store:
        db = PraisonAIDB._from_stores(conversation_store=store)

        payload = await db.aexport_session("session-42")

        new_sid = await db.aimport_session({
            "session_id": "restored-42",
            "messages": [{"role": "user", "content": "ping"}],
        })
        print(new_sid)

asyncio.run(main())

From a sync caller you can still use db.export_session(...) / db.import_session(...) — since PR #5380 they dispatch correctly against an async store instead of leaking un-awaited coroutines. The async twins are for code already running inside an event loop.

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

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Async Conversation Store

Quick Start

How It Works

Configuration Options

AsyncConversationStore API Reference

Common Patterns

Context Manager Usage

Upsert Session Helper

Custom Async Store Implementation

Session export / import

Best Practices

Async DB Hooks

Persistence Overview

​Quick Start

​How It Works

​Configuration Options

AsyncConversationStore API Reference

​Common Patterns

​Context Manager Usage

​Upsert Session Helper

​Custom Async Store Implementation

​Session export / import

​Best Practices

​Related

Async DB Hooks

Persistence Overview

Quick Start

How It Works

Configuration Options

Common Patterns

Context Manager Usage

Upsert Session Helper

Custom Async Store Implementation

Session export / import

Best Practices

Related