🔍 Code Extractor

SessionManager is a class that manages conversation sessions and tracking using SQLite database, storing conversations and their exchanges with metadata.

Purpose

This class provides comprehensive session management for conversational AI applications. It handles creating and tracking conversations, storing individual exchanges (user inputs and AI responses), maintaining conversation context, and generating session-aware filenames. It uses SQLite for persistent storage and supports features like conversation history retrieval, context summarization, and active conversation listing. Ideal for applications that need to maintain stateful conversations across multiple interactions.

Source Code

class SessionManager:
    """Manages conversation sessions and tracking"""
    
    def __init__(self, db_path: str = "eink_sessions.db"):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(exist_ok=True)
        self.setup_database()
        print(f"📊 Session manager initialized: {self.db_path}")
    
    def setup_database(self):
        """Initialize SQLite database for session tracking"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                conversation_id TEXT PRIMARY KEY,
                user_id TEXT,
                created_at TEXT,
                last_activity TEXT,
                total_exchanges INTEGER DEFAULT 0,
                status TEXT DEFAULT 'active',
                context_summary TEXT DEFAULT '',
                metadata TEXT DEFAULT '{}'
            )
            """)
            
            conn.execute("""
            CREATE TABLE IF NOT EXISTS exchanges (
                exchange_id TEXT PRIMARY KEY,
                conversation_id TEXT,
                sequence_number INTEGER,
                input_file TEXT,
                input_type TEXT,
                response_text TEXT,
                processing_time REAL,
                tokens_used INTEGER,
                created_at TEXT,
                metadata TEXT DEFAULT '{}',
                FOREIGN KEY (conversation_id) REFERENCES conversations(conversation_id)
            )
            """)
            
            conn.commit()
    
    def create_conversation(self, user_id: Optional[str] = None) -> str:
        """Create a new conversation with unique ID"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        short_uuid = uuid.uuid4().hex[:8]
        conversation_id = f"conv_{timestamp}_{short_uuid}"
        
        now = datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            INSERT INTO conversations 
            (conversation_id, user_id, created_at, last_activity, status)
            VALUES (?, ?, ?, ?, 'active')
            """, (conversation_id, user_id, now, now))
            conn.commit()
        
        print(f"🆔 New conversation started: {conversation_id}")
        return conversation_id
    
    def add_exchange(self, conversation_id: str, input_file: str, input_type: str, 
                    response_text: str, processing_time: float, tokens_used: int,
                    metadata: Dict[str, Any] = None) -> str:
        """Add an exchange to a conversation"""
        
        if metadata is None:
            metadata = {}
        
        exchange_id = f"ex_{uuid.uuid4().hex[:8]}"
        now = datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            # Get current exchange count
            cursor = conn.execute(
                "SELECT total_exchanges FROM conversations WHERE conversation_id = ?",
                (conversation_id,)
            )
            result = cursor.fetchone()
            if not result:
                raise ValueError(f"Conversation {conversation_id} not found")
            
            sequence_number = result[0] + 1
            
            # Insert exchange
            conn.execute("""
            INSERT INTO exchanges 
            (exchange_id, conversation_id, sequence_number, input_file, input_type,
             response_text, processing_time, tokens_used, created_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (exchange_id, conversation_id, sequence_number, input_file, input_type,
                  response_text, processing_time, tokens_used, now, json.dumps(metadata)))
            
            # Update conversation
            conn.execute("""
            UPDATE conversations 
            SET total_exchanges = ?, last_activity = ?
            WHERE conversation_id = ?
            """, (sequence_number, now, conversation_id))
            
            conn.commit()
        
        print(f"📝 Exchange added: {conversation_id} #{sequence_number}")
        return exchange_id
    
    def get_conversation(self, conversation_id: str) -> Optional[ConversationState]:
        """Retrieve conversation state"""
        with sqlite3.connect(self.db_path) as conn:
            # Get conversation info
            cursor = conn.execute("""
            SELECT conversation_id, user_id, created_at, last_activity, 
                   total_exchanges, status, context_summary, metadata
            FROM conversations WHERE conversation_id = ?
            """, (conversation_id,))
            
            conv_row = cursor.fetchone()
            if not conv_row:
                return None
            
            # Get exchanges
            cursor = conn.execute("""
            SELECT exchange_id, sequence_number, input_file, input_type,
                   response_text, processing_time, tokens_used, created_at, metadata
            FROM exchanges WHERE conversation_id = ? ORDER BY sequence_number
            """, (conversation_id,))
            
            exchanges = []
            for ex_row in cursor.fetchall():
                exchange = Exchange(
                    exchange_id=ex_row[0],
                    sequence_number=ex_row[1],
                    input_file=ex_row[2],
                    input_type=ex_row[3],
                    response_text=ex_row[4],
                    processing_time=ex_row[5],
                    tokens_used=ex_row[6],
                    created_at=ex_row[7],
                    metadata=json.loads(ex_row[8] or '{}')
                )
                exchanges.append(exchange)
            
            return ConversationState(
                conversation_id=conv_row[0],
                user_id=conv_row[1],
                created_at=conv_row[2],
                last_activity=conv_row[3],
                total_exchanges=conv_row[4],
                status=conv_row[5],
                exchanges=exchanges,
                context_summary=conv_row[6] or '',
                metadata=json.loads(conv_row[7] or '{}')
            )
    
    def update_context_summary(self, conversation_id: str, summary: str):
        """Update conversation context summary"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            UPDATE conversations SET context_summary = ? WHERE conversation_id = ?
            """, (summary, conversation_id))
            conn.commit()
    
    def generate_session_filename(self, conversation_id: str, exchange_number: int, 
                                 original_filename: str, is_error: bool = False) -> str:
        """Generate filename with session context"""
        base_name = Path(original_filename).stem
        
        if is_error:
            prefix = "ERROR"
        else:
            prefix = "RESPONSE"
            
        return f"{prefix}_{conversation_id}_ex{exchange_number:03d}_{base_name}.pdf"
    
    def get_conversation_context(self, conversation_id: str, max_exchanges: int = 3) -> str:
        """Get recent conversation context for LLM"""
        conversation = self.get_conversation(conversation_id)
        if not conversation or not conversation.exchanges:
            return ""
        
        # Get last N exchanges for context
        recent_exchanges = conversation.exchanges[-max_exchanges:]
        
        context_parts = []
        if conversation.context_summary:
            context_parts.append(f"Context: {conversation.context_summary}")
        
        context_parts.append(f"Recent exchanges from conversation {conversation_id}:")
        
        for ex in recent_exchanges:
            context_parts.append(f"Exchange #{ex.sequence_number}: {ex.input_type} -> {ex.response_text[:100]}...")
        
        return "\n".join(context_parts)
    
    def list_active_conversations(self) -> List[Dict[str, Any]]:
        """List all active conversations"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
            SELECT conversation_id, user_id, created_at, last_activity, total_exchanges
            FROM conversations WHERE status = 'active'
            ORDER BY last_activity DESC
            """)
            
            conversations = []
            for row in cursor.fetchall():
                conversations.append({
                    'conversation_id': row[0],
                    'user_id': row[1],
                    'created_at': row[2],
                    'last_activity': row[3],
                    'total_exchanges': row[4]
                })
            
            return conversations

Parameters

Name	Type	Default	Kind
bases	-	-

Parameter Details

db_path: Path to the SQLite database file for storing session data. Defaults to 'eink_sessions.db'. The parent directory will be created if it doesn't exist. This parameter allows customization of where session data is persisted.

Return Value

Instantiation returns a SessionManager object that manages conversation sessions. Key method returns: create_conversation() returns a unique conversation ID string; add_exchange() returns an exchange ID string; get_conversation() returns a ConversationState object or None; get_conversation_context() returns a formatted string with recent conversation history; list_active_conversations() returns a list of dictionaries containing conversation metadata.

Class Interface

Methods

Attributes

Name	Type	Description	Scope
db_path	Path	Path object pointing to the SQLite database file location	instance

Dependencies

• uuid
• json
• sqlite3
• datetime
• pathlib
• typing
• dataclasses

Required Imports

import uuid
import json
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict

Usage Example

# Initialize the session manager
session_mgr = SessionManager(db_path="my_sessions.db")

# Create a new conversation
conv_id = session_mgr.create_conversation(user_id="user123")

# Add an exchange to the conversation
exchange_id = session_mgr.add_exchange(
    conversation_id=conv_id,
    input_file="document.pdf",
    input_type="pdf",
    response_text="This document discusses...",
    processing_time=2.5,
    tokens_used=150,
    metadata={"model": "gpt-4"}
)

# Retrieve conversation state
conversation = session_mgr.get_conversation(conv_id)
if conversation:
    print(f"Total exchanges: {conversation.total_exchanges}")

# Get recent context for LLM
context = session_mgr.get_conversation_context(conv_id, max_exchanges=3)

# Generate session-aware filename
filename = session_mgr.generate_session_filename(
    conversation_id=conv_id,
    exchange_number=1,
    original_filename="input.pdf"
)

# List all active conversations
active_convs = session_mgr.list_active_conversations()
for conv in active_convs:
    print(f"{conv['conversation_id']}: {conv['total_exchanges']} exchanges")

Best Practices

• Always call create_conversation() before adding exchanges to establish a valid conversation context
• The database is automatically initialized on instantiation via setup_database()
• Use add_exchange() after each user-AI interaction to maintain complete conversation history
• Call update_context_summary() periodically to maintain a high-level summary of long conversations
• Use get_conversation_context() to provide recent history to LLMs for context-aware responses
• The class uses context managers (with statements) for database connections, ensuring proper resource cleanup
• Exchange sequence numbers are automatically managed and incremented
• Conversation IDs are unique and include timestamps for easy identification
• The metadata parameter in add_exchange() allows storing arbitrary JSON-serializable data
• Database operations are atomic within each method using transactions
• Consider the max_exchanges parameter in get_conversation_context() to balance context richness vs token usage

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function test_session_manager 71.9% similar

  A comprehensive test function that validates the SessionManager class functionality including conversation creation, exchange tracking, filename generation, and context retrieval.

  From: /tf/active/vicechatdev/e-ink-llm/test_improvements.py

• class ChatSession 69.4% similar

  A class that manages a chat session associated with a specific document section, tracking messages, context documents, references, and timestamps.

  From: /tf/active/vicechatdev/vice_ai/complex_app.py

• class DatabaseManager 67.4% similar

  SQLite database manager for SmartStat application that handles persistence of statistical analysis sessions, steps, and results.

  From: /tf/active/vicechatdev/smartstat/models.py

• class ChatSession_v1 66.2% similar

  A dataclass representing a chat session associated with a specific text section in a document, managing conversation messages, context, and references.

  From: /tf/active/vicechatdev/vice_ai/models.py

• class SessionManager 66.1% similar

  A session management class for Panel applications that provides user authentication and arbitrary key-value storage using Panel's built-in state management system.

  From: /tf/active/vicechatdev/CDocs/auth/session_manager.py

class SessionManager:
    """Manages conversation sessions and tracking"""
    
    def __init__(self, db_path: str = "eink_sessions.db"):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(exist_ok=True)
        self.setup_database()
        print(f"📊 Session manager initialized: {self.db_path}")
    
    def setup_database(self):
        """Initialize SQLite database for session tracking"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            CREATE TABLE IF NOT EXISTS conversations (
                conversation_id TEXT PRIMARY KEY,
                user_id TEXT,
                created_at TEXT,
                last_activity TEXT,
                total_exchanges INTEGER DEFAULT 0,
                status TEXT DEFAULT 'active',
                context_summary TEXT DEFAULT '',
                metadata TEXT DEFAULT '{}'
            )
            """)
            
            conn.execute("""
            CREATE TABLE IF NOT EXISTS exchanges (
                exchange_id TEXT PRIMARY KEY,
                conversation_id TEXT,
                sequence_number INTEGER,
                input_file TEXT,
                input_type TEXT,
                response_text TEXT,
                processing_time REAL,
                tokens_used INTEGER,
                created_at TEXT,
                metadata TEXT DEFAULT '{}',
                FOREIGN KEY (conversation_id) REFERENCES conversations(conversation_id)
            )
            """)
            
            conn.commit()
    
    def create_conversation(self, user_id: Optional[str] = None) -> str:
        """Create a new conversation with unique ID"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        short_uuid = uuid.uuid4().hex[:8]
        conversation_id = f"conv_{timestamp}_{short_uuid}"
        
        now = datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            INSERT INTO conversations 
            (conversation_id, user_id, created_at, last_activity, status)
            VALUES (?, ?, ?, ?, 'active')
            """, (conversation_id, user_id, now, now))
            conn.commit()
        
        print(f"🆔 New conversation started: {conversation_id}")
        return conversation_id
    
    def add_exchange(self, conversation_id: str, input_file: str, input_type: str, 
                    response_text: str, processing_time: float, tokens_used: int,
                    metadata: Dict[str, Any] = None) -> str:
        """Add an exchange to a conversation"""
        
        if metadata is None:
            metadata = {}
        
        exchange_id = f"ex_{uuid.uuid4().hex[:8]}"
        now = datetime.now().isoformat()
        
        with sqlite3.connect(self.db_path) as conn:
            # Get current exchange count
            cursor = conn.execute(
                "SELECT total_exchanges FROM conversations WHERE conversation_id = ?",
                (conversation_id,)
            )
            result = cursor.fetchone()
            if not result:
                raise ValueError(f"Conversation {conversation_id} not found")
            
            sequence_number = result[0] + 1
            
            # Insert exchange
            conn.execute("""
            INSERT INTO exchanges 
            (exchange_id, conversation_id, sequence_number, input_file, input_type,
             response_text, processing_time, tokens_used, created_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (exchange_id, conversation_id, sequence_number, input_file, input_type,
                  response_text, processing_time, tokens_used, now, json.dumps(metadata)))
            
            # Update conversation
            conn.execute("""
            UPDATE conversations 
            SET total_exchanges = ?, last_activity = ?
            WHERE conversation_id = ?
            """, (sequence_number, now, conversation_id))
            
            conn.commit()
        
        print(f"📝 Exchange added: {conversation_id} #{sequence_number}")
        return exchange_id
    
    def get_conversation(self, conversation_id: str) -> Optional[ConversationState]:
        """Retrieve conversation state"""
        with sqlite3.connect(self.db_path) as conn:
            # Get conversation info
            cursor = conn.execute("""
            SELECT conversation_id, user_id, created_at, last_activity, 
                   total_exchanges, status, context_summary, metadata
            FROM conversations WHERE conversation_id = ?
            """, (conversation_id,))
            
            conv_row = cursor.fetchone()
            if not conv_row:
                return None
            
            # Get exchanges
            cursor = conn.execute("""
            SELECT exchange_id, sequence_number, input_file, input_type,
                   response_text, processing_time, tokens_used, created_at, metadata
            FROM exchanges WHERE conversation_id = ? ORDER BY sequence_number
            """, (conversation_id,))
            
            exchanges = []
            for ex_row in cursor.fetchall():
                exchange = Exchange(
                    exchange_id=ex_row[0],
                    sequence_number=ex_row[1],
                    input_file=ex_row[2],
                    input_type=ex_row[3],
                    response_text=ex_row[4],
                    processing_time=ex_row[5],
                    tokens_used=ex_row[6],
                    created_at=ex_row[7],
                    metadata=json.loads(ex_row[8] or '{}')
                )
                exchanges.append(exchange)
            
            return ConversationState(
                conversation_id=conv_row[0],
                user_id=conv_row[1],
                created_at=conv_row[2],
                last_activity=conv_row[3],
                total_exchanges=conv_row[4],
                status=conv_row[5],
                exchanges=exchanges,
                context_summary=conv_row[6] or '',
                metadata=json.loads(conv_row[7] or '{}')
            )
    
    def update_context_summary(self, conversation_id: str, summary: str):
        """Update conversation context summary"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
            UPDATE conversations SET context_summary = ? WHERE conversation_id = ?
            """, (summary, conversation_id))
            conn.commit()
    
    def generate_session_filename(self, conversation_id: str, exchange_number: int, 
                                 original_filename: str, is_error: bool = False) -> str:
        """Generate filename with session context"""
        base_name = Path(original_filename).stem
        
        if is_error:
            prefix = "ERROR"
        else:
            prefix = "RESPONSE"
            
        return f"{prefix}_{conversation_id}_ex{exchange_number:03d}_{base_name}.pdf"
    
    def get_conversation_context(self, conversation_id: str, max_exchanges: int = 3) -> str:
        """Get recent conversation context for LLM"""
        conversation = self.get_conversation(conversation_id)
        if not conversation or not conversation.exchanges:
            return ""
        
        # Get last N exchanges for context
        recent_exchanges = conversation.exchanges[-max_exchanges:]
        
        context_parts = []
        if conversation.context_summary:
            context_parts.append(f"Context: {conversation.context_summary}")
        
        context_parts.append(f"Recent exchanges from conversation {conversation_id}:")
        
        for ex in recent_exchanges:
            context_parts.append(f"Exchange #{ex.sequence_number}: {ex.input_type} -> {ex.response_text[:100]}...")
        
        return "\n".join(context_parts)
    
    def list_active_conversations(self) -> List[Dict[str, Any]]:
        """List all active conversations"""
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute("""
            SELECT conversation_id, user_id, created_at, last_activity, total_exchanges
            FROM conversations WHERE status = 'active'
            ORDER BY last_activity DESC
            """)
            
            conversations = []
            for row in cursor.fetchall():
                conversations.append({
                    'conversation_id': row[0],
                    'user_id': row[1],
                    'created_at': row[2],
                    'last_activity': row[3],
                    'total_exchanges': row[4]
                })
            
            return conversations