Skip to main content
Persistence backend plugins extend PraisonAI with custom storage solutions for conversations, knowledge, and state management through a centralized registry system.

​
Quick Start

1

Register a Custom Backend

from praisonai.persistence.registry import CONVERSATION_STORES

def my_store_factory(url=None, **kwargs):
    from my_pkg import MyConversationStore
    return MyConversationStore(url=url, **kwargs)

CONVERSATION_STORES.register("mybackend", my_store_factory, aliases=("mb",))

store = CONVERSATION_STORES.create("mb", url="proto://host")
2

Create an Entry-Point Plugin

[project.entry-points."praisonai.conversation_stores"]
mybackend = "my_pkg.factory:create_store"

[project.entry-points."praisonai.knowledge_stores"]
mybackend = "my_pkg.factory:create_kstore"

[project.entry-points."praisonai.state_stores"]
mybackend = "my_pkg.factory:create_sstore"

​
How It Works

The plugin system provides three global registries for different storage types:
RegistryKindEntry-point GroupPurpose
CONVERSATION_STORES"conversation"praisonai.conversation_storesChat history and sessions
KNOWLEDGE_STORES"knowledge"praisonai.knowledge_storesVector databases and RAG
STATE_STORES"state"praisonai.state_storesApplication state and cache

​
Registry API Reference

Each StoreRegistry provides a thread-safe API for backend management:
MethodSignatureDescription
registerregister(name: str, factory: Callable, *, aliases=()) -> NoneRegister a backend factory with optional aliases
createcreate(name: str, **kwargs) -> AnyCreate a store instance by name or alias
list_registered() -> list[str]Get all registered backend names (sorted)
list_aliases() -> dict[str, str]Get alias → canonical name mappings
default@classmethod default(cls) -> "PluginRegistry[T]"New in PR #1829: Process-default registry instance (thread-safe, per-subclass)

​
Built-in Backends

​
Conversation Stores

Primary backends: postgres, async_postgres, mysql, async_mysql, MySQLConversationStore, sqlite, sync_sqlite, async_sqlite, json, singlestore, supabase, surrealdb, turso
New in PR #1829: MySQLConversationStore is available from praisonai.persistence.conversation.mysql_new but not re-exported from the main package. It inherits the unified schema from _SQLConversationStoreBase.
Aliases available:
  • neon, cockroachdb, crdb, cockroach, xata → postgres
  • asyncpg, postgres_async → async_postgres
  • aiomysql, mysql_async → async_mysql
  • sqlite_sync → sync_sqlite
  • aiosqlite, sqlite_async → async_sqlite
  • libsql → turso

​
Knowledge Stores

Primary backends: chroma, qdrant, pinecone, weaviate, lancedb, milvus, pgvector, redis, cassandra, clickhouse, mongodb_vector, couchbase, singlestore_vector, surrealdb_vector, upstash_vector, lightrag, langchain, llamaindex, cosmosdb Aliases available:
  • chromadb → chroma
  • mongodb_atlas, mongo_vector → mongodb_vector
  • singlestore_v → singlestore_vector
  • surrealdb_v → surrealdb_vector
  • upstash_v → upstash_vector
  • langchain_adapter → langchain
  • llama_index, llamaindex_adapter → llamaindex
  • cosmos, azure_cosmos, cosmosdb_vector → cosmosdb

​
State Stores

Primary backends: redis, dynamodb, firestore, mongodb, async_mongodb, upstash, memory, gcs Aliases available:
  • motor, mongodb_async → async_mongodb

​
Extending SQL Backends

For building new SQL conversation stores, inherit from _SQLConversationStoreBase to get unified schema and retry logic: 
from praisonai.persistence.conversation._sql_base import _SQLConversationStoreBase

