Redis State Store

High-performance in-memory state storage for fast data access and caching

Redis provides high-performance in-memory state storage, perfect for applications requiring fast access to agent state, session data, and real-time caching.

Quick Start

1. Basic State Storage

   from praisonaiagents import Agent, db
   
   agent = Agent(
       name="FastBot",
       instructions="You are a helpful assistant.",
       db=db(state_url="redis://localhost:6379"),
       session_id="fast-session"
   )
   
   # Agent state automatically stored in Redis
   response = agent.chat("Remember my preference for fast responses")
   print(response)  # State persisted to Redis for quick access
   
2. Combined SQL + Redis

   from praisonaiagents import Agent, db
   
   # SQL for conversations, Redis for state
   agent = Agent(
       name="HybridBot",
       instructions="You are a helpful assistant.",
       db=db(
           database_url="postgresql://user:pass@localhost/conversations",
           state_url="redis://localhost:6379"
       ),
       session_id="hybrid-session"
   )
   
   agent.chat("This message goes to PostgreSQL, state to Redis")
   

---

How It Works

Redis stores different types of state data optimized for performance:

Data Type	Use Case	Performance
Strings	Simple key-value state	O(1) get/set
Hashes	Agent metadata, user preferences	O(1) field access
Sets	Active sessions, user lists	O(1) membership tests
Sorted Sets	Leaderboards, time-based data	O(log N) range queries
JSON	Complex state objects	Native JSON operations

---

Configuration Options

Redis URL Formats

# Basic connection
db(state_url="redis://localhost:6379")

# With password
db(state_url="redis://:password@localhost:6379")

# Specific database number
db(state_url="redis://localhost:6379/1")

# Redis Cluster
db(state_url="redis://node1:6379,node2:6379,node3:6379")

# SSL connection
db(state_url="rediss://username:password@secure-redis:6380")

Advanced Redis Configuration

from praisonaiagents import Agent, db

# Custom Redis state store
redis_db = db.RedisDB(
    host="redis.example.com",
    port=6379,
    password="secure_password",
    db=0,  # Redis database number
    # Connection pool settings
    max_connections=20,
    retry_on_timeout=True,
    # Performance settings
    socket_timeout=5.0,
    socket_connect_timeout=5.0,
    # Key prefix for organization
    prefix="praisonai:"
)

agent = Agent(
    name="RedisBot",
    instructions="You are a high-performance assistant.",
    db=db(state_url=redis_db),
    session_id="redis-session"
)

---

Docker Setup

Quick Redis setup with Docker:

# Start Redis container
docker run -d \
    --name praisonai-redis \
    -p 6379:6379 \
    redis:7-alpine redis-server --appendonly yes

# Connect and verify
docker exec -it praisonai-redis redis-cli ping
# Should return: PONG

With password protection:

# Start Redis with authentication
docker run -d \
    --name praisonai-redis-auth \
    -p 6379:6379 \
    redis:7-alpine redis-server --requirepass "your_secure_password" --appendonly yes

Then use in your agent:

from praisonaiagents import Agent, db

agent = Agent(
    name="DockerBot",
    db=db(state_url="redis://:your_secure_password@localhost:6379"),
    session_id="docker-session"
)

---

State Management Patterns

Key-Value State

from praisonaiagents import Agent, db

# Direct Redis access for custom state management
redis_store = db.RedisDB(host="localhost", port=6379)

# Store simple agent state
redis_store.set("agent:preferences:user123", "theme=dark,language=en")
redis_store.set("agent:session:active", "session-456")

# Retrieve state with expiration
redis_store.setex("agent:temp:token", 3600, "temporary_auth_token")  # 1 hour TTL

# Get state
preferences = redis_store.get("agent:preferences:user123")
active_session = redis_store.get("agent:session:active")

print(f"User preferences: {preferences}")
print(f"Active session: {active_session}")

JSON State Objects

import json
from praisonaiagents import Agent, db

redis_store = db.RedisDB(host="localhost", port=6379)

# Store complex state as JSON
agent_state = {
    "user_id": "user123",
    "conversation_context": ["greeting", "question", "response"],
    "preferences": {
        "response_style": "concise",
        "topics_of_interest": ["AI", "programming", "science"]
    },
    "metrics": {
        "total_messages": 45,
        "session_start": "2024-01-15T10:30:00Z"
    }
}

# Store JSON state
redis_store.set_json("agent:state:user123", agent_state)

# Retrieve and modify JSON state
current_state = redis_store.get_json("agent:state:user123")
current_state["metrics"]["total_messages"] += 1
redis_store.set_json("agent:state:user123", current_state)

