Storage Backends

Pluggable storage backends for training, sessions, and data persistence

Storage Backends

PraisonAI provides pluggable storage backends for training data, session state, and general persistence. Switch between file-based and database storage without changing your application code.

Supported Backends

Backend	Type	Dependencies	Best For
FileBackend	JSON files	None (built-in)	Development, simple deployments
SQLiteBackend	SQLite DB	None (built-in)	Production, concurrent access
PostgreSQL	Via adapters	psycopg2	Large-scale production
MySQL	Via adapters	mysql-connector	Enterprise deployments
Redis	Via adapters	redis	High-speed caching

Quick Start

File Backend (Default)

from praisonaiagents.storage import FileBackend, BaseJSONStore

# File-based storage (default, zero config)
backend = FileBackend(storage_dir="~/.praisonai/data")
store = BaseJSONStore("session.json", backend=backend)

store.save({"messages": ["Hello", "World"]})
data = store.load()

from praisonaiagents.storage import SQLiteBackend, BaseJSONStore

# SQLite for better concurrency and performance
backend = SQLiteBackend(db_path="~/.praisonai/data.db")
store = BaseJSONStore("session.json", backend=backend)

store.save({"messages": ["Hello", "World"]})
data = store.load()

from praisonaiagents.storage import get_backend

# Use factory for easy switching
backend = get_backend("sqlite", db_path="data.db")
# or
backend = get_backend("file", storage_dir="./data")

Any Component with Any Backend

All storage components support pluggable backends. Create a backend once and use it with any component:

from praisonaiagents.storage import SQLiteBackend, RedisBackend

# SQLite for production
backend = SQLiteBackend(db_path="~/.praisonai/data.db")

# Redis for distributed systems
# backend = RedisBackend(url="redis://localhost:6379", prefix="praison:", ttl=3600)

# Use with any component
from praisonai.recipe.history import RunHistory
from praisonai.cli.state.sessions import SessionManager
from praisonai.mcp_server.tool_index import MCPToolIndex
from praisonai.train.agents.storage import TrainingStorage

history = RunHistory(backend=backend)
manager = SessionManager(backend=backend)
index = MCPToolIndex(backend=backend)
storage = TrainingStorage(session_id="train-123", backend=backend)

FileBackend

JSON file-based storage. Each key becomes a separate .json file.

from praisonaiagents.storage import FileBackend

backend = FileBackend(
    storage_dir="~/.praisonai/storage",  # Directory for files
    suffix=".json",                     # File extension
    pretty=True                         # Pretty-print JSON
)

# CRUD operations
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

Features:

• Zero dependencies (uses built-in json)
• Thread-safe with atomic writes
• Human-readable files
• Easy debugging and inspection

SQLiteBackend

SQLite database storage. All data in a single .db file.

from praisonaiagents.storage import SQLiteBackend

backend = SQLiteBackend(
    db_path="~/.praisonai/storage.db",  # Database file
    table_name="praison_storage",      # Table name
    auto_create=True                   # Create table if missing
)

# Same API as FileBackend
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

# Close when done (optional, auto-closes on exit)
backend.close()

Features:

• Zero dependencies (uses built-in sqlite3)
• ACID transactions
• Better concurrent access
• Faster for large datasets
• Single file deployment

Using with Training Storage

from praisonai.train.agents.storage import TrainingStorage
from praisonaiagents.storage import SQLiteBackend

# Default: file-based storage
storage = TrainingStorage(session_id="train-123")

# With SQLite backend for better performance
backend = SQLiteBackend(db_path="training.db")
storage = TrainingStorage(
    session_id="train-123",
    backend=backend
)

# Save training iterations
storage.save_iteration(iteration)
iterations = storage.load_iterations()

Using with Learn Stores

from praisonaiagents.memory.learn.stores import PersonaStore, InsightStore
from praisonaiagents.storage import SQLiteBackend

# Default: file-based
persona_store = PersonaStore(user_id="alice")

# With SQLite backend
backend = SQLiteBackend(db_path="learn.db")
persona_store = PersonaStore(user_id="alice", backend=backend)
insight_store = InsightStore(user_id="alice", backend=backend)

# Add entries
persona_store.add_preference("Prefers concise responses")
insight_store.add_insight("User is interested in AI")

Async Support

For async applications, use AsyncBaseJSONStore:

from praisonaiagents.storage import AsyncBaseJSONStore, SQLiteBackend

backend = SQLiteBackend(db_path="data.db")
store = AsyncBaseJSONStore("session.json", backend=backend)

# Async operations
data = await store.load_async()
await store.save_async({"key": "value"})
exists = await store.exists_async()
await store.delete_async()

Custom Backends

Implement StorageBackendProtocol for custom backends:

from praisonaiagents.storage import StorageBackendProtocol
from typing import Dict, Any, List, Optional

