resume-agent/backend/app/chat_intent_classifier.py

"""Chat intent classifier: LLM primary, legacy keyword rules as degraded fallback.

The LLM only *proposes* an intent from the fixed registry in chat_intents.py;
handler binding stays deterministic in code. RuleBasedChatIntentClassifier mirrors
the legacy keyword routing so the conversation keeps working when the LLM is down.
"""

from __future__ import annotations

import logging
import re
from typing import Any, Protocol, runtime_checkable

from .builder_conversation.predicates import (
    _is_no_information_reply,
    _is_revision_instruction,
    _requested_section,
    _requests_identity_change,
    _requests_new_entry,
)
from .chat_intents import (
    CHAT_INTENT_REGISTRY_VERSION,
    INTENT_DESCRIPTIONS,
    INTENT_FEWSHOTS,
    ChatIntent,
    ChatTurnClassification,
    ExtractedFact,
)
from .llm_services import OpenAICompatibleStructuredClient, log_ai_event
from .settings import Settings

_ENTRY_LABEL_KEYS = ("company", "project_name", "school", "organization", "title", "name", "position", "role")
_CHITCHAT = {"可以", "好的", "好", "谢谢", "感谢", "继续", "没问题", "知道了", "嗯", "ok", "okay"}
_QUESTION_TOKENS = ("?", "？", "吗", "怎么", "如何", "为什么", "哪", "能不能", "可以不可以")
_EDIT_TOKENS = ("修改", "编辑", "调整", "改一下", "改下")


@runtime_checkable
class ChatIntentClassifier(Protocol):
    def classify(self, message: str, *, state_summary: dict[str, Any]) -> ChatTurnClassification: ...


def build_chat_state_summary(profile: dict[str, Any], state: dict[str, Any] | None = None) -> dict[str, Any]:
    """Compact, metadata-only snapshot fed to the classifier (never full resume text)."""
    state = state or {}
    entries: list[dict[str, str]] = []
    resume_content = profile.get("resume_content") or {}
    for section in resume_content.get("sections") or []:
        if not isinstance(section, dict):
            continue
        kind = str(section.get("kind") or "")
        for entry in section.get("items") or []:
            if not isinstance(entry, dict):
                continue
            label = next(
                (str(entry.get(key)).strip() for key in _ENTRY_LABEL_KEYS if str(entry.get(key) or "").strip()),
                "",
            )
            entries.append({"section": kind, "label": label})
    draft = state.get("draft") if isinstance(state.get("draft"), dict) else None
    return {
        "job_type": str(profile.get("job_type") or ""),
        "target_position": str(profile.get("target_position") or ""),
        "confirmed_entries": entries,
        "draft_section": str(draft.get("section") or "") if draft else None,
    }


def _intent_system_prompt() -> str:
    lines = [
        "你是简历对话的意图分类器。根据用户消息与对话状态，从固定意图集合中选择唯一意图。",
        "facts 只能摘录或紧贴改写用户原话中的事实，不得编造用户没说过的内容。",
        f"注册表版本: {CHAT_INTENT_REGISTRY_VERSION}",
        "意图定义:",
    ]
    lines += [f"- {intent.value}: {INTENT_DESCRIPTIONS[intent]}" for intent in ChatIntent]
    lines.append("示例:")
    lines += [f"- 消息: {shot['message']} → {shot['intent'].value}" for shot in INTENT_FEWSHOTS]
    return "\n".join(lines)


_INTENT_SYSTEM_PROMPT = _intent_system_prompt()


class RuleBasedChatIntentClassifier:
    """Legacy keyword routing, kept verbatim as the degraded path."""

    def classify(self, message: str, *, state_summary: dict[str, Any]) -> ChatTurnClassification:
        content = message.strip()
        normalized = re.sub(r"[\s,，。；;！!？?]", "", content.casefold())
        if _is_no_information_reply(content):
            return self._result(ChatIntent.NO_INFO)
        if _is_revision_instruction(content):
            return self._result(ChatIntent.REVISE_PROPOSAL, revision_instruction=content)
        if _requests_identity_change(content):
            return self._result(ChatIntent.EDIT_IDENTITY)
        section = _requested_section(content)
        if section or _requests_new_entry(content):
            return self._result(ChatIntent.NEW_ENTRY, target_section=section)
        if any(token in normalized for token in _EDIT_TOKENS):
            hint = self._entry_hint(normalized, state_summary)
            return self._result(ChatIntent.EDIT_ENTRY, target_entry_hint=hint)
        if any(token in content for token in _QUESTION_TOKENS):
            return self._result(ChatIntent.ASK_QUESTION, user_question=content)
        parts = [part for part in re.split(r"[\s,，。；;！!？?]+", content.casefold()) if part]
        if parts and all(part in _CHITCHAT for part in parts):
            return self._result(ChatIntent.CHITCHAT)
        facts = [ExtractedFact(text=content)] if content else []
        return self._result(ChatIntent.PROVIDE_FACTS, facts=facts)

    @staticmethod
    def _entry_hint(normalized: str, state_summary: dict[str, Any]) -> str | None:
        for entry in state_summary.get("confirmed_entries") or []:
            label = str(entry.get("label") or "").strip()
            squashed = re.sub(r"[\s,，。；;！!？?]", "", label.casefold())
            if squashed and squashed in normalized:
                return label
        return None

    @staticmethod
    def _result(intent: ChatIntent, **fields: Any) -> ChatTurnClassification:
        return ChatTurnClassification(intent=intent, confidence=0.4, reason="rule_keyword", **fields)


class LLMChatIntentClassifier:
    def __init__(self, client: OpenAICompatibleStructuredClient) -> None:
        self._client = client

    def classify(self, message: str, *, state_summary: dict[str, Any]) -> ChatTurnClassification:
        return self._client.complete(
            schema=ChatTurnClassification,
            schema_name="chat_intent_classification",
            system_prompt=_INTENT_SYSTEM_PROMPT,
            payload={
                "message": message,
                "state_summary": state_summary,
                "registry_version": CHAT_INTENT_REGISTRY_VERSION,
            },
        )


class FallbackChatIntentClassifier:
    def __init__(self, primary: ChatIntentClassifier, fallback: ChatIntentClassifier) -> None:
        self.primary = primary
        self.fallback = fallback

    def classify(self, message: str, *, state_summary: dict[str, Any]) -> ChatTurnClassification:
        try:
            return self.primary.classify(message, state_summary=state_summary)
        except Exception as exc:
            log_ai_event(
                "chat_intent_classification_failed",
                level=logging.ERROR,
                reason_code=getattr(exc, "reason_code", type(exc).__name__.lower()[:48]),
                trace_id=getattr(exc, "trace_id", None),
                exception=type(exc).__name__,
            )
            return self.fallback.classify(message, state_summary=state_summary)


def build_chat_intent_classifier(settings: Settings, client: Any | None = None) -> ChatIntentClassifier:
    rules = RuleBasedChatIntentClassifier()
    if not settings.use_openai:
        return rules
    completion = OpenAICompatibleStructuredClient(settings, client)
    return FallbackChatIntentClassifier(LLMChatIntentClassifier(completion), rules)