VoiceBotの設定を行う対話に関するFSA

dialogue_graph.py

#!/usr/bin/env python3
"""
pydantic-graph による対話状態遷移エンジン実装
"""

import asyncio
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
import uuid

from pydantic import BaseModel
from pydantic_graph import Graph, BaseNode, End, GraphRunContext


class FlowType(StrEnum):
    """対話フロータイプ"""

    UNIFORM = "uniform"
    CONDITIONAL = "conditional"


class ActionType(StrEnum):
    """アクションタイプ"""

    AI_RECEPTION = "ai_reception"
    TRANSFER = "transfer"
    GUIDANCE = "guidance"
    SMS = "sms"


class InputType(StrEnum):
    """UI入力タイプ"""

    RADIO = "radio"
    RADIO_CARD = "radio-card"
    CHECKBOX = "checkbox"
    TEXT_SIMPLE = "text-simple"
    TEXT_LIST_TEL = "text-list-tel"
    RADIO_TEXTAREA = "radio-textarea"


class Answer(BaseModel):
    """ユーザー回答"""

    value: str | list[str]  # 単一選択 or 複数選択
    text: str  # UIに表示されるテキスト


class Rule(BaseModel):
    """電話応対ルール"""

    id: str
    query: str
    action: ActionType
    details: list[str] | str  # transfer/ai_reception: list, guidance/sms: str
    announcement: str | None = None


class Question(BaseModel):
    """質問定義"""

    id: str
    text: str
    type: InputType
    options: list[str | dict[str, Any]] | None = None
    placeholder: str | None = None
    is_optional: bool = False
    layout: str | None = None
    samples: list[str] | None = None


# ================================
# 質問定義データ
# ================================

QUESTION_DEFINITIONS = {
    "start": Question(
        id="start",
        text="どのような電話応対を希望しますか？<br><span class='text-base text-gray-500 font-normal'>まずは、実現したい応対のタイプを選択してください。</span>",
        type=InputType.RADIO_CARD,
        options=[
            {
                "text": "すべての電話に同じ案内をする",
                "value": "uniform",
                "description": "営業時間外の案内や、すべての電話を同じ窓口に転送するなど、シンプルなルールを設定します。",
            },
            {
                "text": "問い合わせ内容に応じて案内を分ける",
                "value": "conditional",
                "description": "「予約」「料金」「その他」など、お客様の用件に応じて、最適な案内や担当者へ自動で振り分けます。",
            },
        ],
    ),
    # 一律応対フロー（対話パス#1）
    "uniform_action": Question(
        id="uniform_action",
        text="承知しました。「すべての電話に同じ案内をする」ですね。<br>電話に、どのような対応をしますか？",
        type=InputType.RADIO,
        options=[
            "用件をヒアリングして、テキストで受け取る",
            "担当者の電話に転送する",
            "別の時間にかけ直すよう案内する",
        ],
    ),
    "uniform_transfer_number": Question(
        id="uniform_transfer_number",
        text="転送先の電話番号を入力してください。複数登録も可能です。入力後、「追加」ボタンを押してください。",
        type=InputType.TEXT_LIST_TEL,
        placeholder="例: 09012345678",
    ),
    "uniform_guidance_text": Question(
        id="uniform_guidance_text",
        text="お客様に何と伝えますか？案内したいメッセージのサンプルを選択するか、直接編集してください。",
        type=InputType.RADIO_TEXTAREA,
        samples=[
            "恐れ入りますが、平日の午前10時から午後5時の間におかけ直しください。",
            "ただいまのお時間は営業を終了しております。明日午前9時以降に改めてお電話いただきますようお願いいたします。",
            "現在、電話が大変混み合っております。しばらく経ってからおかけ直しください。",
        ],
    ),
    # 条件分岐フロー（対話パス#2共通部分）
    "conditional_industry": Question(
        id="conditional_industry",
        text="承知しました。「問い合わせ内容に応じて案内を分ける」ですね。<br>まず、あなたのビジネスの業種を教えてください。",
        type=InputType.RADIO,
        layout="grid-2",
        options=[
            "飲食店",
            "クリニック・病院",
            "美容室・サロン",
            "不動産",
            "学習塾・スクール",
            "その他",
        ],
    ),
    "conditional_use_case": Question(
        id="conditional_use_case",
        text="この電話番号は、主にどのような目的で使われますか？",
        type=InputType.RADIO,
        options=[
            "会社の代表電話",
            "お客様のお問い合わせ窓口",
            "特定の部署の電話",
            "その他",
        ],
    ),
    "conditional_caller_type": Question(
        id="conditional_caller_type",
        text="次に、主にどのような方から電話がかかってきますか？（複数選択可）",
        type=InputType.CHECKBOX,
        options=["既存のお客様", "新規のお客様", "取引先", "求人応募者", "その他"],
    ),
    "conditional_website": Question(
        id="conditional_website",
        text="もしあれば、会社のホームページのURLを教えてください。Webサイトの内容を参考に、より精度の高いルールを提案します。（任意）",
        type=InputType.TEXT_SIMPLE,
        placeholder="https://example.com",
        is_optional=True,
    ),
}

