Chatbot Conversation Design Guide: UX Patterns, Dialog Flows, and User Experience Optimization

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

class DialogState(Enum):
    GREETING = "greeting"
    INTENT_DETECTION = "intent_detection"
    SLOT_FILLING = "slot_filling"
    CONFIRMATION = "confirmation"
    EXECUTION = "execution"
    ERROR_RECOVERY = "error_recovery"
    HANDOFF = "handoff"
    FAREWELL = "farewell"

@dataclass
class ConversationContext:
    session_id: str
    current_state: DialogState = DialogState.GREETING
    slots: dict = field(default_factory=dict)
    turn_count: int = 0
    error_count: int = 0
    max_errors: int = 3
    history: list = field(default_factory=list)

class DialogStateMachine:
    def __init__(self):
        self.transitions: dict[DialogState, dict[str, DialogState]] = {
            DialogState.GREETING: {
                "intent_detected": DialogState.SLOT_FILLING,
                "unclear": DialogState.INTENT_DETECTION,
                "quit": DialogState.FAREWELL,
            },
            DialogState.INTENT_DETECTION: {
                "intent_detected": DialogState.SLOT_FILLING,
                "max_retries": DialogState.HANDOFF,
                "quit": DialogState.FAREWELL,
            },
            DialogState.SLOT_FILLING: {
                "slots_complete": DialogState.CONFIRMATION,
                "missing_slots": DialogState.SLOT_FILLING,
                "error": DialogState.ERROR_RECOVERY,
            },
            DialogState.CONFIRMATION: {
                "confirmed": DialogState.EXECUTION,
                "denied": DialogState.SLOT_FILLING,
                "cancel": DialogState.FAREWELL,
            },
            DialogState.EXECUTION: {
                "success": DialogState.FAREWELL,
                "failure": DialogState.ERROR_RECOVERY,
            },
            DialogState.ERROR_RECOVERY: {
                "retry": DialogState.SLOT_FILLING,
                "escalate": DialogState.HANDOFF,
                "resolved": DialogState.CONFIRMATION,
            },
        }
        self.state_handlers: dict[DialogState, Callable] = {}

    def register_handler(self, state: DialogState, handler: Callable):
        self.state_handlers[state] = handler

    def transition(self, ctx: ConversationContext, event: str) -> DialogState:
        current = ctx.current_state
        if current not in self.transitions:
            raise ValueError(f"No transitions defined for state: {current}")

        if event not in self.transitions[current]:
            ctx.error_count += 1
            if ctx.error_count >= ctx.max_errors:
                ctx.current_state = DialogState.HANDOFF
                return ctx.current_state
            ctx.current_state = DialogState.ERROR_RECOVERY
            return ctx.current_state

        ctx.current_state = self.transitions[current][event]
        ctx.turn_count += 1
        return ctx.current_state

    async def process(self, ctx: ConversationContext, user_input: str) -> str:
        handler = self.state_handlers.get(ctx.current_state)
        if handler is None:
            return "I'm not sure how to help with that. Let me connect you with a human agent."
        return await handler(ctx, user_input)

interface SlotDefinition {
  name: string
  required: boolean
  validator: (value: string) => { valid: boolean; normalized?: string; error?: string }
  prompt: string
  reprompt: string
  maxAttempts: number
}

interface DialogFlow {
  id: string
  slots: SlotDefinition[]
  confirmationTemplate: (slots: Record<string, string>) => string
  execute: (slots: Record<string, string>) => Promise<string>
}

class SlotFillingManager {
  private attempts: Map<string, number> = new Map()

