Chatbot Design Patterns

from openai import OpenAI
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
import json


class ConversationState(Enum):
    """States in the conversation flow."""
    GREETING = "greeting"
    GATHERING_INFO = "gathering_info"
    CONFIRMING = "confirming"
    PROCESSING = "processing"
    COMPLETED = "completed"
    ERROR = "error"


@dataclass
class ConversationContext:
    """Holds conversation state and collected data."""
    state: ConversationState = ConversationState.GREETING
    collected_data: dict = field(default_factory=dict)
    history: list = field(default_factory=list)
    retry_count: int = 0
    metadata: dict = field(default_factory=dict)


class StatefulChatbot:
    """Chatbot with explicit state management."""
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.context = ConversationContext()
    
    def _get_system_prompt(self) -> str:
        """Get system prompt based on current state."""
        prompts = {
            ConversationState.GREETING: """You are a helpful assistant. 
                Greet the user warmly and ask how you can help them today.
                Keep it brief and friendly.""",
            
            ConversationState.GATHERING_INFO: """You are gathering information.
                Ask clarifying questions one at a time.
                Be conversational but focused.
                Acknowledge what the user tells you.""",
            
            ConversationState.CONFIRMING: """You are confirming details.
                Summarize what you've collected and ask for confirmation.
                Be clear and organized in your summary.""",
            
            ConversationState.PROCESSING: """You are processing the request.
                Acknowledge the request and explain next steps.
                Be reassuring and informative.""",
            
            ConversationState.COMPLETED: """The task is complete.
                Thank the user and offer further assistance.
                Be warm and professional.""",
        }
        
        return prompts.get(
            self.context.state,
            "You are a helpful assistant. Respond appropriately."
        )
    
    def _build_messages(self, user_input: str) -> list[dict]:
        """Build message list for API call."""
        messages = [{"role": "system", "content": self._get_system_prompt()}]
        
        # Add conversation history
        for msg in self.context.history[-10:]:  # Last 10 messages
            messages.append(msg)
        
        # Add current user input
        messages.append({"role": "user", "content": user_input})
        
        return messages
    
    def _determine_transition(self, user_input: str, response: str) -> ConversationState:
        """Determine next state based on conversation.
        
        Pitfall: Using an LLM for state transitions adds latency and cost
        to every turn. In production, prefer rule-based transitions for
        predictable flows (e.g., "did the user say yes?") and reserve
        LLM-based classification for ambiguous cases only.
        """
        analysis_prompt = f"""Analyze this conversation turn and determine the appropriate next state.

Current state: {self.context.state.value}
Collected data: {json.dumps(self.context.collected_data)}
User said: {user_input}
Assistant said: {response}

Available states:
- greeting: Initial greeting
- gathering_info: Collecting required information
- confirming: Verifying collected information
- processing: Executing the request
- completed: Task finished
- error: Something went wrong

Return JSON: {{"next_state": "state_name", "reason": "brief reason"}}"""
        
        result = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": analysis_prompt}],
            response_format={"type": "json_object"}
        )
        
        data = json.loads(result.choices[0].message.content)
        return ConversationState(data.get("next_state", "gathering_info"))
    
    def process_message(self, user_input: str) -> str:
        """Process user message and return response."""
        messages = self._build_messages(user_input)
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages
        )
        
        assistant_message = response.choices[0].message.content
        
        # Update history
        self.context.history.append({"role": "user", "content": user_input})
        self.context.history.append({"role": "assistant", "content": assistant_message})
        
        # Determine and apply state transition
        new_state = self._determine_transition(user_input, assistant_message)
        self.context.state = new_state
        
        return assistant_message
    
    def reset(self):
        """Reset conversation to initial state."""
        self.context = ConversationContext()


# Usage
chatbot = StatefulChatbot()

# Simulate conversation
print(chatbot.process_message("Hello!"))
print(f"State: {chatbot.context.state}")

print(chatbot.process_message("I need help booking a flight"))
print(f"State: {chatbot.context.state}")

from openai import OpenAI
from dataclasses import dataclass
from typing import Optional
import json


