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"