  async fillSlots(
    flow: DialogFlow,
    currentSlots: Record<string, string>,
    userInput: string
  ): Promise<{ response: string; complete: boolean; slots: Record<string, string> }> {
    const missingSlots = flow.slots.filter((s) => s.required && !currentSlots[s.name])

    if (missingSlots.length === 0) {
      return {
        response: flow.confirmationTemplate(currentSlots),
        complete: true,
        slots: currentSlots,
      }
    }

    const currentSlot = missingSlots[0]
    const attemptCount = this.attempts.get(currentSlot.name) ?? 0

    if (userInput) {
      const result = currentSlot.validator(userInput)
      if (result.valid) {
        currentSlots[currentSlot.name] = result.normalized ?? userInput
        this.attempts.delete(currentSlot.name)

        const nextMissing = flow.slots.filter((s) => s.required && !currentSlots[s.name])
        if (nextMissing.length === 0) {
          return {
            response: flow.confirmationTemplate(currentSlots),
            complete: true,
            slots: currentSlots,
          }
        }
        return {
          response: nextMissing[0].prompt,
          complete: false,
          slots: currentSlots,
        }
      } else {
        this.attempts.set(currentSlot.name, attemptCount + 1)
        if (attemptCount + 1 >= currentSlot.maxAttempts) {
          return {
            response:
              "I'm having trouble understanding. Let me transfer you to an agent who can help.",
            complete: false,
            slots: currentSlots,
          }
        }
        return {
          response: `${result.error} ${currentSlot.reprompt}`,
          complete: false,
          slots: currentSlots,
        }
      }
    }

    return {
      response: currentSlot.prompt,
      complete: false,
      slots: currentSlots,
    }
  }
}

// Usage example: appointment booking flow
const appointmentFlow: DialogFlow = {
  id: 'book_appointment',
  slots: [
    {
      name: 'date',
      required: true,
      validator: (v) => {
        const parsed = new Date(v)
        if (isNaN(parsed.getTime()))
          return { valid: false, error: "That doesn't look like a valid date." }
        if (parsed < new Date()) return { valid: false, error: 'Please choose a future date.' }
        return { valid: true, normalized: parsed.toISOString().split('T')[0] }
      },
      prompt: 'What date works best for you?',
      reprompt: "Could you give me a date like 'March 15' or '2026-03-15'?",
      maxAttempts: 3,
    },
    {
      name: 'time',
      required: true,
      validator: (v) => {
        const match = v.match(/(\d{1,2}):?(\d{2})?\s*(am|pm)?/i)
        if (!match) return { valid: false, error: "I couldn't parse that time." }
        return { valid: true, normalized: v.trim() }
      },
      prompt: 'What time would you prefer?',
      reprompt: "Please enter a time like '2:30 PM' or '14:30'.",
      maxAttempts: 3,
    },
  ],
  confirmationTemplate: (slots) =>
    `Great! I have an appointment for ${slots.date} at ${slots.time}. Should I go ahead and book it?`,
  execute: async (slots) => `Your appointment on ${slots.date} at ${slots.time} is confirmed!`,
}

from dataclasses import dataclass
from enum import IntEnum

class ErrorSeverity(IntEnum):
    LOW = 1       # Minor misunderstanding
    MEDIUM = 2    # Repeated misunderstanding
    HIGH = 3      # System error or user frustration
    CRITICAL = 4  # Requires immediate human intervention

@dataclass
class ErrorContext:
    severity: ErrorSeverity
    consecutive_errors: int
    user_sentiment: float  # -1.0 to 1.0
    last_successful_state: str
    error_message: str

