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

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)

# TrainingStorage is a context manager — closes its backend on exit
with TrainingStorage(session_id="train-123", backend=backend) as storage:
    storage.save_iteration(...)

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

TrainingStorage is a context manager — use with to close the backend deterministically.

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

# Default: file-based storage
with TrainingStorage(session_id="train-123") as storage:
    storage.save_iteration(iteration)

# With SQLite backend for better performance
with TrainingStorage(
    session_id="train-123",
    backend=SQLiteBackend(db_path="training.db"),
) as storage:
    storage.save_iteration(iteration)
    iterations = storage.load_iterations()
# Connection closed automatically — safe on Windows, no ResourceWarning

Using with (or calling .close() explicitly) releases the underlying SQLite or Redis connection so the file can be deleted or reopened immediately — essential on Windows, and best practice for long-running processes. On backends without an open resource (e.g. FileBackend) .close() is a safe no-op.

When you use a SQLiteBackend (or RedisBackend) directly — for example via list_sessions_from_backend(SQLiteBackend(db_path=db)) — close the backend yourself, or wrap the call in try / finally. TrainingStorage’s context manager only closes backends it owns.

from praisonaiagents.storage import SQLiteBackend

backend = SQLiteBackend(db_path="training.db")
try:
    sessions = list_sessions_from_backend(backend)
finally:
    backend.close()

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

Database Overview

All supported databases

Memory Storage

Agent memory persistence

Training

Agent training with storage

Session Resume

Continue conversations

​Storage Backends

​Supported Backends

​Quick Start

​Any Component with Any Backend

​FileBackend

​SQLiteBackend

​Using with Training Storage

​Using with Learn Stores

​Async Support

​Custom Backends

​Protocol Reference

​RedisBackend

​Using with RunHistory (Recipe History)

​Using with SessionManager (Session State)

​Using with MCPToolIndex (Tool Index)

​Choosing a Backend

​Backend Comparison

​Related

Database Overview

Memory Storage

Training

Session Resume

Storage Backends

Supported Backends

Quick Start

Any Component with Any Backend

FileBackend

SQLiteBackend

Using with Training Storage

Using with Learn Stores

Async Support

Custom Backends

Protocol Reference

RedisBackend

Using with RunHistory (Recipe History)

Using with SessionManager (Session State)

Using with MCPToolIndex (Tool Index)

Choosing a Backend

Backend Comparison

Related