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

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

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

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

# ✅ 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

# ✅ 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

Related