class ErrorRecoveryEngine:
    def __init__(self, max_clarifications: int = 2, max_suggestions: int = 2):
        self.max_clarifications = max_clarifications
        self.max_suggestions = max_suggestions

    def determine_strategy(self, ctx: ErrorContext) -> dict:
        # Detect user frustration via sentiment
        if ctx.user_sentiment < -0.5 or ctx.severity == ErrorSeverity.CRITICAL:
            return self._human_handoff(ctx)

        if ctx.consecutive_errors == 0:
            return self._clarify(ctx)
        elif ctx.consecutive_errors <= self.max_clarifications:
            return self._suggest(ctx)
        elif ctx.consecutive_errors <= self.max_clarifications + self.max_suggestions:
            return self._guided_recovery(ctx)
        else:
            return self._human_handoff(ctx)

    def _clarify(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "clarification",
            "message": "I didn't quite catch that. Could you rephrase your request?",
            "show_options": False,
        }

    def _suggest(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "suggestion",
            "message": "I'm not sure I understand. Did you mean one of these?",
            "show_options": True,
            "options": self._get_closest_intents(ctx),
        }

    def _guided_recovery(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "guided_recovery",
            "message": "Let me help you get back on track. What would you like to do?",
            "show_options": True,
            "options": [
                {"label": "Start over", "action": "reset"},
                {"label": "Talk to a human", "action": "handoff"},
                {"label": "Go to main menu", "action": "main_menu"},
            ],
        }

    def _human_handoff(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "human_handoff",
            "message": "I want to make sure you get the help you need. Let me connect you with a team member.",
            "show_options": False,
            "escalate": True,
        }

    def _get_closest_intents(self, ctx: ErrorContext) -> list:
        # In production, use semantic similarity to find closest intents
        return [
            {"label": "Check order status", "action": "intent:order_status"},
            {"label": "Return an item", "action": "intent:return"},
            {"label": "Something else", "action": "intent:other"},
        ]

# BAD: Infinite clarification loop
User: I want to change my thing
Bot: I didn't understand that. Could you rephrase?
User: Change my subscription
Bot: I'm not sure what you mean. Can you try again?
User: CHANGE THE PLAN
Bot: I didn't understand that. Could you rephrase?
User: [leaves in frustration]

# GOOD: Progressive escalation
User: I want to change my thing
Bot: I want to make sure I help you correctly. Did you mean:
     [Change subscription plan] [Update payment method] [Edit profile]
User: Change subscription plan
Bot: Got it! Let me pull up your subscription options...

from dataclasses import dataclass
from typing import Literal

@dataclass
class PersonalityConfig:
    name: str
    tone: Literal["formal", "friendly", "playful", "empathetic"]
    verbosity: Literal["concise", "balanced", "detailed"]
    emoji_usage: bool
    humor_level: float  # 0.0 to 1.0
    formality_level: float  # 0.0 (casual) to 1.0 (formal)
    error_empathy_level: float  # 0.0 to 1.0

    def to_system_prompt(self) -> str:
        tone_guide = {
            "formal": "Use professional, polished language. Avoid slang and colloquialisms.",
            "friendly": "Be warm and approachable. Use conversational language while remaining helpful.",
            "playful": "Be lighthearted and fun. Use casual language and occasional wordplay.",
            "empathetic": "Show deep understanding of user feelings. Acknowledge emotions before solving problems.",
        }

        verbosity_guide = {
            "concise": "Keep responses to 1-2 sentences when possible. Get straight to the point.",
            "balanced": "Provide enough context without overwhelming. 2-4 sentences is ideal.",
            "detailed": "Offer thorough explanations with examples when helpful.",
        }

        emoji_rule = "Use emojis sparingly to add warmth." if self.emoji_usage else "Do not use emojis."

        return f"""You are {self.name}, a helpful assistant.

Tone: {tone_guide[self.tone]}
Verbosity: {verbosity_guide[self.verbosity]}
Emojis: {emoji_rule}

When users encounter errors or frustration:
- Acknowledge the issue with empathy (level: {self.error_empathy_level})
- Never blame the user
- Offer a clear path forward

Always maintain this personality consistently across all interactions."""

# Example configurations for different contexts
support_persona = PersonalityConfig(
    name="Alex",
    tone="empathetic",
    verbosity="balanced",
    emoji_usage=False,
    humor_level=0.1,
    formality_level=0.6,
    error_empathy_level=0.9,
)

sales_persona = PersonalityConfig(
    name="Jordan",
    tone="friendly",
    verbosity="balanced",
    emoji_usage=True,
    humor_level=0.3,
    formality_level=0.4,
    error_empathy_level=0.7,
)

interface OnboardingStep {
  id: string
  message: string
  quickReplies?: string[]
  condition?: (userProfile: UserProfile) => boolean
  nextStep: string | null
}

interface UserProfile {
  isNewUser: boolean
  previousInteractions: number
  preferredLanguage: string
}

