チャットボット会話設計ガイド: UXパターン、ダイアログフロー、ユーザー体験最適化

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 "申し訳ございませんが、対応方法がわかりません。担当者におつなぎします。"
        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: 'うまく理解できませんでした。担当者におつなぎいたします。',
            complete: false,
            slots: currentSlots,
          }
        }
        return {
          response: `${result.error} ${currentSlot.reprompt}`,
          complete: false,
          slots: currentSlots,
        }
      }
    }

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

// 使用例：予約フロー
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: '有効な日付ではないようです。' }
        if (parsed < new Date()) return { valid: false, error: '未来の日付を選択してください。' }
        return { valid: true, normalized: parsed.toISOString().split('T')[0] }
      },
      prompt: 'ご希望の日付を教えてください。',
      reprompt: '「3月15日」や「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: '時間を解析できませんでした。' }
        return { valid: true, normalized: v.trim() }
      },
      prompt: 'ご希望の時間帯はいつですか？',
      reprompt: '「14:30」や「午後2時30分」のような形式で入力してください。',
      maxAttempts: 3,
    },
  ],
  confirmationTemplate: (slots) =>
    `${slots.date}の${slots.time}で予約を承りました。このまま予約を確定してよろしいですか？`,
  execute: async (slots) => `${slots.date}の${slots.time}の予約が確定しました！`,
}

from dataclasses import dataclass
from enum import IntEnum

class ErrorSeverity(IntEnum):
    LOW = 1       # 軽微な誤解
    MEDIUM = 2    # 繰り返しの誤解
    HIGH = 3      # システムエラーまたはユーザーフラストレーション
    CRITICAL = 4  # 即座の人間介入が必要

@dataclass
class ErrorContext:
    severity: ErrorSeverity
    consecutive_errors: int
    user_sentiment: float  # -1.0 から 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:
        # センチメントでユーザーフラストレーションを検出
        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": "うまく聞き取れませんでした。もう一度言い換えていただけますか？",
            "show_options": False,
        }

    def _suggest(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "suggestion",
            "message": "よく理解できませんでした。以下のいずれかでしょうか？",
            "show_options": True,
            "options": self._get_closest_intents(ctx),
        }

    def _guided_recovery(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "guided_recovery",
            "message": "元の流れに戻りましょう。何をされたいですか？",
            "show_options": True,
            "options": [
                {"label": "最初からやり直す", "action": "reset"},
                {"label": "担当者と話す", "action": "handoff"},
                {"label": "メインメニューに戻る", "action": "main_menu"},
            ],
        }

    def _human_handoff(self, ctx: ErrorContext) -> dict:
        return {
            "strategy": "human_handoff",
            "message": "適切なサポートをお届けしたいので、担当者におつなぎいたします。",
            "show_options": False,
            "escalate": True,
        }

    def _get_closest_intents(self, ctx: ErrorContext) -> list:
        # 本番では、セマンティック類似度で最も近いインテントを検索
        return [
            {"label": "注文状況の確認", "action": "intent:order_status"},
            {"label": "商品の返品", "action": "intent:return"},
            {"label": "その他", "action": "intent:other"},
        ]

# 悪い例: 無限明確化ループ
User: 私の設定を変えたい
Bot: 理解できませんでした。もう一度お願いします。
User: サブスクリプションの変更
Bot: よくわかりません。もう一度試してください。
User: プランを変更して！
Bot: 理解できませんでした。もう一度お願いします。
User: [フラストレーションで離脱]

# 良い例: 段階的エスカレーション
User: 私の設定を変えたい
Bot: 正確にお手伝いしたいので確認させてください。以下のどれでしょうか？
     [サブスクリプションプラン変更] [支払い方法の更新] [プロフィール編集]
