​

Persistence Overview

​

Quick Start

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    memory={
        "db": "postgresql://localhost/mydb",
        "session_id": "my-session"
    }
)
response = agent.start("Hello!")  # Auto-persists
print(response)

​

Installation

pip install "praisonaiagents[tools]"

​

Supported Backends

​

Backend Aliases

• neon, cockroachdb, xata → postgres
• asyncpg, postgres_async → async_postgres
• aiomysql, mysql_async → async_mysql
• sqlite_sync → sync_sqlite
• aiosqlite, sqlite_async → async_sqlite
• libsql → turso

• chromadb → chroma
• mongodb_atlas, cosmos, azure_cosmos → Vector store variants
• llama_index, langchain_adapter → Framework adapters

• motor, mongodb_async → async_mongodb (fully implements the StateStore contract as of PR #4287 — hash / TTL / keys operations are now available on the async backend). See Async MongoDB (motor).

​

Non-blocking prefix scan (Redis, Upstash)

​

Architecture

┌─────────────────────────────────────────┐
│           praisonaiagents.Agent         │
│         (memory={db: "..."})            │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│     Memory/Knowledge/State Adapters     │
│         (DbAdapter Protocol)            │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│        Database Backends Layer          │
├─────────────┬─────────────┬─────────────┤
│ Conversation│  Knowledge  │    State    │
│   Store     │    Store    │    Store    │
└─────────────┴─────────────┴─────────────┘

​

Two Entry Points, One Owner

• PraisonAIDB (the live path) owns the lifecycle-hook and store-dispatch logic — on_message / aon_message, on_agent_end / aon_agent_end, and the sync/async dispatch decision. MemoryConfig(db=PraisonAIDB(...)) routes straight to it and is the recommended user-facing wiring.
• PersistenceOrchestrator delegates its store-write side to a PraisonAIDB instance built lazily from its own stores (internal PraisonAIDB._from_stores(...) plumbing), while keeping its unique surface: the resume-aware on_agent_start return contract, the LRU session cache, and the knowledge / state / context helpers.
• The orchestrator path (via wrap_agent_with_persistence, PersistentAgent, or create_persistent_session) continues to work unchanged and now shares the same store-write logic under the hood.

​

Key Features

• Zero Config: SQLite works out of the box with memory=True
• Session Resume: praisonai session resume <id> is a first-class restore — it brings back chat history, model, and agent name, not just a transcript. Pass a follow-up prompt to continue immediately: praisonai session resume <id> "<prompt>".
• Runtime State Mirroring: lightweight per-turn artefacts for native↔plugin handoff (opt-in via SessionConfig.mirror_runtime_state)
• Lazy Loading: No performance impact until used
• CLI Support: praisonai persistence doctor/run/resume

​

Concurrency & Thread Safety

​

Async-store-safe session export / import (PR #5380)

​

Non-lossy tool-call persistence (PR #3837)

​

SQLite conversation store — clean shutdown across threads

​

Session Cache Size

# Raise the cap for a chat server that juggles many concurrent sessions
export PRAISONAI_SESSION_CACHE_MAX=8192

​

Shutting Down the Orchestrator

from praisonaiagents import Agent
from praisonai.persistence import PersistenceOrchestrator

# Sync agent / script: context manager
with PersistenceOrchestrator.from_env() as orch:
    agent = Agent(name="Assistant", memory={"orchestrator": orch})
    agent.start("Hello!")
# stores closed on exit, per-store errors isolated

import asyncio
from praisonaiagents import Agent
from praisonai.persistence import PersistenceOrchestrator

async def main():
    async with PersistenceOrchestrator.from_env() as orch:
        agent = Agent(name="Assistant", memory={"orchestrator": orch})
        await agent.astart("Hello!")
    # stores closed on exit — never blocks the event loop

asyncio.run(main())

from contextlib import asynccontextmanager
from fastapi import FastAPI
from praisonai.persistence import PersistenceOrchestrator

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.orch = PersistenceOrchestrator.from_env()
    yield
    await app.state.orch.aclose()  # non-blocking, isolates per-store errors

app = FastAPI(lifespan=lifespan)

• close() (sync) — closes each of conversation / knowledge / state in turn. Per-store errors are isolated so one failing store never strands the others’ pooled connections. A store’s attribute is reset to None only on a successful close, so a re-close is idempotent; a store that raised keeps its reference and can be retried by a later close() / aclose() instead of silently leaking its pooled connections.
• aclose() (async, new in PR #5348) — awaits each store’s native aclose when available, else offloads close via asyncio.to_thread. Same per-store error isolation and idempotence as close(). Safe to call from a live event loop — close() is not (it routes through run_sync_or_offload, which raises from a running loop).
• Async context manager (async with, new in PR #5348) — __aenter__ returns self, __aexit__ calls aclose(). Pair with FastAPI lifespans or async agent runners.
• Sync context manager (with) — already existed and continues to call close() on exit.

​

Schema Versioning (PR #1829)

• Standardized table schemas for sessions and messages
• Dialect-specific type mappings (_id_type, _json_type, _float_type)
• Unified retry logic with max_retries and retry_delay parameters
• Automatic serverless database detection and exponential backoff
• Consistent table_prefix handling across all SQL stores

​

Schema Migration (PR #1597)

1. Pass table_prefix="praisonai_" explicitly to keep your old tables, or
2. Rename existing tables: ALTER TABLE praisonai_sessions RENAME TO praison_sessions; (and likewise for _messages).

ALTER TABLE praison_sessions ADD COLUMN state TEXT;
ALTER TABLE praison_messages ADD COLUMN tool_calls TEXT;
ALTER TABLE praison_messages ADD COLUMN tool_call_id TEXT;

​

Custom Stores & Aliases

from praisonai.persistence import get_default_registry
from praisonaiagents import Agent

def my_custom_store(connection_string, **kwargs):
    return CustomStoreImplementation(connection_string, **kwargs)

# Register on the shared default so Agent(memory={"db": "my_store://..."}) sees it
registry = get_default_registry("conversation")
registry.register_store(
    "my_store",
    my_custom_store,
    aliases=("mine", "ms", "custom_db"),
)

# All of these now resolve to my_custom_store
agent = Agent(memory={"db": "my_store://config"})
agent = Agent(memory={"db": "mine://config"})       # alias
agent = Agent(memory={"db": "ms://config"})         # alias
agent = Agent(memory={"db": "custom_db://config"})  # alias

​

Next Steps

• Quickstart - Get started in 5 minutes
• Session Resume - Continue conversations
• Session Runtime State - Per-turn runtime artefacts for replay and handoff
• Async Conversation Store - AsyncConversationStore protocol for non-blocking persistence
• Async State Store - AsyncStateStore protocol for non-blocking key-value state
• Async Knowledge Store - AsyncKnowledgeStore protocol for non-blocking vector/RAG persistence
• CLI Reference - Command-line usage
• Backend Plugins - Custom storage backends