171 lines
4.7 KiB
Python
171 lines
4.7 KiB
Python
"""
|
|
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(f"[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
|