class RedisBackend:
    """Custom Redis backend example."""
    
    def __init__(self, url: str = "redis://localhost:6379"):
        import redis
        self.client = redis.from_url(url)
    
    def save(self, key: str, data: Dict[str, Any]) -> None:
        import json
        self.client.set(key, json.dumps(data))
    
    def load(self, key: str) -> Optional[Dict[str, Any]]:
        import json
        value = self.client.get(key)
        return json.loads(value) if value else None
    
    def delete(self, key: str) -> bool:
        return self.client.delete(key) > 0
    
    def list_keys(self, prefix: str = "") -> List[str]:
        pattern = f"{prefix}*" if prefix else "*"
        return [k.decode() for k in self.client.keys(pattern)]
    
    def exists(self, key: str) -> bool:
        return self.client.exists(key) > 0

# Use custom backend
backend = RedisBackend("redis://localhost:6379")
store = BaseJSONStore("session.json", backend=backend)

Protocol Reference

from typing import Protocol, Dict, Any, List

class StorageBackendProtocol(Protocol):
    """Protocol for pluggable storage backends."""
    
    def save(self, key: str, data: Dict[str, Any]) -> None:
        """Save data with the given key."""
        ...
    
    def load(self, key: str) -> Any:
        """Load data by key. Returns None if not found."""
        ...
    
    def delete(self, key: str) -> bool:
        """Delete data by key. Returns True if deleted."""
        ...
    
    def list_keys(self, prefix: str = "") -> List[str]:
        """List all keys, optionally filtered by prefix."""
        ...
    
    def exists(self, key: str) -> bool:
        """Check if a key exists."""
        ...

RedisBackend

Redis-based storage for high-speed caching and ephemeral data.

from praisonaiagents.storage import RedisBackend

backend = RedisBackend(
    url="redis://localhost:6379",  # Redis connection URL
    prefix="praison:",              # Key prefix
    ttl=3600,                       # Optional TTL in seconds
    db=0                            # Redis database number
)

# CRUD operations
backend.save("session_123", {"user": "alice", "messages": []})
data = backend.load("session_123")
backend.delete("session_123")
keys = backend.list_keys(prefix="session_")
exists = backend.exists("session_123")

# Set TTL on specific key
backend.set_ttl("session_123", 1800)  # 30 minutes

# Clear all data with prefix
backend.clear()

# Close connection
backend.close()

Features:

• Sub-millisecond latency
• Built-in TTL support
• Automatic key prefixing
• Requires redis package: pip install redis

Using with RunHistory (Recipe History)

Store recipe run history with pluggable backends:

from praisonai.recipe.history import RunHistory
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based storage in ~/.praisonai/runs/
history = RunHistory()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/runs.db")
history = RunHistory(backend=backend)

# With Redis backend (for distributed systems)
backend = RedisBackend(url="redis://localhost:6379", prefix="runs:")
history = RunHistory(backend=backend)

# Store and retrieve runs
run_id = history.store(result, input_data={"query": "test"})
run_data = history.get(run_id)
history.delete(run_id)

Using with SessionManager (Session State)

Persist CLI session state with different backends:

from praisonai.cli.state.sessions import SessionManager
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based in ~/.praisonai/sessions/
manager = SessionManager()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/sessions.db")
manager = SessionManager(backend=backend)

# With Redis backend (for shared sessions)
backend = RedisBackend(url="redis://localhost:6379", prefix="session:")
manager = SessionManager(backend=backend)

# Session operations
session = manager.create(run_context)
manager.append_event(session.session_id, {"type": "message", "content": "Hello"})
events = manager.get_events(session.session_id)
sessions = manager.list(limit=10)

Using with MCPToolIndex (Tool Index)

Store MCP tool schemas with pluggable backends:

from praisonai.mcp_server.tool_index import MCPToolIndex
from praisonaiagents.storage import SQLiteBackend, RedisBackend

# Default: file-based in ~/.praisonai/mcp/
index = MCPToolIndex()

# With SQLite backend
backend = SQLiteBackend(db_path="~/.praisonai/mcp.db")
index = MCPToolIndex(backend=backend)

# With Redis backend (for shared tool registry)
backend = RedisBackend(url="redis://localhost:6379", prefix="mcp:")
index = MCPToolIndex(backend=backend)

# Tool operations
index.sync("brave-search", tools=[...])
tools = index.list_tools("brave-search")
status = index.get_status("brave-search")

Choosing a Backend

Use Case	Recommended Backend	Why
Development	FileBackend	Easy debugging, human-readable
Single-user app	SQLiteBackend	Better performance, single file
Multi-process	SQLiteBackend	Handles concurrent access
Production web	PostgreSQL adapter	Scalable, reliable
High-speed cache	RedisBackend	Sub-ms latency, TTL support
Serverless	SQLiteBackend or cloud DB	No server management
Session caching	RedisBackend	Fast, ephemeral data
Distributed systems	RedisBackend	Shared state across nodes
Recipe history	SQLiteBackend	Reliable, queryable
MCP tool registry	SQLiteBackend or RedisBackend	Fast lookups

Backend Comparison

Feature	FileBackend	SQLiteBackend	RedisBackend
Dependencies	None	None	redis package
Concurrency	File locks	ACID transactions	Native
Performance	Good	Better	Best
TTL Support	❌	❌	✅
Distributed	❌	❌	✅
Human-readable	✅	❌	❌
Single file	❌	✅	N/A

Related

Database Overview

All supported databases

Memory Storage

Agent memory persistence

Training

Agent training with storage

Session Resume

Continue conversations