​
Session Management

Learn how to implement stateful applications with persistent sessions, enabling memory across conversations, user-specific contexts, and checkpoint-based recovery.

​
Overview

Session management in PraisonAI enables:

  • Persistent conversations across multiple interactions
  • User-specific memory and preferences
  • State checkpointing and recovery
  • Multi-agent state sharing
  • Remote session connectivity

​
Basic Session Usage

​
Creating a Session

from praisonaiagents import Agent, Session
import uuid

# Create a unique session ID
session_id = str(uuid.uuid4())

# Create an agent
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant with memory"
)

# Create a session
session = Session(session_id=session_id, agent=agent)

# Interact with the session
response1 = session.chat("My name is Alice")
response2 = session.chat("What's my name?")  # Will remember "Alice"

​
Persisting Session State

# Save session state
session.save("sessions/alice_session.pkl")

# Later, restore the session
restored_session = Session.load("sessions/alice_session.pkl")

# Continue conversation with context
response = restored_session.chat("What did we discuss earlier?")

​
Memory Integration

​
Short-term and Long-term Memory

from praisonaiagents import Agent, Session, Memory

# Configure memory
memory = Memory(
    provider="sqlite",  # or "chromadb", "mem0"
    config={
        "db_path": "memories.db",
        "embedding_model": "text-embedding-ada-002"
    }
)

# Create agent with memory
agent = Agent(
    name="MemoryAgent",
    instructions="Remember user preferences and past conversations",
    memory=memory
)

# Session with persistent memory
session = Session(
    session_id="user_123",
    agent=agent,
    persist_memory=True
)

# Memory is automatically saved and retrieved
session.chat("I prefer Python over JavaScript")
session.chat("My favourite colour is blue")

# Later sessions will remember
new_session = Session(session_id="user_123", agent=agent)
response = new_session.chat("What's my favourite programming language?")
# Response: "Based on our previous conversation, your favourite programming language is Python"

​
Memory Quality and Filtering

from praisonaiagents import Memory, QualityFilter

# Memory with quality scoring
memory = Memory(
    provider="chromadb",
    quality_threshold=0.7,  # Only store high-quality memories
    max_memories=1000      # Limit memory size
)

# Custom quality filter
def quality_scorer(memory_item):
    # Score based on relevance, recency, and importance
    relevance_score = memory_item.get("relevance", 0.5)
    recency_factor = calculate_recency(memory_item["timestamp"])
    importance = memory_item.get("importance", 0.5)
    
    return (relevance_score * 0.4 + 
            recency_factor * 0.3 + 
            importance * 0.3)

memory.set_quality_scorer(quality_scorer)

​
Advanced Session Patterns

​
Multi-Agent Sessions

from praisonaiagents import Agent, Session, SharedState

# Shared state for agents
shared_state = SharedState()

# Create specialised agents
research_agent = Agent(
    name="Researcher",
    instructions="Research topics thoroughly"
)

writer_agent = Agent(
    name="Writer",
    instructions="Write based on research"
)

# Create session with multiple agents
session = Session(
    session_id="project_x",
    agents={
        "research": research_agent,
        "writer": writer_agent
    },
    shared_state=shared_state
)

# Agents share state
session.run_agent("research", "Research quantum computing")
session.run_agent("writer", "Write an article based on the research")

​
Session Checkpointing

class CheckpointedSession(Session):
    def __init__(self, *args, checkpoint_interval=5, **kwargs):
        super().__init__(*args, **kwargs)
        self.checkpoint_interval = checkpoint_interval
        self.interaction_count = 0
        self.checkpoint_dir = f"checkpoints/{self.session_id}"
    
    def chat(self, message):
        response = super().chat(message)
        
        self.interaction_count += 1
        if self.interaction_count % self.checkpoint_interval == 0:
            self.create_checkpoint()
        
        return response
    
    def create_checkpoint(self):
        checkpoint_path = f"{self.checkpoint_dir}/checkpoint_{self.interaction_count}.pkl"
        self.save(checkpoint_path)
        print(f"Checkpoint created: {checkpoint_path}")
    
    def restore_checkpoint(self, checkpoint_number):
        checkpoint_path = f"{self.checkpoint_dir}/checkpoint_{checkpoint_number}.pkl"
        restored = Session.load(checkpoint_path)
        self.__dict__.update(restored.__dict__)
        print(f"Restored to checkpoint {checkpoint_number}")

