File size: 5,119 Bytes
c863b63 | 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
Structured call-context engine.
Kept fully SEPARATE from the neural model: it consumes only business/context
fields (caller ID, transaction, account) and produces a deterministic
context_risk (0-100) plus explainable signals. It never touches audio.
HEURISTIC ONLY: the risk_category mapping is a demo rule set, not a validated
model. Nothing here is claimed to be scientifically validated.
"""
import threading
import uuid
import time
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional
from ..config import settings
TRANSACTION_TYPES = ["general_transfer", "high_value_transfer", "account_admin", "credit_card", "other"]
RISK_CATEGORIES = ["low", "medium", "high"]
@dataclass
class CallContext:
call_id: str = ""
caller_id: str = ""
known_contact: Optional[bool] = None
transaction_type: str = "general_transfer"
transaction_amount: float = 0.0
risk_category: str = "low"
account_context: str = ""
note: str = ""
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
class ContextEngine:
"""Per-connection call context store + heuristic context risk."""
def __init__(self):
self._lock = threading.RLock()
self._contexts: Dict[str, CallContext] = {}
def create_context(self, connection_id: str) -> CallContext:
ctx = CallContext(
call_id=str(uuid.uuid4()),
caller_id="",
known_contact=None,
transaction_type="general_transfer",
transaction_amount=0.0,
risk_category="low",
)
with self._lock:
self._contexts[connection_id] = ctx
return ctx
def get(self, connection_id: str) -> Optional[CallContext]:
with self._lock:
return self._contexts.get(connection_id)
def get_or_create(self, connection_id: str) -> CallContext:
ctx = self.get(connection_id)
if ctx is None:
ctx = self.create_context(connection_id)
return ctx
def update(
self,
connection_id: str,
caller_id: Optional[str] = None,
known_contact: Optional[bool] = None,
transaction_type: Optional[str] = None,
transaction_amount: Optional[float] = None,
risk_category: Optional[str] = None,
account_context: Optional[str] = None,
note: Optional[str] = None,
) -> CallContext:
ctx = self.get_or_create(connection_id)
if caller_id is not None:
ctx.caller_id = str(caller_id)
if known_contact is not None:
ctx.known_contact = bool(known_contact)
if transaction_type is not None:
ctx.transaction_type = transaction_type if transaction_type in TRANSACTION_TYPES else "other"
if transaction_amount is not None:
ctx.transaction_amount = float(transaction_amount)
if risk_category is not None:
ctx.risk_category = risk_category if risk_category in RISK_CATEGORIES else "low"
if account_context is not None:
ctx.account_context = str(account_context)
if note is not None:
ctx.note = str(note)
return ctx
def get_or_create_with_callid(self, connection_id: str, call_id: str) -> CallContext:
ctx = self.get_or_create(connection_id)
if call_id:
ctx.call_id = str(call_id)
return ctx
def context_risk(self, connection_id: str) -> Dict[str, Any]:
"""Deterministic, explainable context risk (0-100) + signals."""
ctx = self.get_or_create(connection_id)
signals: List[str] = []
base = {"low": 5.0, "medium": 40.0, "high": 80.0}.get(ctx.risk_category, 5.0)
amount_risk = 0.0
if ctx.transaction_amount > 0:
amount_risk = min(50.0, 100.0 * ctx.transaction_amount / max(settings.context_high_value_threshold, 1e-6))
if ctx.transaction_amount > settings.context_high_value_threshold:
signals.append("HIGH_VALUE_TRANSACTION")
contact_risk = 0.0
if ctx.known_contact is False:
contact_risk = float(settings.context_unknown_contact_risk)
signals.append("UNKNOWN_CALLER")
elif ctx.known_contact is True:
signals.append("KNOWN_CALLER")
if ctx.transaction_type in ("account_admin", "credit_card"):
base = min(100.0, base + 10.0)
signals.append(f"SENSITIVE_TRANSACTION_TYPE:{ctx.transaction_type}")
raw = base + 0.5 * amount_risk + contact_risk
risk = min(100.0, raw)
return {
"context_risk": round(risk, 2),
"signals": signals,
"context": ctx.to_dict(),
}
def remove(self, connection_id: str) -> None:
with self._lock:
self._contexts.pop(connection_id, None)
def reset(self, connection_id: str) -> CallContext:
with self._lock:
return self.create_context(connection_id)
def list_contexts(self) -> Dict[str, CallContext]:
with self._lock:
return dict(self._contexts) |