# ================================
# 業種別ルール生成エンジン
# ================================


class IndustryRuleGenerator:
    """全6業種対応のルール生成エンジン"""

    INDUSTRY_PATTERNS = {
        "クリニック・病院": {
            "queries": [
                "診療予約",
                "予約の変更・キャンセル",
                "診療時間の確認",
                "場所・アクセス",
                "予防接種",
            ],
            "actions": {
                "診療予約": {
                    "action": "ai_reception",
                    "details": [
                        "お名前",
                        "診察券番号",
                        "ご希望の日時",
                        "ご希望の診療科",
                    ],
                },
                "予約の変更・キャンセル": {
                    "action": "transfer",
                    "details": ["09012345678"],
                },
                "診療時間の確認": {
                    "action": "guidance",
                    "details": "診療時間は、平日の午前9時から午後6時までです。土曜日は午後1時までとなります。",
                },
                "場所・アクセス": {
                    "action": "guidance",
                    "details": "当院は〇〇駅西口から徒歩5分、市役所の隣のビル3階です。",
                },
                "予防接種": {"action": "sms", "details": "https://example.com/vaccine"},
            },
        },
        "飲食店": {
            "queries": [
                "予約",
                "営業時間の確認",
                "デリバリー・テイクアウト",
                "場所・アクセス",
                "その他",
            ],
            "actions": {
                "予約": {
                    "action": "ai_reception",
                    "details": ["お名前", "人数", "ご希望の日時"],
                },
                "営業時間の確認": {
                    "action": "guidance",
                    "details": "営業時間は、毎日午前11時から午後10時までです。ラストオーダーは午後9時半です。",
                },
                "デリバリー・テイクアウト": {
                    "action": "sms",
                    "details": "https://example.com/takeout-menu",
                },
                "場所・アクセス": {
                    "action": "guidance",
                    "details": "当店は〇〇駅東口から徒歩3分、赤い看板が目印です。",
                },
                "その他": {"action": "transfer", "details": ["09012345678"]},
            },
        },
        "美容室・サロン": {
            "queries": [
                "予約",
                "予約の変更・キャンセル",
                "メニュー・料金の確認",
                "場所・アクセス",
                "その他",
            ],
            "actions": {
                "予約": {
                    "action": "ai_reception",
                    "details": ["お名前", "ご希望の日時", "ご希望のメニュー"],
                },
                "予約の変更・キャンセル": {
                    "action": "transfer",
                    "details": ["09012345678"],
                },
                "メニュー・料金の確認": {
                    "action": "sms",
                    "details": "https://example.com/menu",
                },
                "場所・アクセス": {
                    "action": "guidance",
                    "details": "当店は〇〇駅南口から徒歩2分です。",
                },
                "その他": {"action": "transfer", "details": ["09012345678"]},
            },
        },
        "不動産": {
            "queries": [
                "物件の問い合わせ",
                "内見の予約",
                "入居に関する手続き",
                "家賃・契約について",
                "その他",
            ],
            "actions": {
                "物件の問い合わせ": {
                    "action": "ai_reception",
                    "details": ["お名前", "ご希望のエリア", "予算"],
                },
                "内見の予約": {"action": "transfer", "details": ["09012345678"]},
                "入居に関する手続き": {
                    "action": "transfer",
                    "details": ["08098765432"],
                },
                "家賃・契約について": {
                    "action": "sms",
                    "details": "https://example.com/contract",
                },
                "その他": {"action": "transfer", "details": ["09012345678"]},
            },
        },
        "学習塾・スクール": {
            "queries": [
                "入塾・体験授業の申し込み",
                "授業料・コースの確認",
                "欠席・振替の連絡",
                "その他",
            ],
            "actions": {
                "入塾・体験授業の申し込み": {
                    "action": "ai_reception",
                    "details": ["お名前", "学年", "ご希望のコース"],
                },
                "授業料・コースの確認": {
                    "action": "sms",
                    "details": "https://example.com/courses",
                },
                "欠席・振替の連絡": {"action": "transfer", "details": ["09012345678"]},
                "その他": {"action": "transfer", "details": ["09012345678"]},
            },
        },
        "その他": {
            "queries": [
                "サービスに関する質問",
                "料金に関する質問",
                "担当者への連絡",
                "採用に関する問い合わせ",
                "その他",
            ],
            "actions": {
                "サービスに関する質問": {
                    "action": "transfer",
                    "details": ["09012345678"],
                },
                "料金に関する質問": {
                    "action": "sms",
                    "details": "https://example.com/pricing",
                },
                "担当者への連絡": {"action": "transfer", "details": ["09012345678"]},
                "採用に関する問い合わせ": {
                    "action": "transfer",
                    "details": ["08098765432"],
                },
                "その他": {"action": "ai_reception", "details": ["お名前", "ご用件"]},
            },
        },
    }

    @classmethod
    def generate_rules(cls, industry: str) -> list[Rule]:
        """業種に応じたルール5つを自動生成"""
        pattern = cls.INDUSTRY_PATTERNS.get(industry, cls.INDUSTRY_PATTERNS["その他"])
        rules = []

        for i, query in enumerate(pattern["queries"]):
            action_config = pattern["actions"][query]
            rule = Rule(
                id=f"rule_{i}",
                query=query,
                action=ActionType(action_config["action"]),
                details=action_config["details"],
                announcement=cls._generate_announcement(query, action_config["action"]),
            )
            rules.append(rule)
        return rules

    @classmethod
    def _generate_announcement(cls, query: str, action: str) -> str | None:
        """アクションタイプ別の案内文生成"""
        if action == "guidance":
            return None  # guidanceは詳細内容を直接読み上げ

        action_messages = {
            "transfer": "担当者にお繋ぎします。少々お待ちください。",
            "sms": "SMSで情報をお送りしますので、ご確認ください。",
            "ai_reception": "それではご用件をお伺いしますので、続けてお話しください。",
        }

        message = action_messages.get(action, "少々お待ちください。")
        return f"「{query}ですね。{message}」"