User: サブスクリプションプラン変更
Bot: 承知しました！サブスクリプションのオプションを表示します...

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 から 1.0
    formality_level: float  # 0.0（カジュアル）から 1.0（フォーマル）
    error_empathy_level: float  # 0.0 から 1.0

    def to_system_prompt(self) -> str:
        tone_guide = {
            "formal": "プロフェッショナルで洗練された言葉遣いを使用。スラングや口語表現を避ける。",
            "friendly": "温かくフレンドリーに。親しみやすい会話調を維持しつつ丁寧に対応。",
            "playful": "軽快で楽しい雰囲気を演出。カジュアルな言葉遣いや言葉遊びも適度に使用。",
            "empathetic": "ユーザーの気持ちに深い理解を示す。問題解決の前に感情を認める。",
        }

        verbosity_guide = {
            "concise": "可能な限り1〜2文で応答。要点を端的に伝える。",
            "balanced": "過不足のない情報提供。2〜4文が理想的。",
            "detailed": "必要に応じて具体例を含む丁寧な説明を提供。",
        }

        emoji_rule = "絵文字を控えめに使用して親しみを出す。" if self.emoji_usage else "絵文字は使用しない。"

        return f"""あなたは{self.name}、親切なアシスタントです。

トーン: {tone_guide[self.tone]}
文量: {verbosity_guide[self.verbosity]}
絵文字: {emoji_rule}

ユーザーがエラーやフラストレーションに遭遇した場合:
- 共感を持って問題を認める（レベル: {self.error_empathy_level}）
- ユーザーを責めない
- 明確な解決策を提示する

すべてのインタラクションでこのパーソナリティを一貫して維持してください。"""

# 異なるコンテキストの設定例
support_persona = PersonalityConfig(
    name="アキラ",
    tone="empathetic",
    verbosity="balanced",
    emoji_usage=False,
    humor_level=0.1,
    formality_level=0.6,
    error_empathy_level=0.9,
)

sales_persona = PersonalityConfig(
    name="ユウキ",
    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:
      'こんにちは！アシスタントです。注文に関すること、アカウントの質問、おすすめ商品のご案内をお手伝いできます。',
    quickReplies: ['できることを教えて', 'やりたいことがある', '担当者と話したい'],
    nextStep: 'capability_showcase',
    condition: (user) => user.isNewUser,
  },
  {
    id: 'welcome_returning',
    message: 'おかえりなさい！本日はどのようなご用件でしょうか？',
    quickReplies: ['注文確認', '商品を探す', 'サポートを受ける'],
    nextStep: null,
    condition: (user) => !user.isNewUser && user.previousInteractions > 3,
  },
  {
    id: 'capability_showcase',
    message:
      '以下のことをお手伝いできます：\n\n- 注文のリアルタイム追跡\n- 最適な商品の検索\n- 返品・交換の手続き\n- 請求に関するご質問への回答\n\nまず何から試してみますか？',
    quickReplies: ['注文を追跡する', '商品を探す', 'その他'],
    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": "指定期間内にセッションがありません"}

        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(
                "タスク完了率が70%を下回っています。離脱ホットスポットを確認し、フローを簡素化してください。"
            )
        if metrics.get("handoff_rate", 0) > 0.3:
            opportunities.append(
                "ハンドオフ率が30%を超えています。ハンドオフのトリガーを分析し、上位の理由に自動化を追加してください。"
            )
        if metrics.get("avg_turns_to_resolution", 0) > 8:
            opportunities.append(
                "平均ターン数が多すぎます。スロット入力ステップの統合やクイックリプライボタンの追加を検討してください。"
            )
        hotspots = metrics.get("drop_off_hotspots", {})
        for state, count in sorted(hotspots.items(), key=lambda x: -x[1])[:3]:
            opportunities.append(
                f"'{state}'ステートで高い離脱率（{count}セッション）。UXとエラーハンドリングを調査してください。"
            )
        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 から 1.0、全バリアントの合計は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(`バリアントの重みの合計は1.0でなければなりません。現在: ${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

    // 既存の割り当てを確認（スティッキーセッション）
    const userAssignments = this.assignments.get(userId)
    if (userAssignments?.has(experimentId)) {
      const variantId = userAssignments.get(experimentId)!
      return experiment.variants.find((v) => v.id === variantId) ?? null
    }

    // ユーザーIDハッシュに基づく決定論的割り当て
    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)
  }
}

// 例: 2つのオンボーディング戦略のテスト
const onboardingExperiment: Experiment = {
  id: 'onboarding_v2',
  name: 'オンボーディングフロー比較',
  description: '新規ユーザー向けにガイドツアーとクイックスタートメニューを比較テスト',
  variants: [
    {
      id: 'guided_tour',
      name: 'ガイドツアー',
      weight: 0.5,
      config: { onboardingStyle: 'guided_tour', showExamples: true },
    },
    {
      id: 'quick_start',
      name: 'クイックスタートメニュー',
      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',
}