import logging import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from agent.intent_classifier import IntentType, ClassificationResult from database.view_manager import has_database_signals, extract_identifier, IdentifierType, has_view_specific_keywords logger = logging.getLogger(__name__) _GREETING_RE = re.compile( r"^\s*(merhaba|selam|hey|hi|hello|günaydın|iyi\s*günler|iyi\s*akşamlar|teşekkür|sağol|sagol|nasılsın|nasilsin)\s*[.!?]*\s*$", re.IGNORECASE, ) @dataclass class RoutingDecision: execution_mode: str intent: Optional[IntentType] = None confidence: float = 0.0 reasoning: str = "" database_available: bool = False classification_result: Optional[ClassificationResult] = None identifier_type: Optional[IdentifierType] = None identifier_value: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: return { "execution_mode": self.execution_mode, "intent": self.intent.value if self.intent else None, "confidence": round(self.confidence, 3), "reasoning": self.reasoning, "database_available": self.database_available, "identifier_type": self.identifier_type.value if self.identifier_type else None, "identifier_value": self.identifier_value, "metadata": self.metadata, } def route(query: str, database_available: bool = False, **_kwargs) -> RoutingDecision: if _GREETING_RE.match(query): return RoutingDecision( execution_mode="conversational", intent=IntentType.CONVERSATIONAL, confidence=0.95, reasoning="Greeting or small talk detected", database_available=database_available, ) if database_available and has_database_signals(query): id_type, id_value = extract_identifier(query) return RoutingDecision( execution_mode="database", intent=IntentType.DATABASE_QUERY, confidence=0.95, reasoning=f"{id_type.value if id_type else 'unknown'} + domain keyword detected", database_available=True, identifier_type=id_type, identifier_value=id_value, metadata={"credit_file_id": id_value if id_type == IdentifierType.CREDIT_FILE_ID else None, "identifier_type": id_type.value if id_type else None, "identifier_value": id_value}, ) return RoutingDecision( execution_mode="documents", intent=IntentType.DOCUMENT_QUERY, confidence=0.8, reasoning="Default to document retrieval", database_available=database_available, ) class IntentRouter: def __init__(self, confidence_threshold: float = 0.6): self._confidence_threshold = confidence_threshold def route(self, query: str, database_available: bool = False, **kwargs) -> RoutingDecision: return route(query, database_available=database_available, **kwargs) _router_instance: Optional[IntentRouter] = None def get_intent_router(confidence_threshold: float = 0.6) -> IntentRouter: global _router_instance if _router_instance is None: _router_instance = IntentRouter(confidence_threshold=confidence_threshold) return _router_instance def reset_intent_router() -> None: global _router_instance _router_instance = None