@dataclass
class DialogueGraphState:
    """pydantic-graph用の対話状態"""

    session_id: str = ""
    current_question_id: str | None = None
    flow_type: FlowType | None = None
    answers: dict[str, Answer] | None = None
    question_history: list[str] | None = None
    generated_conditional_rules: list[Rule] | None = None
    current_rule_index: int = 0
    company_name: str = "貴社"

    def __post_init__(self):
        """初期化後処理"""
        if not self.session_id:
            self.session_id = str(uuid.uuid4())
        if self.answers is None:
            self.answers = {}
        if self.question_history is None:
            self.question_history = []
        if self.generated_conditional_rules is None:
            self.generated_conditional_rules = []


@dataclass
class DialogueDeps:
    """対話処理に必要な依存性"""

    user_input: str | list[str] | None = None
    user_input_text: str | None = None


# ================================
# Node Implementations
# ================================


@dataclass
class EndUniformNode(BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]):
    """一律応対フロー終了ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> End[dict[str, Any]]:
        """終了処理"""
        return End(
            {
                "node_type": "end",
                "flow_type": "uniform",
                "message": "一律応対設定が完了しました！",
                "final_state": {
                    "session_id": ctx.state.session_id,
                    "flow_type": ctx.state.flow_type.value
                    if ctx.state.flow_type
                    else None,
                    "answers": {
                        k: {"value": v.value, "text": v.text}
                        for k, v in ctx.state.answers.items()
                    }
                    if ctx.state.answers
                    else {},
                },
            }
        )


@dataclass
class UniformTransferNumberNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """電話転送番号入力ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> EndUniformNode | End[dict[str, Any]]:
        """転送番号入力処理"""
        question = QUESTION_DEFINITIONS["uniform_transfer_number"]

        # ユーザー入力がある場合は終了ノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return EndUniformNode()

        # 初回表示時は質問情報を返す
        return End(
            {
                "node_type": "question",
                "question": {
                    "id": question.id,
                    "text": question.text,
                    "type": question.type.value,
                    "placeholder": question.placeholder,
                },
            }
        )