const onboardingFlow: OnboardingStep[] = [
  {
    id: 'welcome',
    message:
      "Hi there! I'm your assistant. I can help you with orders, account questions, and product recommendations.",
    quickReplies: ['Show me what you can do', 'I know what I need', 'Talk to a human'],
    nextStep: 'capability_showcase',
    condition: (user) => user.isNewUser,
  },
  {
    id: 'welcome_returning',
    message: 'Welcome back! How can I help you today?',
    quickReplies: ['Check my order', 'Browse products', 'Get support'],
    nextStep: null,
    condition: (user) => !user.isNewUser && user.previousInteractions > 3,
  },
  {
    id: 'capability_showcase',
    message:
      'Here are some things I can do for you:\n\n- Track your orders in real time\n- Help you find the perfect product\n- Process returns and exchanges\n- Answer billing questions\n\nWhat would you like to try first?',
    quickReplies: ['Track an order', 'Find a product', 'Something else'],
    nextStep: null,
  },
]

class OnboardingManager {
  private completedSteps: Set<string> = new Set()

  getNextStep(userProfile: UserProfile): OnboardingStep | null {
    for (const step of onboardingFlow) {
      if (this.completedSteps.has(step.id)) continue
      if (step.condition && !step.condition(userProfile)) continue
      return step
    }
    return null
  }

  markCompleted(stepId: string): void {
    this.completedSteps.add(stepId)
  }

  shouldShowOnboarding(userProfile: UserProfile): boolean {
    return userProfile.isNewUser || userProfile.previousInteractions < 2
  }
}

import json
from datetime import datetime, timezone
from dataclasses import dataclass, asdict
from typing import Optional

@dataclass
class ConversationEvent:
    session_id: str
    timestamp: str
    event_type: str  # "message", "state_change", "error", "handoff", "completion"
    state: str
    user_input: Optional[str] = None
    bot_response: Optional[str] = None
    intent: Optional[str] = None
    confidence: Optional[float] = None
    metadata: Optional[dict] = None

class ConversationAnalytics:
    def __init__(self, storage_backend):
        self.storage = storage_backend
        self.session_events: dict[str, list] = {}

    def track_event(self, event: ConversationEvent):
        if event.session_id not in self.session_events:
            self.session_events[event.session_id] = []
        self.session_events[event.session_id].append(event)
        self.storage.store(asdict(event))

    def compute_metrics(self, time_window_hours: int = 24) -> dict:
        sessions = self._get_recent_sessions(time_window_hours)
        total = len(sessions)
        if total == 0:
            return {"error": "No sessions in time window"}

        completed = sum(1 for s in sessions if self._is_completed(s))
        handed_off = sum(1 for s in sessions if self._has_handoff(s))
        errored = sum(1 for s in sessions if self._has_errors(s))

        avg_turns = sum(self._count_turns(s) for s in sessions) / total

        drop_off_states: dict[str, int] = {}
        for s in sessions:
            if not self._is_completed(s) and not self._has_handoff(s):
                last_state = s[-1].state if s else "unknown"
                drop_off_states[last_state] = drop_off_states.get(last_state, 0) + 1

        return {
            "total_sessions": total,
            "task_completion_rate": completed / total,
            "handoff_rate": handed_off / total,
            "error_rate": errored / total,
            "avg_turns_to_resolution": round(avg_turns, 1),
            "drop_off_hotspots": drop_off_states,
            "containment_rate": (total - handed_off) / total,
        }

    def identify_improvement_opportunities(self, metrics: dict) -> list[str]:
        opportunities = []
        if metrics.get("task_completion_rate", 1) < 0.7:
            opportunities.append(
                "Task completion rate below 70%. Review drop-off hotspots and simplify those flows."
            )
        if metrics.get("handoff_rate", 0) > 0.3:
            opportunities.append(
                "Handoff rate above 30%. Analyze handoff triggers and add automation for top handoff reasons."
            )
        if metrics.get("avg_turns_to_resolution", 0) > 8:
            opportunities.append(
                "Average turns too high. Consider combining slot-filling steps or adding quick-reply buttons."
            )
        hotspots = metrics.get("drop_off_hotspots", {})
        for state, count in sorted(hotspots.items(), key=lambda x: -x[1])[:3]:
            opportunities.append(
                f"High drop-off at '{state}' state ({count} sessions). Investigate UX and error handling."
            )
        return opportunities

    def _get_recent_sessions(self, hours: int) -> list[list[ConversationEvent]]:
        return list(self.session_events.values())

    def _is_completed(self, events: list[ConversationEvent]) -> bool:
        return any(e.event_type == "completion" for e in events)

    def _has_handoff(self, events: list[ConversationEvent]) -> bool:
        return any(e.event_type == "handoff" for e in events)

    def _has_errors(self, events: list[ConversationEvent]) -> bool:
        return any(e.event_type == "error" for e in events)

    def _count_turns(self, events: list[ConversationEvent]) -> int:
        return sum(1 for e in events if e.event_type == "message")

