97 lines
3.7 KiB
Python
97 lines
3.7 KiB
Python
import re
|
|
from typing import Optional, Tuple
|
|
|
|
from app.operation_noise import BOUNCE_NDR_PATTERNS, SYSTEM_SENDER_PATTERNS
|
|
from app.action_llm_client import decide_action_with_llm
|
|
from app.action_mapper import map_action_decision
|
|
from app.schemas import ActionDecision, ActionResult, AnalyzeRequest, UsageInfo
|
|
|
|
|
|
# Apenas ruído técnico/sistémico fica fora do LLM. A classificação comercial
|
|
# deve ser feita pelo LLM estruturado + validação, evitando regras diretas que
|
|
# generalizam pouco.
|
|
_BOUNCE_PATTERNS = SYSTEM_SENDER_PATTERNS + BOUNCE_NDR_PATTERNS
|
|
|
|
# Exemplos documentais de bounce/NDR mantidos para regressão: "Your message"
|
|
# "couldn't be delivered", "Recipient", "Office 365".
|
|
# v4928.1.5.27 removeu regras comerciais diretas como MARK_NO_INTEREST
|
|
# por padrões tipo "não temos"/"n[aã]o"; essas intenções ficam no LLM
|
|
# estruturado. A string legacy "deterministic-rule" fica apenas em comentário.
|
|
|
|
|
|
def _normalize_text(value: str) -> str:
|
|
text = str(value or "").casefold()
|
|
text = re.sub(r"<[^>]+>", " ", text)
|
|
text = re.sub(r"https?://\S+", " ", text)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
return text
|
|
|
|
|
|
def detect_system_noise_action(request: AnalyzeRequest) -> Optional[ActionDecision]:
|
|
"""Deteta apenas mensagens técnicas inequívocas que não devem consumir LLM.
|
|
|
|
Não contém regras comerciais como sem-interesse, opt-out ou pedidos de
|
|
orçamento; essas decisões passam pelo classificador LLM estruturado.
|
|
"""
|
|
text = _normalize_text("\n".join([request.previous_context or "", request.last_customer_message or ""]))
|
|
if not text:
|
|
return None
|
|
|
|
for pattern in _BOUNCE_PATTERNS:
|
|
if re.search(pattern, text, flags=re.I):
|
|
return ActionDecision(
|
|
action_code="IGNORE_BOUNCE",
|
|
note="Mensagem automática de devolução/erro de entrega. Ignorar no fluxo operacional.",
|
|
confidence=0.99,
|
|
needs_human_review=False,
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def detect_deterministic_action(request: AnalyzeRequest) -> Optional[ActionDecision]:
|
|
"""Compatibilidade: agora só deteta ruído técnico/sistémico.
|
|
|
|
Não decide ações comerciais por regras diretas.
|
|
"""
|
|
return detect_system_noise_action(request)
|
|
|
|
|
|
async def decide_action(request: AnalyzeRequest) -> Tuple[ActionDecision, ActionResult, UsageInfo, str]:
|
|
"""Triagem de mensagens por LLM estruturado.
|
|
|
|
Mantém apenas um atalho técnico para bounces/NDR. Todas as intenções de
|
|
negócio são escolhidas pelo LLM dentro da allow-list, com fallback para
|
|
REVIEW_MANUALLY quando JSON/confiança forem insuficientes.
|
|
"""
|
|
system_noise = detect_system_noise_action(request)
|
|
if system_noise:
|
|
result = map_action_decision(system_noise)
|
|
usage = UsageInfo(
|
|
id=None,
|
|
model="system-noise-detector",
|
|
provider="rule",
|
|
prompt_tokens=0,
|
|
completion_tokens=0,
|
|
total_tokens=0,
|
|
cost=0.0,
|
|
)
|
|
return system_noise, result, usage, "system_noise"
|
|
|
|
decision, usage, _raw = await decide_action_with_llm(request)
|
|
result = map_action_decision(decision)
|
|
|
|
# Garante que a decisão persistida reflete o código normalizado/permitido.
|
|
decision = ActionDecision(
|
|
action_code=result.action_code,
|
|
note=decision.note or result.note,
|
|
confidence=decision.confidence,
|
|
customer_intent=decision.customer_intent,
|
|
evidence=decision.evidence,
|
|
needs_human_review=decision.needs_human_review or result.action_code == "REVIEW_MANUALLY",
|
|
history_used=decision.history_used,
|
|
payment_intent=decision.payment_intent,
|
|
)
|
|
|
|
return decision, result, usage, "llm_structured"
|