@dataclass
class UniformGuidanceTextNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """案内テキスト入力ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> EndUniformNode | End[dict[str, Any]]:
        """案内テキスト入力処理"""
        question = QUESTION_DEFINITIONS["uniform_guidance_text"]

        # ユーザー入力がある場合は終了ノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return EndUniformNode()

        # 初回表示時は質問情報を返す
        return End(
            {
                "node_type": "question",
                "question": {
                    "id": question.id,
                    "text": question.text,
                    "type": question.type.value,
                    "samples": question.samples,
                },
            }
        )


@dataclass
class UniformActionNode(BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]):
    """一律応対アクション選択ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> (
        UniformTransferNumberNode
        | UniformGuidanceTextNode
        | EndUniformNode
        | End[dict[str, Any]]
    ):
        """一律応対アクション処理"""
        question = QUESTION_DEFINITIONS["uniform_action"]

        # ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )

            answer = ctx.deps.user_input
            if answer == "用件をヒアリングして、テキストで受け取る":
                return EndUniformNode()
            elif answer == "担当者の電話に転送する":
                return UniformTransferNumberNode()
            elif answer == "別の時間にかけ直すよう案内する":
                return UniformGuidanceTextNode()

        # 初回表示時は質問情報を返す
        return End(
            {
                "node_type": "question",
                "question": {
                    "id": question.id,
                    "text": question.text,
                    "type": question.type.value,
                    "options": question.options,
                    "placeholder": question.placeholder,
                    "is_optional": question.is_optional,
                },
            }
        )


@dataclass
class EndConditionalNode(BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]):
    """条件分岐フロー終了ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> End[dict[str, Any]]:
        """終了処理"""
        return End(
            {
                "node_type": "end",
                "flow_type": "conditional",
                "message": "条件分岐設定が完了しました！",
                "final_state": {
                    "session_id": ctx.state.session_id,
                    "flow_type": ctx.state.flow_type.value
                    if ctx.state.flow_type
                    else None,
                    "answers": {
                        k: {"value": v.value, "text": v.text}
                        for k, v in ctx.state.answers.items()
                    }
                    if ctx.state.answers
                    else {},
                    "company_name": ctx.state.company_name,
                    "generated_rules": [
                        {
                            "id": rule.id,
                            "query": rule.query,
                            "action": rule.action.value,
                            "details": rule.details,
                            "announcement": rule.announcement,
                        }
                        for rule in ctx.state.generated_conditional_rules
                    ]
                    if ctx.state.generated_conditional_rules
                    else [],
                },
            }
        )


@dataclass
class RuleGenerationNode(BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]):
    """ルール生成ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> EndConditionalNode | End[dict[str, Any]]:
        """ルール生成処理"""
        # 業種ベースでルール生成
        if not ctx.state.answers or not (
            industry := ctx.state.answers.get("conditional_industry")
        ):
            return End({"error": "Industry not specified"})

        industry_value = (
            industry.value if isinstance(industry.value, str) else industry.value[0]
        )
        ctx.state.generated_conditional_rules = IndustryRuleGenerator.generate_rules(
            industry_value
        )
        ctx.state.current_rule_index = 0

        # 簡単化のため直接終了ノードに進む
        return EndConditionalNode()