print(f"Updated state: {current_state}")

Hash-Based State

from praisonaiagents import Agent, db

redis_store = db.RedisDB(host="localhost", port=6379)

# Use Redis hashes for structured state
session_key = "session:user123"

# Set multiple fields at once
redis_store.hmset(session_key, {
    "status": "active",
    "started_at": "2024-01-15T10:30:00Z",
    "message_count": "1",
    "last_activity": "2024-01-15T10:35:00Z"
})

# Update individual fields
redis_store.hset(session_key, "message_count", "2")
redis_store.hset(session_key, "last_activity", "2024-01-15T10:40:00Z")

# Get all session data
session_data = redis_store.hgetall(session_key)
print(f"Session data: {session_data}")

# Get specific field
message_count = redis_store.hget(session_key, "message_count")
print(f"Message count: {message_count}")

---

Real-Time Features

Session Tracking

import redis
from datetime import datetime
from praisonaiagents import Agent, db

# Direct Redis client for real-time features
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def track_active_sessions():
    """Track and monitor active agent sessions"""
    
    # Add session to active set with timestamp
    session_id = "user123-session"
    timestamp = datetime.now().isoformat()
    
    # Active sessions sorted set (score = timestamp)
    r.zadd("active_sessions", {session_id: timestamp})
    
    # Session details hash
    r.hset(f"session:{session_id}", mapping={
        "user_id": "user123",
        "started_at": timestamp,
        "status": "active",
        "agent_name": "ProductionBot"
    })
    
    # Set session expiration
    r.expire(f"session:{session_id}", 7200)  # 2 hours
    
    # Get active sessions count
    active_count = r.zcard("active_sessions")
    print(f"Active sessions: {active_count}")
    
    # Get sessions from last hour
    one_hour_ago = (datetime.now().timestamp() - 3600)
    recent_sessions = r.zrangebyscore("active_sessions", one_hour_ago, "+inf")
    print(f"Sessions in last hour: {len(recent_sessions)}")

# Use with agent
agent = Agent(
    name="TrackedBot",
    db=db(state_url="redis://localhost:6379"),
    session_id="user123-session"
)

track_active_sessions()

Pub/Sub for Real-Time Updates

import redis
import threading
import json
from praisonaiagents import Agent, db

# Redis client for pub/sub
r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def agent_event_publisher(agent_id, event_type, data):
    """Publish agent events for real-time monitoring"""
    event = {
        "agent_id": agent_id,
        "event_type": event_type,
        "timestamp": datetime.now().isoformat(),
        "data": data
    }
    r.publish("agent_events", json.dumps(event))

def agent_event_subscriber():
    """Subscribe to agent events"""
    pubsub = r.pubsub()
    pubsub.subscribe("agent_events")
    
    for message in pubsub.listen():
        if message["type"] == "message":
            event = json.loads(message["data"])
            print(f"Agent Event: {event['event_type']} from {event['agent_id']}")

# Start subscriber in background
subscriber_thread = threading.Thread(target=agent_event_subscriber)
subscriber_thread.daemon = True
subscriber_thread.start()

# Create agent with event publishing
agent = Agent(
    name="EventBot",
    db=db(state_url="redis://localhost:6379"),
    session_id="event-session"
)

# Publish events during agent interactions
agent_event_publisher("EventBot", "conversation_start", {"session_id": "event-session"})
response = agent.chat("Hello!")
agent_event_publisher("EventBot", "message_processed", {"response_length": len(response)})

---

Best Practices

Memory Management

• Set appropriate maxmemory limit for your Redis instance
• Use LRU or LFU eviction policies for automatic cleanup
• Monitor memory usage and plan for peak loads
• Use key expiration (TTL) for temporary data

Data Organization

• Use consistent key naming patterns (prefix:type:identifier)
• Group related data using key prefixes or Redis databases
• Consider data access patterns when choosing Redis data types
• Use Redis modules (JSON, Search) for complex operations

Persistence and Backup

• Enable Redis persistence (RDB snapshots + AOF logs)
• Regular backup of RDB files for disaster recovery
• Test backup restoration procedures
• Monitor Redis logs for persistence issues

High Availability

• Set up Redis Sentinel for automatic failover
• Use Redis Cluster for horizontal scaling
• Monitor Redis health and performance metrics
• Plan for network partitions and split-brain scenarios

---

Related

MongoDB State Store

Alternative NoSQL state storage with document features

Database Persistence Overview

Compare all available persistence backends