vpd-permission-poc/ai-web-agent-console/ai_web_agent_console/questions.py

"""AI Web Agent Console preset catalog와 현재 질문 기반의 결정적 intent router."""

from __future__ import annotations

from dataclasses import dataclass
import re
import unicodedata


QUESTION_CATEGORIES = ("STRUCTURED", "RAG", "HYBRID")
GENERIC_RAG_QUESTION_ID = "R0"
MAX_QUESTION_CHARS = 2_000


@dataclass(frozen=True)
class DemoQuestion:
    """ID와 분류가 고정된 데모 질문."""

    question_id: str
    category: str
    text: str

    def __post_init__(self) -> None:
        if self.category not in QUESTION_CATEGORIES:
            raise ValueError("unsupported demo question category")
        if not self.question_id or not self.text.strip():
            raise ValueError("demo question id/text is required")


@dataclass(frozen=True)
class ResolvedQuestionIntent:
    """현재 질문만으로 결정된 실행 intent.

    ``question_id``는 local route/fixture를 설명하는 분류 label이다. 8500 provider
    입력으로 전달되지 않으며 UI에서 선택한 scenario ID도 이 모델에 들어오지
    않는다.
    """

    question_id: str
    category: str

    def __post_init__(self) -> None:
        if self.category not in QUESTION_CATEGORIES:
            raise ValueError("unsupported resolved question category")
        if self.question_id not in {
            GENERIC_RAG_QUESTION_ID,
            "S1",
            "S2",
            "S3",
            "S4",
            "S5",
            "R1",
            "R2",
            "H1",
            "H2",
            "H3",
            "H4",
        }:
            raise ValueError("unsupported resolved question id")


COMMON_DEMO_QUESTIONS = (
    DemoQuestion("S1", "STRUCTURED", "상품별 계약 건수를 보여줘."),
    DemoQuestion("S2", "STRUCTURED", "총 지급보험금이 가장 큰 상품은?"),
    DemoQuestion(
        "S3",
        "STRUCTURED",
        "숫자로 계산 가능한 평균 보장금액이 높은 상품 10개를 보여줘.",
    ),
    DemoQuestion("S4", "STRUCTURED", "고객 등급별 평균 보험료를 보여줘."),
    DemoQuestion("S5", "STRUCTURED", "이해관계자 역할별 인원 수를 보여줘."),
    DemoQuestion("R1", "RAG", "자동차보험 약관의 면책 사항은?"),
    DemoQuestion("R2", "RAG", "보험금 청구 시 필요한 서류는?"),
    DemoQuestion(
        "H1",
        "HYBRID",
        "보험금이 가장 큰 상품의 주요 면책 조항을 알려줘.",
    ),
    DemoQuestion(
        "H2",
        "HYBRID",
        "청구가 많은 상품군의 보장 제외 조건을 알려줘.",
    ),
    DemoQuestion(
        "H3",
        "HYBRID",
        "고객 등급별 보험료 수준을 보고, 관련 약관상 유의해야 할 보장 제외 조건도 함께 알려줘.",
    ),
)

_QUESTION_BY_ID = {item.question_id: item for item in COMMON_DEMO_QUESTIONS}
if len(_QUESTION_BY_ID) != len(COMMON_DEMO_QUESTIONS):
    raise RuntimeError("duplicate AI Web Agent Console demo question id")


def question_by_id(question_id: str) -> DemoQuestion:
    """정규화된 ID로 질문을 찾되, 알 수 없는 ID는 거부한다."""

    normalized = str(question_id).strip().upper()
    try:
        return _QUESTION_BY_ID[normalized]
    except KeyError:
        raise ValueError("unknown AI Web Agent Console demo question id") from None


def normalize_scenario_id(value: object) -> str | None:
    """선택적인 preset ID를 정규화하되 실행 routing에는 관여하지 않는다."""

    if value is None:
        return None
    if not isinstance(value, str):
        raise ValueError("scenario id must be a string")
    normalized = value.strip().upper()
    if not normalized:
        return None
    question_by_id(normalized)
    return normalized


def _normalized_question(question: object) -> tuple[str, str]:
    if not isinstance(question, str):
        raise ValueError("question must be a non-empty string")
    if len(question) > MAX_QUESTION_CHARS:
        raise ValueError("question exceeds the supported length")
    normalized = re.sub(
        r"\s+", " ", unicodedata.normalize("NFKC", question).strip().lower()
    )
    if not normalized:
        raise ValueError("question must be a non-empty string")
    return normalized, normalized.replace(" ", "")


def _contains_any(text: str, candidates: tuple[str, ...]) -> bool:
    return any(candidate in text for candidate in candidates)


