Managing State Across Interactions › Implementing the Three-Tier State Model

import os
import anthropic
from typing import Dict, List, Any, Optional
import datetime
import json

class StatefulAgent:
    """Agent with proper state lifecycle management"""
    
    def __init__(self, user_id: str, model="claude-sonnet-4-5"):
        # Using Claude Sonnet 4.5 for superior agent reasoning
        self.client = anthropic.Anthropic()
        self.model = model
        self.user_id = user_id
        
        # Three tiers of state
        self.ephemeral_state = {}
        self.session_state = {
            "conversation_history": [],
            "session_start": datetime.datetime.now(),
            "current_goal": None
        }
        self.persistent_state = self._load_persistent_state()
    
    def _load_persistent_state(self) -> Dict[str, Any]:
        """Load long-term state from storage"""
        # In a real system, this would load from a database
        # For now, we'll use a simple file-based approach
        try:
            with open(f"user_{self.user_id}_state.json", "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return {
                "preferences": {},
                "facts": {},
                "history": []
            }
    
    def _save_persistent_state(self):
        """Save long-term state to storage"""
        with open(f"user_{self.user_id}_state.json", "w") as f:
            json.dump(self.persistent_state, f, indent=2)
    
    def start_task(self, task_description: str):
        """Initialize ephemeral state for a new task"""
        self.ephemeral_state = {
            "task": task_description,
            "steps": [],
            "results": {},
            "tools_used": []
        }
    
    def complete_task(self):
        """Clear ephemeral state after task completion"""
        # Optionally save a summary to persistent state
        if self.ephemeral_state:
            summary = {
                "task": self.ephemeral_state.get("task"),
                "completed_at": datetime.datetime.now().isoformat()
            }
            self.persistent_state["history"].append(summary)
            self._save_persistent_state()
        
        self.ephemeral_state = {}
    
    def end_session(self):
        """Clear session state, preserve persistent state"""
        # Save session summary to persistent state
        if self.session_state["conversation_history"]:
            session_summary = {
                "date": datetime.datetime.now().isoformat(),
                "message_count": len(self.session_state["conversation_history"]),
                "topics": self.session_state.get("current_goal")
            }
            self.persistent_state["history"].append(session_summary)
            self._save_persistent_state()
        
        # Clear session
        self.session_state = {
            "conversation_history": [],
            "session_start": None,
            "current_goal": None
        }
    
    def run(self, user_input: str) -> str:
        """Process user input with proper state management"""
        # Add to session state
        self.session_state["conversation_history"].append({
            "role": "user",
            "content": user_input,
            "timestamp": datetime.datetime.now().isoformat()
        })
        
        # Build context from all three state tiers
        context = self._build_context()
        
        # Get response from model
        response = self.client.messages.create(
            model=self.model,
            messages=context,
            max_tokens=1024
        )
        
        assistant_message = response.content[0].text
        
        # Update session state
        self.session_state["conversation_history"].append({
            "role": "assistant",
            "content": assistant_message,
            "timestamp": datetime.datetime.now().isoformat()
        })
        
        # Check if we should update persistent state
        self._check_for_persistent_updates(user_input, assistant_message)
        
        return assistant_message
    
    def _build_context(self) -> List[Dict[str, str]]:
        """Build conversation context from all state tiers"""
        context = []
        
        # Add system message with persistent facts
        if self.persistent_state["preferences"] or self.persistent_state["facts"]:
            system_info = "What I know about you:\n"
            for key, value in self.persistent_state["preferences"].items():
                system_info += f"- {key}: {value}\n"
            for fact in self.persistent_state["facts"].values():
                system_info += f"- {fact}\n"
            
            context.append({
                "role": "user",
                "content": f"[System Context]\n{system_info}"
            })
            context.append({
                "role": "assistant",
                "content": "I understand. I'll keep this information in mind."
            })
        
        # Add recent conversation history (session state)
        # Limit to last 10 messages to avoid token bloat
        recent_history = self.session_state["conversation_history"][-10:]
        context.extend([
            {"role": msg["role"], "content": msg["content"]}
            for msg in recent_history
        ])
        
        return context
    
    def _check_for_persistent_updates(self, user_input: str, assistant_message: str):
        """Check if we should update persistent state"""
        # Simple heuristic: look for preference-setting language
        if "prefer" in user_input.lower() or "remember" in user_input.lower():
            # In a real system, you'd use the LLM to extract the preference
            # For now, just store the raw statement
            self.persistent_state["facts"][datetime.datetime.now().isoformat()] = user_input
            self._save_persistent_state()

## Usage example
agent = StatefulAgent(user_id="user123")

## Morning interaction
print("=== Morning Session ===")
print(agent.run("Remember, I prefer morning flights."))
print(agent.run("What's the weather in Paris?"))
agent.end_session()

## Afternoon interaction (new session, but same persistent state)
print("\n=== Afternoon Session ===")
agent2 = StatefulAgent(user_id="user123")  # Same user, new session
print(agent2.run("Book me a flight to London next week."))
## The agent should remember the morning flight preference