@dataclass
class Slot:
    """A piece of required information."""
    name: str
    description: str
    required: bool = True
    value: Optional[str] = None
    validation_prompt: str = None
    
    @property
    def is_filled(self) -> bool:
        return self.value is not None


class SlotFillingBot:
    """Bot that collects required information through conversation."""
    
    def __init__(self, slots: list[Slot], model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.slots = {s.name: s for s in slots}
        self.history: list[dict] = []
    
    def _get_unfilled_slots(self) -> list[Slot]:
        """Get list of unfilled required slots."""
        return [s for s in self.slots.values() if s.required and not s.is_filled]
    
    def _extract_slot_values(self, user_input: str) -> dict:
        """Extract slot values from user input.
        
        Tip: Users often provide multiple pieces of info at once ("fly from
        NYC to London next Friday for 2 people"). This method handles that
        by extracting all slots in a single LLM call rather than asking
        one question at a time -- which would feel robotic.
        """
        slot_descriptions = {
            name: slot.description 
            for name, slot in self.slots.items()
        }
        
        prompt = f"""Extract information from the user message.

Required slots:
{json.dumps(slot_descriptions, indent=2)}

User message: "{user_input}"

Return JSON with extracted values. Use null for slots not mentioned:
{{"slot_name": "extracted_value_or_null", ...}}"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    def _validate_slot(self, slot: Slot, value: str) -> tuple[bool, str]:
        """Validate a slot value."""
        if not slot.validation_prompt:
            return True, value
        
        prompt = f"""{slot.validation_prompt}

Value to validate: "{value}"

Return JSON: {{"valid": true/false, "corrected_value": "value or null", "reason": "if invalid"}}"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        
        if result.get("valid"):
            return True, result.get("corrected_value", value)
        return False, result.get("reason", "Invalid value")
    
    def _generate_question(self, slot: Slot) -> str:
        """Generate a natural question for a slot."""
        context = ""
        filled = [s for s in self.slots.values() if s.is_filled]
        if filled:
            context = "Already collected: " + ", ".join(
                f"{s.name}={s.value}" for s in filled
            )
        
        prompt = f"""Generate a natural, conversational question to ask for this information.
{context}

Slot to ask about:
- Name: {slot.name}
- Description: {slot.description}

Generate only the question, no preamble:"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content.strip()
    
    def _generate_confirmation(self) -> str:
        """Generate confirmation message with all slots."""
        slots_summary = "\n".join(
            f"- {slot.name}: {slot.value}"
            for slot in self.slots.values()
            if slot.is_filled
        )
        
        prompt = f"""Generate a confirmation message summarizing this information:

{slots_summary}

Ask the user to confirm or correct anything. Be concise and clear:"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content
    
    def process_message(self, user_input: str) -> dict:
        """Process user message and return response with status."""
        # Extract slot values from input
        extracted = self._extract_slot_values(user_input)
        
        # Validate and fill slots
        for name, value in extracted.items():
            if value and name in self.slots:
                slot = self.slots[name]
                is_valid, result = self._validate_slot(slot, value)
                if is_valid:
                    slot.value = result
        
        # Update history
        self.history.append({"role": "user", "content": user_input})
        
        # Check if all required slots are filled
        unfilled = self._get_unfilled_slots()
        
        if not unfilled:
            # All slots filled - confirm
            response = self._generate_confirmation()
            status = "confirming"
        else:
            # Ask for next unfilled slot
            response = self._generate_question(unfilled[0])
            status = "collecting"
        
        self.history.append({"role": "assistant", "content": response})
        
        return {
            "response": response,
            "status": status,
            "filled_slots": {
                name: slot.value 
                for name, slot in self.slots.items() 
                if slot.is_filled
            },
            "unfilled_slots": [s.name for s in unfilled]
        }


# Usage - Flight booking example
slots = [
    Slot(
        name="origin",
        description="Departure city or airport",
        validation_prompt="Validate this is a valid city or airport name"
    ),
    Slot(
        name="destination",
        description="Arrival city or airport",
        validation_prompt="Validate this is a valid city or airport name"
    ),
    Slot(
        name="date",
        description="Travel date",
        validation_prompt="Validate this is a valid date format (YYYY-MM-DD preferred)"
    ),
    Slot(
        name="passengers",
        description="Number of passengers",
        validation_prompt="Validate this is a positive integer"
    )
]

bot = SlotFillingBot(slots)

# Simulate conversation
result = bot.process_message("I want to fly from New York to London")
print(result["response"])
print(f"Filled: {result['filled_slots']}")
print(f"Still need: {result['unfilled_slots']}")

result = bot.process_message("Next Friday, just me")
print(result["response"])
print(f"Status: {result['status']}")

from openai import OpenAI
from dataclasses import dataclass, field
from typing import Optional
import json


@dataclass
class ConversationTurn:
    """A single turn in conversation."""
    role: str
    content: str
    metadata: dict = field(default_factory=dict)


@dataclass
class Topic:
    """A conversation topic or thread."""
    name: str
    summary: str
    turns: list[ConversationTurn] = field(default_factory=list)
    resolved: bool = False


class ContextManager:
    """Manages multi-turn conversation context."""
    
    def __init__(
        self,
        max_history: int = 20,
        model: str = "gpt-4o-mini"
    ):
        self.client = OpenAI()
        self.model = model
        self.max_history = max_history
        self.history: list[ConversationTurn] = []
        self.topics: list[Topic] = []
        self.current_topic: Optional[Topic] = None
        self.user_profile: dict = {}
    
    def _summarize_old_context(self, turns: list[ConversationTurn]) -> str:
        """Summarize older conversation turns."""
        if not turns:
            return ""
        
        history_text = "\n".join(
            f"{t.role}: {t.content}" for t in turns
        )
        
        prompt = f"""Summarize this conversation history concisely.
Preserve key facts, decisions, and context needed for future turns.

History:
{history_text}

Summary:"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content
    
    def _detect_topic_change(self, message: str) -> bool:
        """Detect if user is changing topics."""
        if not self.current_topic:
            return True
        
        prompt = f"""Is this message changing to a new topic?

Current topic: {self.current_topic.name}
Topic summary: {self.current_topic.summary}

New message: "{message}"

Return JSON: {{"topic_change": true/false, "reason": "brief reason"}}"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        return result.get("topic_change", False)
    
    def _identify_topic(self, message: str) -> str:
        """Identify the topic of a message."""
        prompt = f"""Identify the main topic of this message in 2-4 words.

Message: "{message}"

Topic:"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content.strip()
    
    def _extract_user_info(self, message: str) -> dict:
        """Extract user profile information from message."""
        prompt = f"""Extract any personal information the user mentions about themselves.

Message: "{message}"

Return JSON with any of: name, preferences, location, occupation, interests, or other relevant info.
Return empty object if nothing is mentioned."""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    def add_message(self, role: str, content: str) -> dict:
        """Add a message and manage context."""
        turn = ConversationTurn(role=role, content=content)
        self.history.append(turn)
        
        # Extract user info if user message
        if role == "user":
            user_info = self._extract_user_info(content)
            self.user_profile.update(user_info)
            
            # Handle topic management
            if self._detect_topic_change(content):
                # Archive current topic
                if self.current_topic:
                    self.current_topic.resolved = True
                    self.topics.append(self.current_topic)
                
                # Start new topic
                topic_name = self._identify_topic(content)
                self.current_topic = Topic(
                    name=topic_name,
                    summary=content[:100]
                )
            
            if self.current_topic:
                self.current_topic.turns.append(turn)
        
        # Compress history if needed
        context_summary = ""
        if len(self.history) > self.max_history:
            old_turns = self.history[:-self.max_history]
            context_summary = self._summarize_old_context(old_turns)
            self.history = self.history[-self.max_history:]
        
        return {
            "context_summary": context_summary,
            "current_topic": self.current_topic.name if self.current_topic else None,
            "user_profile": self.user_profile,
            "history_length": len(self.history)
        }
    
    def get_context_for_prompt(self) -> str:
        """Get formatted context for LLM prompt."""
        parts = []
        
        # User profile
        if self.user_profile:
            parts.append(f"User profile: {json.dumps(self.user_profile)}")
        
        # Current topic
        if self.current_topic:
            parts.append(f"Current topic: {self.current_topic.name}")
        
        # Recent topics
        recent_topics = [t.name for t in self.topics[-3:] if t.resolved]
        if recent_topics:
            parts.append(f"Previous topics discussed: {', '.join(recent_topics)}")
        
        return "\n".join(parts)
    
    def get_messages(self) -> list[dict]:
        """Get history as message list for API."""
        return [
            {"role": t.role, "content": t.content}
            for t in self.history
        ]


# Usage
context_mgr = ContextManager(max_history=10)

# User provides information over time
context_mgr.add_message("user", "Hi, I'm Alex and I work in software engineering")
context_mgr.add_message("assistant", "Hello Alex! Nice to meet you. How can I help today?")

context_mgr.add_message("user", "I'm looking for Python learning resources")
context_mgr.add_message("assistant", "I can help with that. What's your current Python level?")

print(f"User profile: {context_mgr.user_profile}")
print(f"Current topic: {context_mgr.current_topic.name}")
print(f"Context: {context_mgr.get_context_for_prompt()}")

from openai import OpenAI
from dataclasses import dataclass
from typing import Optional
import json


@dataclass
class Intent:
    """A user intent with handler."""
    name: str
    description: str
    examples: list[str]  # Few-shot examples improve classification accuracy
    handler: callable = None
    confidence_threshold: float = 0.8  # Below this, escalate or clarify


class IntentClassifier:
    """Classify user intents for routing."""
    
    def __init__(self, intents: list[Intent], model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.intents = {i.name: i for i in intents}
    
    def _build_classification_prompt(self) -> str:
        """Build prompt for intent classification."""
        intent_descriptions = []
        for name, intent in self.intents.items():
            examples = ", ".join(f'"{e}"' for e in intent.examples[:3])
            intent_descriptions.append(
                f"- {name}: {intent.description}\n  Examples: {examples}"
            )
        
        return f"""Classify the user's intent into one of these categories:

{chr(10).join(intent_descriptions)}

Return JSON:
{{
    "intent": "intent_name",
    "confidence": 0.0-1.0,
    "entities": {{"extracted_entity": "value"}},
    "reasoning": "brief explanation"
}}"""
    
    def classify(self, message: str) -> dict:
        """Classify a user message."""
        system_prompt = self._build_classification_prompt()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": message}
            ],
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        
        intent_name = result.get("intent")
        confidence = result.get("confidence", 0)
        
        intent = self.intents.get(intent_name)
        if intent and confidence >= intent.confidence_threshold:
            result["matched"] = True
            result["intent_object"] = intent
        else:
            result["matched"] = False
        
        return result
    
    def route(self, message: str, fallback: callable = None) -> any:
        """Route message to appropriate handler."""
        classification = self.classify(message)
        
        if classification["matched"]:
            intent = classification["intent_object"]
            if intent.handler:
                return intent.handler(
                    message,
                    classification.get("entities", {})
                )
        
        if fallback:
            return fallback(message)
        
        return None


# Define handlers
def handle_booking(message: str, entities: dict) -> str:
    return f"Starting booking process. Extracted: {entities}"

def handle_status(message: str, entities: dict) -> str:
    return f"Checking status for: {entities}"

def handle_cancel(message: str, entities: dict) -> str:
    return f"Processing cancellation: {entities}"

def handle_help(message: str, entities: dict) -> str:
    return "Here are the things I can help with..."


# Create classifier
intents = [
    Intent(
        name="booking",
        description="User wants to make a new booking or reservation",
        examples=[
            "I want to book a flight",
            "Can you help me make a reservation?",
            "I need to schedule an appointment"
        ],
        handler=handle_booking
    ),
    Intent(
        name="status",
        description="User wants to check status of existing booking",
        examples=[
            "What's the status of my order?",
            "Where is my booking?",
            "Track my reservation"
        ],
        handler=handle_status
    ),
    Intent(
        name="cancel",
        description="User wants to cancel something",
        examples=[
            "Cancel my booking",
            "I need to cancel my order",
            "Remove my reservation"
        ],
        handler=handle_cancel
    ),
    Intent(
        name="help",
        description="User needs help or information",
        examples=[
            "Help",
            "What can you do?",
            "I need assistance"
        ],
        handler=handle_help
    )
]

classifier = IntentClassifier(intents)

# Classify messages
result = classifier.classify("I'd like to book a hotel for next week")
print(f"Intent: {result['intent']} (confidence: {result['confidence']})")
print(f"Entities: {result.get('entities', {})}")

# Route to handler
response = classifier.route("Check my order status please", fallback=lambda m: "I didn't understand that")
print(response)

from openai import OpenAI
from dataclasses import dataclass, field
from typing import Callable, Optional
from enum import Enum


class FlowStep(Enum):
    """Standard flow steps."""
    START = "start"
    COLLECT = "collect"
    VALIDATE = "validate"
    CONFIRM = "confirm"
    EXECUTE = "execute"
    COMPLETE = "complete"
    ERROR = "error"


@dataclass
class FlowNode:
    """A node in the conversation flow."""
    name: str
    prompt_template: str
    next_steps: dict = field(default_factory=dict)  # condition -> next_node
    validator: Optional[Callable] = None
    processor: Optional[Callable] = None


class ConversationFlow:
    """Define and execute conversation flows."""
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.nodes: dict[str, FlowNode] = {}
        self.current_node: Optional[str] = None
        self.context: dict = {}
        self.history: list = []
    
    def add_node(self, node: FlowNode):
        """Add a node to the flow."""
        self.nodes[node.name] = node
    
    def start(self, start_node: str):
        """Start the flow at a specific node."""
        self.current_node = start_node
        return self._execute_node()
    
    def _execute_node(self) -> str:
        """Execute current node and return response."""
        node = self.nodes.get(self.current_node)
        if not node:
            return "Flow error: node not found"
        
        # Format prompt with context
        prompt = node.prompt_template.format(**self.context)
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": prompt}
            ] + self.history[-10:]
        )
        
        return response.choices[0].message.content
    
    def process_input(self, user_input: str) -> dict:
        """Process user input and advance flow."""
        self.history.append({"role": "user", "content": user_input})
        
        node = self.nodes.get(self.current_node)
        if not node:
            return {"error": "Invalid flow state"}
        
        # Run validator if present
        if node.validator:
            is_valid, result = node.validator(user_input, self.context)
            if not is_valid:
                response = f"Invalid input: {result}. Please try again."
                self.history.append({"role": "assistant", "content": response})
                return {
                    "response": response,
                    "node": self.current_node,
                    "valid": False
                }
        
        # Run processor if present
        if node.processor:
            self.context = node.processor(user_input, self.context)
        
        # Determine next node
        next_node = self._determine_next(node, user_input)
        if next_node:
            self.current_node = next_node
        
        # Execute new node
        response = self._execute_node()
        self.history.append({"role": "assistant", "content": response})
        
        return {
            "response": response,
            "node": self.current_node,
            "context": self.context,
            "valid": True
        }
    
    def _determine_next(self, node: FlowNode, user_input: str) -> Optional[str]:
        """Determine next node based on input and conditions."""
        # Check explicit conditions
        for condition, next_node in node.next_steps.items():
            if condition == "default":
                continue
            if condition.lower() in user_input.lower():
                return next_node
        
        # Use LLM for complex routing
        if len(node.next_steps) > 1:
            options = list(node.next_steps.keys())
            
            prompt = f"""Based on the user's response, which path should we take?

User said: "{user_input}"
Options: {options}

Return just the option name:"""
            
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}]
            )
            
            choice = response.choices[0].message.content.strip().lower()
            if choice in node.next_steps:
                return node.next_steps[choice]
        
        return node.next_steps.get("default")


# Example: Support ticket flow
def validate_email(user_input: str, context: dict) -> tuple[bool, str]:
    """Validate email format."""
    if "@" in user_input and "." in user_input:
        return True, user_input
    return False, "Please provide a valid email address"

def collect_email(user_input: str, context: dict) -> dict:
    """Store collected email."""
    context["email"] = user_input
    return context


# Build flow
flow = ConversationFlow()

flow.add_node(FlowNode(
    name="welcome",
    prompt_template="Welcome the user and ask for their email address for the support ticket.",
    next_steps={"default": "collect_email"}
))

flow.add_node(FlowNode(
    name="collect_email",
    prompt_template="Ask for the user's email address.",
    validator=validate_email,
    processor=collect_email,
    next_steps={"default": "collect_issue"}
))

flow.add_node(FlowNode(
    name="collect_issue",
    prompt_template="Email collected: {email}. Now ask them to describe their issue.",
    next_steps={
        "billing": "billing_flow",
        "technical": "technical_flow",
        "default": "general_support"
    }
))

# Run flow
print(flow.start("welcome"))
result = flow.process_input("user@example.com")
print(result["response"])

from openai import OpenAI
from dataclasses import dataclass


@dataclass
class ErrorContext:
    """Context for error recovery."""
    error_type: str
    message: str
    retry_count: int
    recoverable: bool  # After max_retries, stop looping and escalate


class RobustChatbot:
    """Chatbot with error handling and recovery."""
    
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model
        self.max_retries = 3
        self.error_count = 0
        self.last_error: ErrorContext = None
    
    def _handle_unclear_input(self, message: str) -> str:
        """Handle unclear or ambiguous input."""
        prompt = f"""The user's input was unclear. Generate a helpful clarification request.

User said: "{message}"

Ask for clarification in a friendly way:"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content
    
    def _detect_frustration(self, messages: list[str]) -> bool:
        """Detect if user is frustrated."""
        if len(messages) < 2:
            return False
        
        recent = " ".join(messages[-3:])
        
        prompt = f"""Analyze if the user seems frustrated in these messages.
Consider: repeated questions, escalating tone, explicit frustration.

Messages: "{recent}"

Return just: yes or no"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return "yes" in response.choices[0].message.content.lower()
    
    def _offer_escalation(self) -> str:
        """Offer to escalate to human support."""
        return """I understand this has been frustrating. Would you like me to:
1. Connect you with a human support agent
2. Try a different approach to help
3. Start fresh with a new question

Please let me know how you'd like to proceed."""
    
    def _recover_from_error(self, error: ErrorContext) -> str:
        """Generate recovery message based on error."""
        prompts = {
            "unclear_input": "Ask the user to rephrase their question more specifically.",
            "missing_info": "Politely request the missing information.",
            "validation_failed": "Explain what was wrong and how to correct it.",
            "system_error": "Apologize for the technical issue and offer alternatives.",
        }
        
        prompt = prompts.get(
            error.error_type,
            "Acknowledge the issue and offer to help differently."
        )
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content
    
    def process_with_recovery(
        self,
        message: str,
        history: list[str]
    ) -> dict:
        """Process message with error recovery."""
        try:
            # Check for frustration
            if self._detect_frustration(history + [message]):
                return {
                    "response": self._offer_escalation(),
                    "escalation_offered": True,
                    "error": None
                }
            
            # Normal processing
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "You are a helpful assistant."},
                    {"role": "user", "content": message}
                ]
            )
            
            self.error_count = 0  # Reset on success
            
            return {
                "response": response.choices[0].message.content,
                "escalation_offered": False,
                "error": None
            }
            
        except Exception as e:
            self.error_count += 1
            
            error = ErrorContext(
                error_type="system_error",
                message=str(e),
                retry_count=self.error_count,
                recoverable=self.error_count < self.max_retries
            )
            
            if error.recoverable:
                recovery = self._recover_from_error(error)
            else:
                recovery = self._offer_escalation()
            
            return {
                "response": recovery,
                "escalation_offered": not error.recoverable,
                "error": error
            }


# Usage
bot = RobustChatbot()

# Simulated conversation with potential issues
history = [
    "How do I reset my password?",
    "I already tried that, it didn't work",
    "This is ridiculous, I've been trying for 20 minutes"
]

result = bot.process_with_recovery(
    "This is so frustrating!!!",
    history
)

print(result["response"])
if result["escalation_offered"]:
    print("(Escalation offered to user)")