def resolve_question_intent(question: object) -> ResolvedQuestionIntent:
    """현재 질문 텍스트만으로 S/R/H route intent를 결정한다.

    명확한 structured/hybrid intent에 해당하지 않는 질문은 임의로 추측하지 않고
    ``R0`` generic RAG 검색으로 보낸다. 이 함수는 scenario 또는
    question ID hint를 받지 않으므로 preset metadata가 실행을 바꿀 수 없다.
    """

    normalized, compact = _normalized_question(question)

    # Canonical preset은 기존 10문항 동작을 byte-for-byte 보존한다.
    for item in COMMON_DEMO_QUESTIONS:
        candidate, _ = _normalized_question(item.text)
        if normalized == candidate:
            return ResolvedQuestionIntent(item.question_id, item.category)

    has_product = _contains_any(compact, ("상품", "상품군"))
    has_exclusion = _contains_any(
        compact,
        ("면책", "보장제외", "제외조건", "보상하지않", "약관상유의"),
    )
    has_top = _contains_any(
        compact, ("가장큰", "최대", "최고", "1위", "상위", "제일많", "높은")
    )

    # 자유 질문에서 자동차보험 보유 규모와 타사 대비 강점을 함께 요구하면
    # generic structured 조회와 자유 evidence 검색을 로컬에서 합성하는 H4로
    # 보낸다. Label은 routing metadata일 뿐 provider query ID가 아니다.
    has_auto_insurance = "자동차보험" in compact
    has_count = _contains_any(
        compact,
        (
            "갯수",
            "개수",
            "건수",
            "계약수",
            "몇개",
            "몇건",
            "보유수",
            "상품수",
        ),
    )
    has_competitor_comparison = _contains_any(
        compact, ("타사", "경쟁사", "다른회사", "타보험사")
    ) and _contains_any(compact, ("강점", "장점", "차별", "우위", "비교"))
    if has_auto_insurance and has_count and has_competitor_comparison:
        return ResolvedQuestionIntent("H4", "HYBRID")

    # Hybrid를 먼저 판별해 정형 키워드가 포함된 복합 질문이 S/R 단일 route로
    # 축소되지 않도록 한다.
    if (
        _contains_any(compact, ("고객등급", "등급별"))
        and "보험료" in compact
        and has_exclusion
    ):
        return ResolvedQuestionIntent("H3", "HYBRID")
    if (
        "청구" in compact
        and _contains_any(compact, ("많은", "빈도", "건수", "상위"))
        and has_product
        and has_exclusion
    ):
        return ResolvedQuestionIntent("H2", "HYBRID")
    if (
        has_product
        and _contains_any(compact, ("지급보험금", "보험금"))
        and has_top
        and has_exclusion
    ):
        return ResolvedQuestionIntent("H1", "HYBRID")

    if (
        has_product
        and "계약" in compact
        and _contains_any(compact, ("건수", "계약수", "몇건", "집계"))
    ):
        return ResolvedQuestionIntent("S1", "STRUCTURED")
    if (
        has_product
        and _contains_any(compact, ("지급보험금", "보험금총액", "총보험금"))
        and has_top
    ):
        return ResolvedQuestionIntent("S2", "STRUCTURED")
    if (
        has_product
        and _contains_any(compact, ("보장금액", "가입금액"))
        and "평균" in compact
        and has_top
    ):
        return ResolvedQuestionIntent("S3", "STRUCTURED")
    if (
        _contains_any(compact, ("고객등급", "등급별"))
        and "보험료" in compact
        and "평균" in compact
    ):
        return ResolvedQuestionIntent("S4", "STRUCTURED")
    if (
        "이해관계자" in compact
        and "역할" in compact
        and _contains_any(compact, ("인원", "사람수", "몇명", "명수", "수"))
    ):
        return ResolvedQuestionIntent("S5", "STRUCTURED")

    if (
        "자동차보험" in compact
        and _contains_any(compact, ("면책", "보상하지않", "보장제외", "제외사항"))
    ):
        return ResolvedQuestionIntent("R1", "RAG")
    if (
        _contains_any(compact, ("보험금", "청구"))
        and _contains_any(compact, ("서류", "문서", "증빙", "제출자료"))
    ):
        return ResolvedQuestionIntent("R2", "RAG")

    return ResolvedQuestionIntent(GENERIC_RAG_QUESTION_ID, "RAG")


__all__ = [
    "COMMON_DEMO_QUESTIONS",
    "DemoQuestion",
    "GENERIC_RAG_QUESTION_ID",
    "MAX_QUESTION_CHARS",
    "QUESTION_CATEGORIES",
    "ResolvedQuestionIntent",
    "normalize_scenario_id",
    "question_by_id",
    "resolve_question_intent",
]