​
Session Context Building

from datetime import datetime

class ContextualSession(Session):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.context_window = 10  # Keep last 10 interactions
        self.user_profile = {}
        self.session_metadata = {
            "created_at": datetime.now(),
            "interactions": 0
        }
    
    def build_context(self):
        # Build context from recent history
        recent_history = self.get_history()[-self.context_window:]
        
        context = {
            "user_profile": self.user_profile,
            "recent_topics": self.extract_topics(recent_history),
            "session_duration": (datetime.now() - self.session_metadata["created_at"]).seconds,
            "interaction_count": self.session_metadata["interactions"]
        }
        
        return context
    
    def chat(self, message):
        # Update metadata
        self.session_metadata["interactions"] += 1
        
        # Add context to agent
        context = self.build_context()
        self.agent.context = context
        
        return super().chat(message)
    
    def update_profile(self, key, value):
        self.user_profile[key] = value

​
Remote Sessions

​
Connecting to Remote Agents

from praisonaiagents import Session

# Connect to remote agent
remote_session = Session(
    agent_url="http://192.168.1.100:8000/agent",
    session_id="user_456"
)

# Interact normally
response = remote_session.chat("Hello from remote client")

# Session state is maintained on the server

​
Distributed Session State

import redis
import json

class DistributedSession(Session):
    def __init__(self, *args, redis_host="localhost", redis_port=6379, **kwargs):
        super().__init__(*args, **kwargs)
        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
        self.state_key = f"session:{self.session_id}"
    
    def save_state(self):
        # Save state to Redis
        state = {
            "history": self.get_history(),
            "user_data": getattr(self, "user_data", {}),
            "metadata": getattr(self, "metadata", {})
        }
        
        self.redis_client.set(
            self.state_key,
            json.dumps(state),
            ex=3600  # Expire after 1 hour
        )
    
    def load_state(self):
        # Load state from Redis
        state_data = self.redis_client.get(self.state_key)
        if state_data:
            state = json.loads(state_data)
            self.set_history(state["history"])
            self.user_data = state["user_data"]
            self.metadata = state["metadata"]
            return True
        return False
    
    def chat(self, message):
        # Load latest state
        self.load_state()
        
        # Process message
        response = super().chat(message)
        
        # Save updated state
        self.save_state()
        
        return response

​
State Operations

​
State Versioning

class VersionedSession(Session):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.state_versions = []
        self.current_version = 0
    
    def save_version(self, description=""):
        version = {
            "version": self.current_version,
            "timestamp": datetime.now(),
            "description": description,
            "state": self.get_state()
        }
        
        self.state_versions.append(version)
        self.current_version += 1
        
        return self.current_version - 1
    
    def restore_version(self, version_number):
        for version in self.state_versions:
            if version["version"] == version_number:
                self.set_state(version["state"])
                return True
        return False
    
    def list_versions(self):
        return [
            {
                "version": v["version"],
                "timestamp": v["timestamp"],
                "description": v["description"]
            }
            for v in self.state_versions
        ]

​
State Merging

class MergableSession(Session):
    def merge_sessions(self, other_session):
        # Merge conversation histories
        merged_history = self.get_history() + other_session.get_history()
        merged_history.sort(key=lambda x: x.get("timestamp", 0))
        
        # Merge user data
        merged_user_data = {**self.user_data, **other_session.user_data}
        
        # Create new session with merged state
        merged_session = Session(
            session_id=f"{self.session_id}_merged",
            agent=self.agent
        )
        
        merged_session.set_history(merged_history)
        merged_session.user_data = merged_user_data
        
        return merged_session

​
Workflow Integration

​
Stateful Workflows

from praisonaiagents import Agent, Task, Workflow, Session

