File size: 3,337 Bytes
037ee88 4b7573c 037ee88 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c 0bb062b 4b7573c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | """
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": "<intent_string>"}
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 |