interface ExperimentVariant {
  id: string
  name: string
  weight: number // 0.0 to 1.0, all variants must sum to 1.0
  config: Record<string, unknown>
}

interface Experiment {
  id: string
  name: string
  description: string
  variants: ExperimentVariant[]
  metrics: string[]
  startDate: string
  endDate: string | null
  status: 'draft' | 'running' | 'paused' | 'completed'
}

interface ExperimentResult {
  variantId: string
  sampleSize: number
  metrics: Record<string, number>
}

class ConversationExperimentEngine {
  private experiments: Map<string, Experiment> = new Map()
  private assignments: Map<string, Map<string, string>> = new Map()

  createExperiment(experiment: Experiment): void {
    const totalWeight = experiment.variants.reduce((sum, v) => sum + v.weight, 0)
    if (Math.abs(totalWeight - 1.0) > 0.001) {
      throw new Error(`Variant weights must sum to 1.0, got ${totalWeight}`)
    }
    this.experiments.set(experiment.id, experiment)
  }

  assignVariant(experimentId: string, userId: string): ExperimentVariant | null {
    const experiment = this.experiments.get(experimentId)
    if (!experiment || experiment.status !== 'running') return null

    // Check for existing assignment (sticky sessions)
    const userAssignments = this.assignments.get(userId)
    if (userAssignments?.has(experimentId)) {
      const variantId = userAssignments.get(experimentId)!
      return experiment.variants.find((v) => v.id === variantId) ?? null
    }

    // Deterministic assignment based on user ID hash
    const hash = this.hashUserId(userId, experimentId)
    const normalized = hash / 0xffffffff
    let cumWeight = 0
    for (const variant of experiment.variants) {
      cumWeight += variant.weight
      if (normalized <= cumWeight) {
        if (!this.assignments.has(userId)) {
          this.assignments.set(userId, new Map())
        }
        this.assignments.get(userId)!.set(experimentId, variant.id)
        return variant
      }
    }
    return experiment.variants[experiment.variants.length - 1]
  }

  private hashUserId(userId: string, salt: string): number {
    const str = userId + salt
    let hash = 0
    for (let i = 0; i < str.length; i++) {
      const char = str.charCodeAt(i)
      hash = (hash << 5) - hash + char
      hash = hash & hash
    }
    return Math.abs(hash)
  }
}

// Example: Testing two onboarding strategies
const onboardingExperiment: Experiment = {
  id: 'onboarding_v2',
  name: 'Onboarding Flow Comparison',
  description: 'Test guided tour vs quick-start menu for new users',
  variants: [
    {
      id: 'guided_tour',
      name: 'Guided Tour',
      weight: 0.5,
      config: { onboardingStyle: 'guided_tour', showExamples: true },
    },
    {
      id: 'quick_start',
      name: 'Quick Start Menu',
      weight: 0.5,
      config: { onboardingStyle: 'quick_start', showExamples: false },
    },
  ],
  metrics: ['task_completion_rate', 'time_to_first_action', 'return_rate_7d'],
  startDate: '2026-03-01',
  endDate: null,
  status: 'running',
}