class StatefulWorkflow:
    def __init__(self, session_id):
        self.session = Session(session_id=session_id)
        self.agents = self.create_agents()
        self.workflow = None
    
    def create_agents(self):
        return {
            "planner": Agent(name="Planner", instructions="Plan tasks"),
            "executor": Agent(name="Executor", instructions="Execute tasks"),
            "reviewer": Agent(name="Reviewer", instructions="Review results")
        }
    
    def create_workflow(self, objective):
        # Load session state
        session_context = self.session.get_context()
        
        tasks = [
            Task(
                description=f"Plan steps for: {objective}",
                agent=self.agents["planner"],
                context=session_context
            ),
            Task(
                description="Execute the planned steps",
                agent=self.agents["executor"],
                depends_on=["task_1"]
            ),
            Task(
                description="Review and summarise results",
                agent=self.agents["reviewer"],
                depends_on=["task_2"]
            )
        ]
        
        self.workflow = Workflow(tasks=tasks)
    
    def run(self, objective):
        self.create_workflow(objective)
        results = self.workflow.run()
        
        # Save results to session
        self.session.add_result(objective, results)
        self.session.save(f"sessions/{self.session.session_id}_workflow.pkl")
        
        return results

​
Best Practices

​
1. Session Lifecycle Management

class ManagedSession(Session):
    def __enter__(self):
        # Load existing state if available
        try:
            self.load(f"sessions/{self.session_id}.pkl")
        except FileNotFoundError:
            pass
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Always save state on exit
        self.save(f"sessions/{self.session_id}.pkl")

# Usage
with ManagedSession(session_id="user_789") as session:
    response = session.chat("Process this request")
    # State automatically saved on exit

​
2. Session Cleanup

import os
from datetime import datetime, timedelta

def cleanup_old_sessions(session_dir="sessions", days_to_keep=30):
    cutoff_date = datetime.now() - timedelta(days=days_to_keep)
    
    for filename in os.listdir(session_dir):
        filepath = os.path.join(session_dir, filename)
        
        # Check file age
        file_modified = datetime.fromtimestamp(os.path.getmtime(filepath))
        
        if file_modified < cutoff_date:
            os.remove(filepath)
            print(f"Removed old session: {filename}")

​
3. Security Considerations

import hmac
import hashlib

class SecureSession(Session):
    def __init__(self, *args, secret_key=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.secret_key = secret_key or os.urandom(32)
    
    def generate_session_token(self):
        # Generate secure session token
        message = f"{self.session_id}:{datetime.now().isoformat()}"
        signature = hmac.new(
            self.secret_key,
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        
        return f"{message}:{signature}"
    
    def verify_session_token(self, token):
        try:
            message, signature = token.rsplit(":", 1)
            expected_signature = hmac.new(
                self.secret_key,
                message.encode(),
                hashlib.sha256
            ).hexdigest()
            
            return hmac.compare_digest(signature, expected_signature)
        except:
            return False

​
Common Patterns

​
User Preference Management

session.update_preference("language", "en-GB")
session.update_preference("timezone", "Europe/London")
session.update_preference("output_format", "detailed")

​
Conversation Branching

# Save current state
branch_point = session.create_checkpoint()

# Explore one path
response1 = session.chat("What if we use approach A?")

# Return to branch point
session.restore_checkpoint(branch_point)

# Explore alternative path
response2 = session.chat("What if we use approach B?")

​
Session Analytics

def analyse_session(session):
    return {
        "total_interactions": len(session.get_history()),
        "session_duration": session.get_duration(),
        "topics_discussed": session.extract_topics(),
        "user_satisfaction": session.calculate_satisfaction_score(),
        "agent_performance": session.get_performance_metrics()
    }

​
Troubleshooting

​
Memory Issues

  • Use memory limits and cleanup strategies
  • Implement sliding window for conversation history
  • Store large data externally with references

​
State Corruption

  • Implement state validation on load
  • Keep backup versions
  • Use atomic save operations

​
Performance

  • Lazy load session data
  • Cache frequently accessed state
  • Use appropriate storage backend for scale