292 lines
14 KiB
Python
292 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterable
|
|
|
|
from .invoice_resolver import InvoiceResolver
|
|
from .vas_variation import varied_avulso_items
|
|
|
|
_REGULATORY_RE = re.compile(r"\b(anatel|procon|justi[cç]a|judicial|advogad[oa]|meus direitos|processar)\b", re.I)
|
|
_HUMAN_RE = re.compile(r"\b(atendente|humano|pessoa|operador|supervisor)\b", re.I)
|
|
_YES_RE = re.compile(r"^\s*(sim|s|pode|pode sim|isso|isso mesmo|quero|claro|ok|confirmo|vamos)\b", re.I)
|
|
_NO_RE = re.compile(r"^\s*(n[aã]o|nao|n|negativo|deixa|deixe|prefiro n[aã]o|quero atendente|atendente)\b", re.I)
|
|
_CLOSE_RE = re.compile(r"\b(era s[oó] isso|s[oó] isso mesmo|obrigad[oa].*era isso|entendi.*obrigad[oa]|pode encerrar|pode finalizar)\b", re.I)
|
|
_PLURAL_CONTINUE_RE = re.compile(r"\b(as duas|os dois|ambas|ambos)\b.*\b(pode|seguir|isso|sim|mesmo)\b", re.I)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PolicyDecision:
|
|
route: str | None = None
|
|
intent: str | None = None
|
|
answer: str | None = None
|
|
reason: str | None = None
|
|
patch: dict[str, Any] | None = None
|
|
|
|
|
|
def _walk_dicts(value: Any) -> Iterable[dict[str, Any]]:
|
|
if isinstance(value, dict):
|
|
yield value
|
|
for child in value.values():
|
|
yield from _walk_dicts(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
yield from _walk_dicts(child)
|
|
|
|
|
|
def _first_dict_value(state: dict[str, Any], keys: tuple[str, ...]) -> dict[str, Any] | None:
|
|
for obj in _walk_dicts(state):
|
|
for key in keys:
|
|
value = obj.get(key)
|
|
if isinstance(value, dict) and value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def _explicit_answer(text: str) -> str | None:
|
|
raw = str(text or "").strip()
|
|
if _YES_RE.search(raw):
|
|
return "YES"
|
|
if _NO_RE.search(raw):
|
|
return "NO"
|
|
return None
|
|
|
|
|
|
def _previous_assistant_intent(state: dict[str, Any]) -> str:
|
|
for item in reversed(state.get("history") or []):
|
|
if not isinstance(item, dict) or str(item.get("role") or "") != "assistant":
|
|
continue
|
|
metadata = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
|
|
direct = str(metadata.get("intent") or "").strip()
|
|
if direct:
|
|
return direct
|
|
decision = metadata.get("route_decision")
|
|
if isinstance(decision, dict):
|
|
return str(decision.get("intent") or "").strip()
|
|
return ""
|
|
|
|
|
|
def _has_live_transaction(state: dict[str, Any]) -> bool:
|
|
status = str(state.get("transaction_status") or "").upper()
|
|
if status in {"COLLECTING_PARAMETERS", "AWAITING_CONFIRMATION", "WORKFLOW_PAUSED", "TOOL_RESULT_CLARIFICATION"}:
|
|
return True
|
|
return bool(state.get("active_transaction") or state.get("pending_tool_call") or state.get("pending_domain_workflow"))
|
|
|
|
|
|
def is_regulatory_threat(text: str) -> bool:
|
|
return bool(_REGULATORY_RE.search(str(text or "")))
|
|
|
|
|
|
def _focused_subjects(state: dict[str, Any]) -> list[str]:
|
|
out: list[str] = []
|
|
candidates = [
|
|
state.get("active_transaction"), state.get("last_transaction"),
|
|
state.get("selected_tool_call"), state.get("pending_tool_call"),
|
|
(state.get("context") or {}).get("focused_items") if isinstance(state.get("context"), dict) else None,
|
|
]
|
|
for candidate in candidates:
|
|
if isinstance(candidate, list):
|
|
for item in candidate:
|
|
if isinstance(item, str) and item.strip():
|
|
out.append(item.strip())
|
|
continue
|
|
if not isinstance(candidate, dict):
|
|
continue
|
|
args = candidate.get("arguments") if isinstance(candidate.get("arguments"), dict) else candidate
|
|
for key in ("subject", "item", "service", "name"):
|
|
value = args.get(key) if isinstance(args, dict) else None
|
|
if isinstance(value, str) and value.strip():
|
|
out.append(value.strip())
|
|
values = args.get("items") if isinstance(args, dict) else None
|
|
if isinstance(values, list):
|
|
for item in values:
|
|
if isinstance(item, dict):
|
|
value = item.get("name") or item.get("desc") or item.get("subject")
|
|
if value:
|
|
out.append(str(value).strip())
|
|
return list(dict.fromkeys(x for x in out if x))
|
|
|
|
|
|
def _retention_items(state: dict[str, Any]) -> list[dict[str, Any]]:
|
|
invoice_detail = _first_dict_value(state, ("invoice_detail", "invoiceDetail"))
|
|
if not invoice_detail:
|
|
return []
|
|
invoice_variation = _first_dict_value(state, ("invoice_variation", "invoiceVariation", "billing_analysis"))
|
|
try:
|
|
items = varied_avulso_items(
|
|
invoice_variation=invoice_variation,
|
|
invoice_detail=invoice_detail,
|
|
resolver=InvoiceResolver(),
|
|
by_charge=True,
|
|
)
|
|
except Exception:
|
|
return []
|
|
if not items:
|
|
return []
|
|
return [
|
|
{
|
|
"name": item.canonical_name,
|
|
"msisdn": item.msisdn,
|
|
"value": str(item.value) if item.value is not None else None,
|
|
"charge_date": item.charge_date,
|
|
}
|
|
for item in items if item.msisdn and item.canonical_name
|
|
]
|
|
|
|
|
|
def evaluate(state: dict[str, Any]) -> PolicyDecision | None:
|
|
"""Contas-only conversational policy executed after generic routing.
|
|
|
|
It never performs a side effect. It may only (a) answer/reprompt, (b) enrich
|
|
routing, or (c) prepare explicit domain context consumed by normal framework
|
|
transaction machinery.
|
|
"""
|
|
text = str(state.get("sanitized_input") or state.get("user_text") or "").strip()
|
|
route_decision = state.get("route_decision") if isinstance(state.get("route_decision"), dict) else {}
|
|
method = str(route_decision.get("method") or "")
|
|
intent = str(state.get("intent") or route_decision.get("intent") or "")
|
|
|
|
previous_count = int(state.get("no_match_count") or 0)
|
|
reset_patch: dict[str, Any] = {"no_match_count": 0} if previous_count else {}
|
|
|
|
# Explicit conversational closure is a lifecycle signal, not a new business
|
|
# intent. It must take precedence over router no-match so deterministic closure
|
|
# phrases are not turned into generic fallback responses. Never consume it while
|
|
# a transaction/workflow is still live.
|
|
if _CLOSE_RE.search(text) and not _has_live_transaction(state):
|
|
return PolicyDecision(
|
|
route="end_session", intent="contas_conversation_close",
|
|
reason="explicit_customer_closure",
|
|
patch={**reset_patch, "terminal_status": "resolvido",
|
|
"conversation_terminal_message": "Atendimento encerrado. Obrigado pelo contato."},
|
|
)
|
|
|
|
# The first utterance after a completed operational boundary is a fresh
|
|
# interaction, never a resume of the workflow that just ended. A short
|
|
# re-engagement such as "ah espera" may legitimately have no business intent
|
|
# yet; answer with a neutral invitation instead of treating it as an error.
|
|
if intent == "contas_no_match" and bool(state.get("operational_context_reset")):
|
|
return PolicyDecision(
|
|
route="conversation_policy_response", intent="contas_post_terminal_reentry",
|
|
answer="Claro. Pode falar. Em que posso ajudar agora?",
|
|
reason="fresh_interaction_after_operational_boundary",
|
|
patch={**reset_patch, "no_match_count": 0},
|
|
)
|
|
|
|
# 1) Consecutive incomprehensible utterances. A router fallback is the generic,
|
|
# architecture-native signal that no intent was understood. Any understood turn
|
|
# resets the counter. Default requirement is three consecutive failures.
|
|
if intent == "contas_no_match":
|
|
count = previous_count + 1
|
|
limit = max(1, int(os.getenv("CONTAS_NO_MATCH_MAX_CONSECUTIVE", "3")))
|
|
if count >= limit:
|
|
return PolicyDecision(
|
|
route="end_session", intent="contas_no_match_terminal",
|
|
reason="no_match_limit_reached",
|
|
patch={"no_match_count": count, "terminal_status": "erro_no_match",
|
|
"conversation_terminal_message": "Não consegui compreender sua solicitação após três tentativas. Vou encerrar este atendimento para evitar uma ação incorreta."},
|
|
)
|
|
return PolicyDecision(
|
|
route="conversation_policy_response", intent="contas_no_match_retry",
|
|
answer="Desculpe, não entendi. Poderia repetir de outra forma?",
|
|
reason="no_match_retry", patch={"no_match_count": count},
|
|
)
|
|
|
|
# A short plural continuation after an invoice explanation belongs to that
|
|
# explanation; it must not be mistaken for generic finalization merely because
|
|
# it contains an affirmative such as "pode seguir". The invoice agent keeps
|
|
# the prior evidence/context and can explain the two referenced charges.
|
|
previous_intent = _previous_assistant_intent(state)
|
|
if (
|
|
previous_intent == "contas_invoice_explanation"
|
|
and _PLURAL_CONTINUE_RE.search(text)
|
|
and not _has_live_transaction(state)
|
|
):
|
|
return PolicyDecision(
|
|
route="faturas_agent", intent="contas_invoice_explanation",
|
|
reason="invoice_plural_context_continuation",
|
|
patch={**reset_patch, "mcp_tools": ["invoice_explanation"],
|
|
"_route_metadata": {"contextual_reentry": True, "original_input": text}},
|
|
)
|
|
|
|
# 2) Regulatory/legal wording is not itself an action. Only preserve action
|
|
# semantics when an entity is already concretely in focus in transaction state.
|
|
if is_regulatory_threat(text):
|
|
focused = _focused_subjects(state)
|
|
if focused:
|
|
subject = focused[0]
|
|
return PolicyDecision(
|
|
route="contestacao_agent", intent="contas_vas_cancel",
|
|
reason="regulatory_threat_with_focused_entity",
|
|
patch={**reset_patch, "regulatory_context": {"threat": True, "focused_items": focused},
|
|
"mcp_tools": ["consultar_vas", "cancelar_vas_avulso"],
|
|
"context": {**(state.get("context") or {}), "focused_items": focused, "regulatory_threat": True}},
|
|
)
|
|
# Broad complaint: never invent an item/action from the invoice.
|
|
return PolicyDecision(
|
|
route="suporte_contas_agent", intent="contas_regulatory_complaint",
|
|
reason="regulatory_threat_without_focused_entity",
|
|
patch={**reset_patch, "regulatory_context": {"threat": True, "focused_items": []}, "mcp_tools": []},
|
|
)
|
|
|
|
# 3) Two-step human-retention policy. It is only eligible when the router would
|
|
# hand off AND authoritative invoice evidence proves a varied cancellable VAS.
|
|
retention = state.get("contas_retention") if isinstance(state.get("contas_retention"), dict) else {}
|
|
pending = str(retention.get("pending_stage") or "")
|
|
offered = list(retention.get("stages_offered") or [])
|
|
retained_items = list(retention.get("items") or [])
|
|
|
|
if pending:
|
|
answer_kind = _explicit_answer(text)
|
|
if pending == "explain":
|
|
if answer_kind == "YES":
|
|
return PolicyDecision(
|
|
route="faturas_agent", intent="contas_invoice_explanation",
|
|
reason="human_retention_explanation_accepted",
|
|
patch={**reset_patch, "mcp_tools": ["invoice_explanation"],
|
|
"contas_retention": {"pending_stage": None, "stages_offered": offered, "items": retained_items}},
|
|
)
|
|
if answer_kind == "NO" or _HUMAN_RE.search(text):
|
|
names = ", ".join(str(x.get("name")) for x in retained_items if isinstance(x, dict) and x.get("name"))
|
|
return PolicyDecision(
|
|
route="conversation_policy_response", intent="contas_retention_continue",
|
|
answer=(f"Antes de transferir, posso tratar o cancelamento dos serviços que participaram da variação da conta: {names}. Deseja que eu prossiga?" if names else "Antes de transferir, posso tratar os serviços que participaram da variação da conta. Deseja que eu prossiga?"),
|
|
reason="human_retention_offer_continue",
|
|
patch={**reset_patch, "contas_retention": {"pending_stage": "continue", "stages_offered": list(dict.fromkeys([*offered, "continue"])), "items": retained_items}},
|
|
)
|
|
elif pending == "continue":
|
|
if answer_kind == "YES":
|
|
names = [str(x.get("name")) for x in retained_items if isinstance(x, dict) and x.get("name")]
|
|
context = {**(state.get("context") or {}), "retention_items": retained_items, "focused_items": names}
|
|
# Route through the normal transactional agent. No side effect here.
|
|
semantic_request = "cancelar " + " e ".join(names) if names else text
|
|
return PolicyDecision(
|
|
route="contestacao_agent", intent="contas_vas_cancel",
|
|
reason="human_retention_cancel_accepted",
|
|
patch={**reset_patch, "mcp_tools": ["consultar_vas", "cancelar_vas_avulso"], "context": context,
|
|
"_route_metadata": {"contextual_reentry": True, "original_input": semantic_request,
|
|
"relevant_conversation_context": "Serviços avulsos comprovados na variação: " + ", ".join(names)},
|
|
"contas_retention": {"pending_stage": None, "stages_offered": offered, "items": retained_items}},
|
|
)
|
|
if answer_kind == "NO" or _HUMAN_RE.search(text):
|
|
return PolicyDecision(
|
|
route="human_handoff", intent="human_handoff", reason="human_retention_declined",
|
|
patch={**reset_patch, "contas_retention": {"pending_stage": None, "stages_offered": offered, "items": retained_items}},
|
|
)
|
|
|
|
if state.get("route") == "human_handoff" and "explain" not in offered:
|
|
items = _retention_items(state)
|
|
if items:
|
|
return PolicyDecision(
|
|
route="conversation_policy_response", intent="contas_retention_explain",
|
|
answer="Antes de transferir, identifiquei serviços avulsos que participaram da variação da sua conta. Posso primeiro explicar essa variação para você?",
|
|
reason="human_retention_offer_explain",
|
|
patch={**reset_patch, "human_handoff_requested": False, "session_control": "",
|
|
"contas_retention": {"pending_stage": "explain", "stages_offered": ["explain"], "items": items}},
|
|
)
|
|
|
|
if reset_patch:
|
|
return PolicyDecision(patch=reset_patch)
|
|
return None
|