ai-robot-core/ai-service/app/services/intent/router.py

"""
Intent router for AI Service.
[AC-AISVC-69, AC-AISVC-70] Intent matching engine with keyword and regex support.
"""

import logging
import re
from dataclasses import dataclass
from typing import Any

from app.models.entities import IntentRule

logger = logging.getLogger(__name__)


@dataclass
class IntentMatchResult:
    """
    [AC-AISVC-69] Result of intent matching.
    Contains the matched rule and match details.
    """

    rule: IntentRule
    match_type: str
    matched: str

    def to_dict(self) -> dict[str, Any]:
        return {
            "rule_id": str(self.rule.id),
            "rule_name": self.rule.name,
            "match_type": self.match_type,
            "matched": self.matched,
            "response_type": self.rule.response_type,
            "target_kb_ids": self.rule.target_kb_ids or [],
            "flow_id": str(self.rule.flow_id) if self.rule.flow_id else None,
            "fixed_reply": self.rule.fixed_reply,
            "transfer_message": self.rule.transfer_message,
        }


class IntentRouter:
    """
    [AC-AISVC-69] Intent matching engine.

    Matching algorithm:
    1. Load rules ordered by priority DESC
    2. For each rule, try keyword matching first
    3. If no keyword match, try regex pattern matching
    4. Return first match (highest priority)
    5. If no match, return None (fallback to default RAG)
    """

    def __init__(self):
        pass

    def match(
        self,
        message: str,
        rules: list[IntentRule],
    ) -> IntentMatchResult | None:
        """
        [AC-AISVC-69] Match user message against intent rules.

        Args:
            message: User input message
            rules: List of enabled rules ordered by priority DESC

        Returns:
            IntentMatchResult if matched, None otherwise
        """
        if not message or not rules:
            return None

        message_lower = message.lower()

        for rule in rules:
            if not rule.is_enabled:
                continue

            keyword_result = self._match_keywords(message, message_lower, rule)
            if keyword_result:
                logger.info(
                    f"[AC-AISVC-69] Intent matched by keyword: "
                    f"rule={rule.name}, matched='{keyword_result.matched}'"
                )
                return keyword_result

            regex_result = self._match_patterns(message, rule)
            if regex_result:
                logger.info(
                    f"[AC-AISVC-69] Intent matched by regex: "
                    f"rule={rule.name}, matched='{regex_result.matched}'"
                )
                return regex_result

        logger.debug("[AC-AISVC-70] No intent matched, will fallback to default RAG")
        return None

    def _match_keywords(
        self,
        message: str,
        message_lower: str,
        rule: IntentRule,
    ) -> IntentMatchResult | None:
        """
        Match message against rule keywords.
        Any keyword match returns a result.
        """
        keywords = rule.keywords or []
        if not keywords:
            return None

        for keyword in keywords:
            if not keyword:
                continue
            if keyword.lower() in message_lower:
                return IntentMatchResult(
                    rule=rule,
                    match_type="keyword",
                    matched=keyword,
                )

        return None

    def _match_patterns(
        self,
        message: str,
        rule: IntentRule,
    ) -> IntentMatchResult | None:
        """
        Match message against rule regex patterns.
        Any pattern match returns a result.
        """
        patterns = rule.patterns or []
        if not patterns:
            return None

        for pattern in patterns:
            if not pattern:
                continue
            try:
                if re.search(pattern, message, re.IGNORECASE):
                    return IntentMatchResult(
                        rule=rule,
                        match_type="regex",
                        matched=pattern,
                    )
            except re.error as e:
                logger.warning(
                    f"Invalid regex pattern in rule {rule.id}: {pattern}, error: {e}"
                )
                continue

        return None

    def match_with_stats(
        self,
        message: str,
        rules: list[IntentRule],
    ) -> tuple[IntentMatchResult | None, str | None]:
        """
        [AC-AISVC-69] Match and return rule_id for statistics update.

        Returns:
            Tuple of (match_result, rule_id_for_stats)
        """
        result = self.match(message, rules)
        if result:
            return result, str(result.rule.id)
        return None, None