@dataclass
class ConditionalWebsiteNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """ウェブサイトURL入力ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> RuleGenerationNode | End[dict[str, Any]]:
        """ウェブサイトURL入力処理"""
        question = QUESTION_DEFINITIONS["conditional_website"]

        # 現在の質問が既に表示済みで、ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.current_question_id == question.id
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return RuleGenerationNode()

        # 質問をまだ表示していない場合は質問情報を返す
        if ctx.state.current_question_id != question.id:
            ctx.state.current_question_id = question.id
            return End(
                {
                    "node_type": "question",
                    "question": {
                        "id": question.id,
                        "text": question.text,
                        "type": question.type.value,
                        "placeholder": question.placeholder,
                        "is_optional": question.is_optional,
                    },
                }
            )

        # その他の場合（重複実行回避）
        return End({"error": "Duplicate execution detected"})


@dataclass
class ConditionalCallerTypeNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """発信者タイプ選択ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> ConditionalWebsiteNode | End[dict[str, Any]]:
        """発信者タイプ選択処理"""
        question = QUESTION_DEFINITIONS["conditional_caller_type"]

        # 現在の質問が既に表示済みで、ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.current_question_id == question.id
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return ConditionalWebsiteNode()

        # 質問をまだ表示していない場合は質問情報を返す
        if ctx.state.current_question_id != question.id:
            ctx.state.current_question_id = question.id
            return End(
                {
                    "node_type": "question",
                    "question": {
                        "id": question.id,
                        "text": question.text,
                        "type": question.type.value,
                        "options": question.options,
                    },
                }
            )

        # その他の場合（重複実行回避）
        return End({"error": "Duplicate execution detected"})


@dataclass
class ConditionalUseCaseNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """用途選択ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> ConditionalCallerTypeNode | End[dict[str, Any]]:
        """用途選択処理"""
        question = QUESTION_DEFINITIONS["conditional_use_case"]

        # 現在の質問が既に表示済みで、ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.current_question_id == question.id
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return ConditionalCallerTypeNode()

        # 質問をまだ表示していない場合は質問情報を返す
        if ctx.state.current_question_id != question.id:
            ctx.state.current_question_id = question.id
            return End(
                {
                    "node_type": "question",
                    "question": {
                        "id": question.id,
                        "text": question.text,
                        "type": question.type.value,
                        "options": question.options,
                    },
                }
            )

        # その他の場合（重複実行回避）
        return End({"error": "Duplicate execution detected"})


@dataclass
class ConditionalIndustryNode(
    BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]
):
    """業種選択ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> ConditionalUseCaseNode | End[dict[str, Any]]:
        """業種選択処理"""
        question = QUESTION_DEFINITIONS["conditional_industry"]

        # 現在の質問が既に表示済みで、ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.current_question_id == question.id
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )
            return ConditionalUseCaseNode()

        # 質問をまだ表示していない場合は質問情報を返す
        if ctx.state.current_question_id != question.id:
            ctx.state.current_question_id = question.id
            return End(
                {
                    "node_type": "question",
                    "question": {
                        "id": question.id,
                        "text": question.text,
                        "type": question.type.value,
                        "options": question.options,
                        "layout": question.layout,
                    },
                }
            )

        # その他の場合（重複実行回避）
        return End({"error": "Duplicate execution detected"})


