| """Human handoff detection and notification.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from typing import Any |
|
|
| import httpx |
|
|
| from app.config import get_settings |
|
|
| logger = logging.getLogger(__name__) |
|
|
| SENSITIVE_HINTS = ( |
| "complaint", |
| "réclamation", |
| "reclamation", |
| "litige", |
| "lawsuit", |
| "lawyer", |
| "avocat", |
| "refund", |
| "remboursement", |
| "scam", |
| "fraud", |
| "мошен", |
| "жалоб", |
| "претенз", |
| "возврат", |
| "суд", |
| ) |
|
|
| HANDOFF_PHRASES = ( |
| "human", |
| "agent", |
| "operator", |
| "speak to someone", |
| "real person", |
| "humain", |
| "conseiller", |
| "parler à quelqu'un", |
| "оператор", |
| "человек", |
| "живой", |
| "специалист", |
| ) |
|
|
|
|
| def should_handoff( |
| *, |
| message: str, |
| failure_streak: int, |
| grounded: bool, |
| ) -> tuple[bool, str]: |
| """Decide whether to escalate to a human.""" |
| lowered = message.lower() |
| if any(p in lowered for p in HANDOFF_PHRASES): |
| return True, "explicit_request" |
| if any(p in lowered for p in SENSITIVE_HINTS): |
| return True, "sensitive_topic" |
| if not grounded and failure_streak >= 2: |
| return True, "consecutive_failures" |
| return False, "" |
|
|
|
|
| async def notify_handoff( |
| *, |
| conversation_id: int, |
| reason: str, |
| history: list[dict[str, str]], |
| lead: dict[str, Any], |
| ) -> None: |
| """Send handoff payload to webhook if configured.""" |
| settings = get_settings() |
| payload = { |
| "conversation_id": conversation_id, |
| "reason": reason, |
| "lead": lead, |
| "history": history[-20:], |
| "app": settings.app_name, |
| } |
| if not settings.handoff_webhook_url: |
| logger.info( |
| "Handoff (no webhook): conversation=%s reason=%s", |
| conversation_id, |
| reason, |
| ) |
| return |
| try: |
| async with httpx.AsyncClient(timeout=10) as client: |
| await client.post( |
| settings.handoff_webhook_url, json=payload |
| ) |
| except Exception as exc: |
| logger.exception("Handoff webhook failed: %s", exc) |
|
|
|
|
| def handoff_message(language: str, reason: str) -> str: |
| """User-facing handoff acknowledgement.""" |
| messages = { |
| "en": ( |
| "I am transferring you to a human teammate. " |
| "They will see this conversation and follow up " |
| f"shortly. (reason: {reason.replace('_', ' ')})" |
| ), |
| "fr": ( |
| "Je vous transfère à un conseiller humain. " |
| "Il verra cet échange et vous répondra bientôt. " |
| f"(motif : {reason.replace('_', ' ')})" |
| ), |
| "ru": ( |
| "Передаю вас живому специалисту. Он увидит " |
| "этот диалог и скоро свяжется с вами. " |
| f"(причина: {reason.replace('_', ' ')})" |
| ), |
| } |
| return messages.get(language, messages["en"]) |
|
|
