Database Persistence

Persist conversations, runs, and traces to PostgreSQL, SQLite, Redis, and vector stores

Persist every agent turn to external databases — conversations resume across restarts and you can query run history later.

import os
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(
        db=db(database_url=os.getenv("PRAISON_CONVERSATION_URL")),
        session_id="user-123",
    ),
)
agent.start("Hello!")

The user continues a conversation; database hooks persist each turn so the session survives restarts.

Quick Start

1. Simple usage

   import os
   from praisonaiagents import Agent
   from praisonaiagents import db
   
   agent = Agent(
       name="Assistant",
       instructions="You are a helpful assistant.",
       memory=db(database_url=os.getenv("PRAISON_CONVERSATION_URL", "sqlite:///conversations.db")),
   )
   agent.start("Hello!")

2. With session continuity

   import os
   from praisonaiagents import Agent, MemoryConfig
   from praisonaiagents import db
   
   store = db(database_url=os.getenv("PRAISON_CONVERSATION_URL"))
   
   agent = Agent(
       name="Assistant",
       memory=MemoryConfig(db=store, session_id="session-001"),
   )
   agent.start("My favourite colour is blue")
   
   # Later run — same session_id restores history
   agent2 = Agent(
       name="Assistant",
       memory=MemoryConfig(db=store, session_id="session-001"),
   )
   agent2.start("What's my favourite colour?")

3. CLI with persistence

   export PRAISON_CONVERSATION_URL="postgresql://localhost/praisonai"
   praisonai persistence doctor
   praisonai persistence run --session-id demo "Remember my name is Alice"
   praisonai persistence resume --session-id demo --continue "What's my name?"

How It Works

On first chat, the adapter loads prior messages for the session_id. Each turn writes user and agent messages automatically — no manual save calls.

Configuration Options

Pass a db() instance via memory= (or MemoryConfig(db=…, session_id=…) for explicit sessions).

The three per-store dicts mirror PersistenceConfig and each routes to exactly one store factory.

import os
from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db

agent = Agent(
    name="Researcher",
    memory=MemoryConfig(
        db=db(
            database_url=os.getenv("PRAISON_CONVERSATION_URL"),
            state_url=os.getenv("PRAISON_STATE_URL"),
            knowledge_url=os.getenv("PRAISON_KNOWLEDGE_URL"),
        ),
        session_id="research-42",
    ),
)

Multi-backend configuration

Per-store option dicts send backend-specific kwargs to a single store, so mixing Postgres, Redis, and Qdrant never leaks an unrelated kwarg.

from praisonaiagents import Agent, MemoryConfig, db

store = db(
    database_url="postgresql://localhost/praisonai",
    state_url="redis://localhost:6379",
    knowledge_url="http://localhost:6333",
    conversation_options={"sslmode": "require"},   # Postgres-only
    state_options={"decode_responses": True},       # Redis-only
    knowledge_options={"prefer_grpc": True},         # Qdrant-only
)

agent = Agent(name="Assistant", memory=MemoryConfig(db=store, session_id="s1"))

Each kwarg reaches exactly one factory:

Environment variables

export PRAISON_CONVERSATION_URL="postgresql://localhost:5432/praisonai"
export PRAISON_STATE_URL="redis://localhost:6379"
export PRAISON_KNOWLEDGE_URL="http://localhost:6333"

Default session ID

If you omit session_id, PraisonAI generates a fresh ID per instance (UTC hour + agent hash + random suffix):

Format: YYYYMMDDHH-{agent_hash[:6]}-{uuid4_hex[:8]}
Example: 2026072914-a1b2c3-9f4e7d21

Custom DB Adapter

Any object with the right method shape works as a DB adapter — no base class, no import required.

from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    name="Assistant",
    instructions="Help the user.",
    memory=MemoryConfig(db=MyDbAdapter(), session_id="chat-42"),
)
agent.start("Read a.txt and summarise it")

The adapter persists each turn and returns prior messages on on_agent_start, so the session resumes across restarts.

Required vs optional methods

The two optional methods (PR #5082) let a resumed session hand the model the same tool transcript the JSON store already keeps.

Minimal adapter

class MyDbAdapter:
    def on_agent_start(self, agent_name, session_id, user_id=None, metadata=None):
        return []  # return DbMessage list to resume

    def on_user_message(self, session_id, content, metadata=None):
        ...

    def on_agent_message(self, session_id, content, metadata=None):
        ...

    # Optional — implement these to persist tool turns faithfully
    def on_assistant_message(self, session_id, content, tool_calls=None, metadata=None):
        ...

    def on_tool_message(self, session_id, content, tool_call_id, metadata=None):
        ...

Reference: PraisonAI PR #5082 (fixes #5075).

Docker (local development)

docker run -d --name praison-postgres -p 5432:5432 \
  -e POSTGRES_PASSWORD=praison123 \
  -e POSTGRES_DB=praisonai postgres:16

docker run -d --name praison-redis -p 6379:6379 redis:7
docker run -d --name praison-qdrant -p 6333:6333 qdrant/qdrant

CLI commands

praisonai persistence doctor
praisonai persistence status
praisonai persistence migrate
praisonai persistence export --session-id my-session --output backup.jsonl
praisonai persistence import --file backup.jsonl

Async-Safe Initialisation

The DatabaseAdapter‘s async callbacks (on_agent_start, on_user_message, on_agent_message, on_tool_call, on_agent_end) never block the event loop. Store construction runs off-loop via asyncio.to_thread, so FastAPI handlers, Jupyter notebooks, and async test suites all work without stalls on cold startup.

Transient Failure Handling

When a database is temporarily unavailable (bad config, network blip, cloud auth glitch), the adapter records the failure and waits before re-attempting. This prevents hammering a down backend on every agent callback while still recovering automatically once the backend is back.

The cool-down period is configurable via init_retry_cooldown (default 30 seconds):

from praisonai.db.adapter import DatabaseAdapter

adapter = DatabaseAdapter(
    database_url="postgresql://localhost/praisonai",
    init_retry_cooldown=10.0,  # try again 10s after a failed init
)

After a soft init failure (raised during _ainit_stores()), the adapter pauses for init_retry_cooldown seconds before re-running store construction on the next callback. Persistence resumes automatically once the backend becomes reachable — no process restart needed.

Best Practices

Related