""" Intent routing data models. [AC-AISVC-111~AC-AISVC-125] Data models for hybrid routing. """ import uuid from dataclasses import dataclass, field from typing import Any @dataclass class RuleMatchResult: """ [AC-AISVC-112] Result of rule matching. Contains matched rule and score. """ rule_id: uuid.UUID | None rule: Any | None match_type: str | None matched_text: str | None score: float duration_ms: int def to_dict(self) -> dict[str, Any]: return { "rule_id": str(self.rule_id) if self.rule_id else None, "rule_name": self.rule.name if self.rule else None, "match_type": self.match_type, "matched_text": self.matched_text, "score": self.score, "duration_ms": self.duration_ms, } @dataclass class SemanticCandidate: """ [AC-AISVC-113] Semantic match candidate. """ rule: Any score: float def to_dict(self) -> dict[str, Any]: return { "rule_id": str(self.rule.id), "rule_name": self.rule.name, "score": self.score, } @dataclass class SemanticMatchResult: """ [AC-AISVC-113] Result of semantic matching. """ candidates: list[SemanticCandidate] top_score: float duration_ms: int skipped: bool skip_reason: str | None def to_dict(self) -> dict[str, Any]: return { "top_candidates": [c.to_dict() for c in self.candidates], "top_score": self.top_score, "duration_ms": self.duration_ms, "skipped": self.skipped, "skip_reason": self.skip_reason, } @dataclass class LlmJudgeInput: """ [AC-AISVC-119] Input for LLM judge. """ message: str candidates: list[dict[str, Any]] conflict_type: str @dataclass class LlmJudgeResult: """ [AC-AISVC-119] Result of LLM judge. """ intent_id: str | None intent_name: str | None score: float reasoning: str | None duration_ms: int tokens_used: int triggered: bool def to_dict(self) -> dict[str, Any]: return { "triggered": self.triggered, "intent_id": self.intent_id, "intent_name": self.intent_name, "score": self.score, "reasoning": self.reasoning, "duration_ms": self.duration_ms, "tokens_used": self.tokens_used, } @classmethod def empty(cls) -> "LlmJudgeResult": return cls( intent_id=None, intent_name=None, score=0.0, reasoning=None, duration_ms=0, tokens_used=0, triggered=False, ) @dataclass class FusionConfig: """ [AC-AISVC-116] Fusion configuration. """ w_rule: float = 0.5 w_semantic: float = 0.3 w_llm: float = 0.2 semantic_threshold: float = 0.7 conflict_threshold: float = 0.2 gray_zone_threshold: float = 0.6 min_trigger_threshold: float = 0.3 clarify_threshold: float = 0.4 multi_intent_threshold: float = 0.15 llm_judge_enabled: bool = True semantic_matcher_enabled: bool = True semantic_matcher_timeout_ms: int = 100 llm_judge_timeout_ms: int = 2000 semantic_top_k: int = 3 def to_dict(self) -> dict[str, Any]: return { "w_rule": self.w_rule, "w_semantic": self.w_semantic, "w_llm": self.w_llm, "semantic_threshold": self.semantic_threshold, "conflict_threshold": self.conflict_threshold, "gray_zone_threshold": self.gray_zone_threshold, "min_trigger_threshold": self.min_trigger_threshold, "clarify_threshold": self.clarify_threshold, "multi_intent_threshold": self.multi_intent_threshold, "llm_judge_enabled": self.llm_judge_enabled, "semantic_matcher_enabled": self.semantic_matcher_enabled, "semantic_matcher_timeout_ms": self.semantic_matcher_timeout_ms, "llm_judge_timeout_ms": self.llm_judge_timeout_ms, "semantic_top_k": self.semantic_top_k, } @classmethod def from_dict(cls, data: dict[str, Any]) -> "FusionConfig": return cls( w_rule=data.get("w_rule", 0.5), w_semantic=data.get("w_semantic", 0.3), w_llm=data.get("w_llm", 0.2), semantic_threshold=data.get("semantic_threshold", 0.7), conflict_threshold=data.get("conflict_threshold", 0.2), gray_zone_threshold=data.get("gray_zone_threshold", 0.6), min_trigger_threshold=data.get("min_trigger_threshold", 0.3), clarify_threshold=data.get("clarify_threshold", 0.4), multi_intent_threshold=data.get("multi_intent_threshold", 0.15), llm_judge_enabled=data.get("llm_judge_enabled", True), semantic_matcher_enabled=data.get("semantic_matcher_enabled", True), semantic_matcher_timeout_ms=data.get("semantic_matcher_timeout_ms", 100), llm_judge_timeout_ms=data.get("llm_judge_timeout_ms", 2000), semantic_top_k=data.get("semantic_top_k", 3), ) @dataclass class RouteTrace: """ [AC-AISVC-122] Route trace log. """ rule_match: dict[str, Any] = field(default_factory=dict) semantic_match: dict[str, Any] = field(default_factory=dict) llm_judge: dict[str, Any] = field(default_factory=dict) fusion: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "rule_match": self.rule_match, "semantic_match": self.semantic_match, "llm_judge": self.llm_judge, "fusion": self.fusion, } @dataclass class FusionResult: """ [AC-AISVC-115] Fusion decision result. """ final_intent: Any | None final_confidence: float decision_reason: str need_clarify: bool clarify_candidates: list[Any] | None trace: RouteTrace def to_dict(self) -> dict[str, Any]: return { "final_intent": { "id": str(self.final_intent.id), "name": self.final_intent.name, "response_type": self.final_intent.response_type, } if self.final_intent else None, "final_confidence": self.final_confidence, "decision_reason": self.decision_reason, "need_clarify": self.need_clarify, "clarify_candidates": [ {"id": str(c.id), "name": c.name} for c in (self.clarify_candidates or []) ], "trace": self.trace.to_dict(), } DEFAULT_FUSION_CONFIG = FusionConfig()