@dataclass
class StartNode(BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]]):
    """開始ノード"""

    async def run(
        self, ctx: GraphRunContext[DialogueGraphState, DialogueDeps]
    ) -> UniformActionNode | ConditionalIndustryNode | End[dict[str, Any]]:
        """開始処理"""
        question = QUESTION_DEFINITIONS["start"]
        ctx.state.current_question_id = question.id

        # ユーザー入力がある場合は次のノードに遷移
        if (
            ctx.deps
            and ctx.deps.user_input is not None
            and ctx.state.answers is not None
        ):
            ctx.state.answers[question.id] = Answer(
                value=ctx.deps.user_input,
                text=ctx.deps.user_input_text or str(ctx.deps.user_input),
            )

            # フロータイプ決定と遷移
            if ctx.deps.user_input == "uniform":
                ctx.state.flow_type = FlowType.UNIFORM
                return UniformActionNode()
            elif ctx.deps.user_input == "conditional":
                ctx.state.flow_type = FlowType.CONDITIONAL
                return ConditionalIndustryNode()

        # 初回表示時は質問情報を返す
        return End(
            {
                "node_type": "question",
                "question": {
                    "id": question.id,
                    "text": question.text,
                    "type": question.type.value,
                    "options": question.options,
                    "placeholder": question.placeholder,
                    "is_optional": question.is_optional,
                    "layout": question.layout,
                    "samples": question.samples,
                },
            }
        )


# ================================
# Graph Engine
# ================================


class DialogueGraphEngine:
    """pydantic-graph ベースの対話状態遷移エンジン"""

    def __init__(self):
        """グラフエンジン初期化"""
        self.graph = Graph(
            nodes=[
                StartNode,
                UniformActionNode,
                UniformTransferNumberNode,
                UniformGuidanceTextNode,
                EndUniformNode,
                ConditionalIndustryNode,
                ConditionalUseCaseNode,
                ConditionalCallerTypeNode,
                ConditionalWebsiteNode,
                RuleGenerationNode,
                EndConditionalNode,
            ]
        )

    async def start_dialogue(self) -> tuple[dict[str, Any], DialogueGraphState]:
        """対話開始"""
        state = DialogueGraphState()
        deps = DialogueDeps()

        result = await self.graph.run(StartNode(), state=state, deps=deps)

        return result.output, state

    async def process_user_input(
        self,
        user_input: str | list[str],
        user_input_text: str,
        state: DialogueGraphState,
        current_node: BaseNode[DialogueGraphState, DialogueDeps, dict[str, Any]],
    ) -> tuple[dict[str, Any], DialogueGraphState]:
        """ユーザー入力処理"""
        deps = DialogueDeps(user_input=user_input, user_input_text=user_input_text)

        result = await self.graph.run(current_node, state=state, deps=deps)

        return result.output, state


# ================================
# Test Function
# ================================


async def test_dialogue_graph():
    """対話グラフのテスト"""
    print("🧪 pydantic-graph ベース対話状態遷移エンジン - テスト実行")

    engine = DialogueGraphEngine()

    # 対話開始
    result, state = await engine.start_dialogue()
    print(f"✅ 対話開始: {result['question']['text'][:50]}...")

    # uniform フローテスト
    current_node = StartNode()
    result, state = await engine.process_user_input(
        "uniform", "すべての電話に同じ案内をする", state, current_node
    )
    print(
        f"✅ Uniform フロー選択: {result.get('question', {}).get('text', 'N/A')[:50]}..."
    )

    # AI受付選択
    current_node = UniformActionNode()
    result, state = await engine.process_user_input(
        "用件をヒアリングして、テキストで受け取る",
        "用件をヒアリングして、テキストで受け取る",
        state,
        current_node,
    )

    if result.get("node_type") == "end":
        print(f"✅ 完了: {result['message']}")
        print(f"✅ セッション結果: {len(result['final_state']['answers'])}個の回答")

    print("🎉 pydantic-graph 実装テスト完了！")

    return engine, state


async def main():
    """メイン実行"""
    try:
        await test_dialogue_graph()
    except Exception as e:
        print(f"❌ テストエラー: {e}")
        import traceback

        traceback.print_exc()


if __name__ == "__main__":
    asyncio.run(main())