Async Conversation Store

Implement non-blocking conversation persistence with the AsyncConversationStore protocol

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

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

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.

---

Related

Async DB Hooks

Event-driven persistence hooks for async stores

Persistence Overview

Architecture and backend options