Run History

Read back persisted run and trace data from your state store

Query runs and traces your agents have written to the state store.

from praisonaiagents import Agent
from praisonaiagents.db import PraisonAIDB

agent = Agent(name="Research Agent", instructions="Research the topic")
agent.start("What is quantum computing?")

db = PraisonAIDB(state_url="sqlite:///agent_state.db")
runs = db.get_runs(session_id=agent.session_id)
print(f"{len(runs)} run(s) for this session")

The user runs an agent, then reads persisted runs and traces from the state store for that session.

Quick Start

1. Configure State Store

   Set PRAISONAI_STATE_URL or pass state_url= when creating agents. See Database Persistence for configuration options.

   import os
   os.environ["PRAISONAI_STATE_URL"] = "sqlite:///agent_state.db"

2. Query Runs

   Run an agent, then read its runs back using get_runs(session_id):

   from praisonaiagents import Agent
   from praisonaiagents.db import PraisonAIDB
   
   # Run an agent
   agent = Agent(name="Research Agent", instructions="Research the topic")
   result = agent.start("What is quantum computing?")
   
   # Query runs for this session
   db = PraisonAIDB(state_url="sqlite:///agent_state.db")
   runs = db.get_runs(session_id=agent.session_id)
   
   for run in runs:
       print(f"Run {run['run_id']}: {run['status']} at {run['started_at']}")

3. Query Traces with Filters

   Use get_traces() with optional filters for session_id, user_id, and limit:

   # Get traces for a specific session
   traces = db.get_traces(session_id=agent.session_id, limit=10)
   
   # Get traces for a specific user
   traces = db.get_traces(user_id="user123", limit=5)
   
   # Get all traces (up to limit)
   all_traces = db.get_traces(limit=100)

How It Works

Both methods read persisted data from the configured state store:

Configuration Options

def get_runs(session_id: str, limit: Optional[int] = None) -> List[dict]:
    """
    Get runs for a session.

    Args:
        session_id: Session ID to filter by (required)
        limit: Maximum number of runs to return (optional)

    Returns:
        List of run dictionaries sorted by started_at desc
    """

def get_traces(
    session_id: Optional[str] = None, 
    user_id: Optional[str] = None, 
    limit: Optional[int] = None
) -> List[dict]:
    """
    Get traces with optional filters.

    Args:
        session_id: Filter by session ID (optional)
        user_id: Filter by user ID (optional)
        limit: Maximum number of traces to return (optional)

    Returns:
        List of trace dictionaries sorted by started_at desc
    """

Behavior:

• If state_store is not configured, returns [] and logs a warning
• limit=0 returns [], limit=None returns everything matching
• Bad/non-dict entries are skipped with a warning
• Uses state_store.scan_prefix(prefix) when available (non-blocking cursor SCAN on Redis and Upstash; prefix-scoped keys() fallback on other backends). This replaces the previous KEYS praison:* pattern, so a “recent runs” read never blocks other clients on the same Redis instance.

Common Patterns

Show last 10 runs for a user:

# Get user's recent sessions first, then query runs
recent_sessions = get_user_sessions(user_id="user123")  # Your implementation
for session in recent_sessions[:5]:  # Last 5 sessions
    runs = db.get_runs(session_id=session['session_id'], limit=2)
    print(f"Session {session['session_id']}: {len(runs)} runs")

Export traces for a session:

import json

# Export complete session trace data
traces = db.get_traces(session_id="session_abc123")
with open("session_traces.json", "w") as f:
    json.dump(traces, f, indent=2, default=str)

Best Practices

import os
os.environ["PRAISONAI_STATE_URL"] = os.getenv("DATABASE_URL", "sqlite:///data.db")

db = PraisonAIDB(state_url=os.getenv("DATABASE_URL", "sqlite:///data.db"))

# Good for UI pagination
recent_runs = db.get_runs(session_id, limit=20)

# Avoid large scans in UI code
all_runs = db.get_runs(session_id)  # Could be thousands

# Cache for dashboard views
@lru_cache(maxsize=100)
def get_cached_runs(session_id: str, limit: int = 10):
    return tuple(db.get_runs(session_id, limit))  # Tuple for hashability

Related