class MyDBConversationStore(_SQLConversationStoreBase):
    # Constants for this dialect
    SCHEMA_VERSION = "1.0.0"  # Inherited from base
    _id_type = "UUID"
    _json_type = "JSONB"
    _float_type = "DOUBLE PRECISION"
    _serverless_hosts = (".mydb.cloud",)
    
    # Required dialect hooks
    def _param(self, n: int) -> str:
        return f"${n}"  # PostgreSQL-style parameters
    
    async def _connect(self, **kwargs):
        # Return connection for this dialect
        pass
    
    def _get_conn(self):
        # Get connection from pool
        pass
    
    def _put_conn(self, conn):
        # Return connection to pool  
        pass
    
    async def _execute(self, conn, query: str, *args):
        # Execute query with this dialect
        pass
    
    # ... implement other dialect hooks
The base class provides:
  • Unified table schema with SCHEMA_VERSION = "1.0.0"
  • Retry logic with configurable max_retries and retry_delay
  • Serverless detection and exponential backoff
  • Standard CRUD operations for sessions and messages

​
Process-Default Registry

Use PluginRegistry.default() for thread-safe per-subclass singleton registries: 
from praisonai._registry import PluginRegistry

class MyRegistry(PluginRegistry["MyThing"]):
    pass

# Thread-safe, per-subclass lazy initialization
reg = MyRegistry.default()

# Alternative registries are independent
class OtherRegistry(PluginRegistry["OtherThing"]):
    pass

other_reg = OtherRegistry.default()  # Different instance
This replaces the manual singleton pattern used in previous versions. Both LLMProviderRegistry.default() and FrameworkAdapterRegistry.default() now use this unified approach. The legacy module-level functions (get_default_registry(), get_default_llm_registry()) remain for backward compatibility and delegate to the respective cls.default() methods.

​
Common Patterns

from praisonai.persistence.registry import CONVERSATION_STORES

class CustomDBConversationStore:
    def __init__(self, url=None, **kwargs):
        self.url = url
        # Initialize your database connection
        
    def save_session(self, session):
        # Save implementation
        pass
        
    def load_session(self, session_id):
        # Load implementation  
        pass

def create_custom_db(url=None, **kwargs):
    return CustomDBConversationStore(url=url, **kwargs)

# Register at runtime
CONVERSATION_STORES.register("customdb", create_custom_db)

# Use it
store = CONVERSATION_STORES.create("customdb", url="custom://localhost:5432/db")

# my_storage_package/__init__.py
def register_all_backends():
    from praisonai.persistence.registry import (
        CONVERSATION_STORES, KNOWLEDGE_STORES, STATE_STORES
    )
    
    CONVERSATION_STORES.register("myconv", create_conversation_store)
    KNOWLEDGE_STORES.register("myknow", create_knowledge_store) 
    STATE_STORES.register("mystate", create_state_store)

# Auto-register on import
register_all_backends()

from praisonai.persistence.registry import (
    CONVERSATION_STORES, KNOWLEDGE_STORES, STATE_STORES,
)

print("Conversation backends:", CONVERSATION_STORES.list_registered())
print("Knowledge backends:", KNOWLEDGE_STORES.list_registered())
print("State backends:", STATE_STORES.list_registered())

# Check aliases
print("Aliases:", CONVERSATION_STORES.list_aliases())

​
Best Practices

All registries are thread-safe with threading.Lock protection. Registrations and creations can be called concurrently without external synchronization. Factory functions should also be thread-safe if used in multi-threaded environments.
Registry creation raises ValueError for unknown backends with a list of available options. Factory functions should handle their own connection errors and invalid parameters appropriately.
Built-in backends use lazy imports to avoid loading unused dependencies. Follow this pattern in custom backends to minimize startup time and reduce import-time failures.
  • Use lowercase names without special characters
  • Provide meaningful aliases for user convenience
  • Follow existing patterns: async_* for async variants, *_vector for vector stores
  • Avoid conflicts with built-in backend names

Framework Adapter Plugins

Plugin system for multi-agent frameworks

Persistence Overview

Storage backends and configuration