Spaces:
Runtime error
Runtime error
| """ | |
| NextActionPredictor inference helper. | |
| Exposes recommend_action() used by the CMA tool recommend_action. | |
| Falls back to DOMAIN_ACTION_PRIORS rule-based routing when no checkpoint exists. | |
| The module-level NextActionPredictor instance is cached after the first call. | |
| Call init_predictor() at server startup for a predictable load moment. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from typing import Optional | |
| from src.next_action.model import ( | |
| EscalationAction, | |
| NextActionPredictor, | |
| build_feature_vector, | |
| entities_dict_to_flags, | |
| ) | |
| logger = logging.getLogger(__name__) | |
| _DEFAULT_MODEL_PATH = "models/next_action/model.pt" | |
| _predictor: Optional[NextActionPredictor] = None | |
| # --------------------------------------------------------------------------- | |
| # Lifecycle helpers | |
| # --------------------------------------------------------------------------- | |
| def init_predictor(model_path: str = _DEFAULT_MODEL_PATH) -> NextActionPredictor: | |
| """ | |
| Explicitly initialise (or reload) the module-level NextActionPredictor. | |
| Always succeeds: if the checkpoint is missing, the rule-based fallback is | |
| used and a warning is logged. Call this at server startup so the load | |
| happens at a known moment rather than on the first request. | |
| """ | |
| global _predictor | |
| _predictor = NextActionPredictor(model_path) | |
| if _predictor.uses_fallback: | |
| logger.warning( | |
| "NextActionPredictor is using rule-based fallback (no checkpoint at '%s'). " | |
| "Train with: python -m src.next_action.train --output_path %s", | |
| model_path, model_path, | |
| ) | |
| return _predictor | |
| # --------------------------------------------------------------------------- | |
| # Public API | |
| # --------------------------------------------------------------------------- | |
| def recommend_action( | |
| domain: str, | |
| entities: dict, | |
| prior_contact: bool = False, | |
| model_path: str = _DEFAULT_MODEL_PATH, | |
| ) -> list[EscalationAction]: | |
| """ | |
| Return a ranked list of EscalationAction for *domain* and *entities*. | |
| Args: | |
| domain: The classified complaint domain (one of the 6 G.U.I.D.E. domains). | |
| entities: Dict of entity_type → value from EvidenceNER, e.g. | |
| {"ORG": "Flipkart", "AMOUNT": "₹4,299", "REF_ID": "OD-123"}. | |
| PERSON is ignored (role reference, not a routing signal). | |
| prior_contact: True if the user has already contacted the company without | |
| resolution. Causes company_support to be deprioritised. | |
| model_path: Path to the .pt checkpoint; falls back to rule-based if absent. | |
| Returns: | |
| All 6 EscalationActions sorted by confidence descending. | |
| The caller (CMA) typically surfaces the top 2–3 to the user. | |
| """ | |
| global _predictor | |
| if _predictor is None: | |
| init_predictor(model_path) | |
| entity_flags = entities_dict_to_flags(entities) | |
| fv = build_feature_vector(domain, entity_flags, float(prior_contact)) | |
| return _predictor.predict(fv) | |