""" intent_agent.py --------------- Stage 2: Intent Detection Agent for Notiflow Uses ModelRouter (Nova primary → Gemini fallback) to classify the business intent of Hinglish messages. Supported intents: order | payment | credit | return | preparation | other Integration note (backend upgrade): The private _call_model() function now delegates to agent/model_router.py instead of calling bedrock_client directly. All parsing and public API logic is unchanged. """ import json import logging import re from pathlib import Path PROMPT_PATH = Path(__file__).parent.parent / "prompts" / "intent_prompt.txt" VALID_INTENTS = {"order", "payment", "credit", "return", "preparation", "other"} logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Prompt loader # --------------------------------------------------------------------------- def _load_prompt(message: str) -> str: template = PROMPT_PATH.read_text(encoding="utf-8") return template.replace("{message}", message.strip()) # --------------------------------------------------------------------------- # Model inference (now via ModelRouter — Nova primary, Gemini fallback) # --------------------------------------------------------------------------- def _call_model(prompt: str) -> str: """ Route the prompt through ModelRouter. Returns raw text response from whichever model was available. """ from agent.model_router import route raw, model_used = route(prompt, max_tokens=64) logger.info("Intent inference served by: %s", model_used) return raw # --------------------------------------------------------------------------- # Response parser # --------------------------------------------------------------------------- def _parse_intent_response(raw: str) -> dict: cleaned = re.sub(r"```(?:json)?|```", "", raw).strip() try: result = json.loads(cleaned) intent = result.get("intent", "other").lower().strip() except json.JSONDecodeError: match = re.search(r'"intent"\s*:\s*"(\w+)"', cleaned) intent = match.group(1).lower() if match else "other" if intent not in VALID_INTENTS: logger.warning("Unknown intent '%s', defaulting to 'other'", intent) intent = "other" return {"intent": intent} # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def detect_intent(message: str) -> dict: """ Detect the business intent of a Hinglish message. Args: message: Raw business message (Hinglish or English). Returns: {"intent": ""} Examples: >>> detect_intent("bhaiya 2 kilo bhej dena") {'intent': 'order'} >>> detect_intent("rahul ne 15000 bheja") {'intent': 'payment'} """ if not message or not message.strip(): logger.warning("Empty message received, returning 'other'") return {"intent": "other"} logger.info("Detecting intent for: %r", message) prompt = _load_prompt(message) raw = _call_model(prompt) result = _parse_intent_response(raw) logger.info("Detected intent: %s", result["intent"]) return result