agentic.email/backend/app/services/handoff_evaluator.py
2026-10-06 02:07:13 +05:30

140 lines
4.9 KiB
Python

import re
ALL_MAIL_LABEL = "All mail"
HANDOFF_LABELS = [ALL_MAIL_LABEL, "Important", "Finance", "Clients", "Projects", "Personal", "Support"]
ALL_TYPES_ID = "all"
EMAIL_TYPE_META = {
"all": {
"title": "All types",
"patterns": "",
},
"utility": {
"title": "Utility",
"patterns": r"account update|alert|notice|security|verification|password|service information|noreply|no-reply",
},
"transactional": {
"title": "Transactional",
"patterns": r"order|payment|invoice|receipt|billing|subscription|shipped|confirmation|refund|approval",
},
"promotional": {
"title": "Promotional",
"patterns": r"offer|deal|discount|promo|sale|limited time|save \d+%",
},
"marketing": {
"title": "Marketing",
"patterns": r"newsletter|campaign|marketing|outreach|webinar|subscribe|unsubscribe",
},
}
def _email_text(email: dict) -> str:
parts = [
email.get("subject", ""),
email.get("snippet", ""),
email.get("body_text", ""),
]
return " ".join(parts).lower()
def infer_labels(email: dict) -> list[str]:
text = _email_text(email)
matched: list[str] = []
if re.search(r"urgent|important|asap|priority", text) or email.get("is_starred"):
matched.append("Important")
if re.search(r"invoice|bill|payment|finance|receipt", text):
matched.append("Finance")
if re.search(r"client|customer", text):
matched.append("Clients")
if re.search(r"project|proposal|contract", text):
matched.append("Projects")
if re.search(r"personal|family", text):
matched.append("Personal")
if re.search(r"support|help desk|ticket|assistance", text):
matched.append("Support")
return matched
def infer_email_types(email: dict) -> list[str]:
text = _email_text(email)
matched: list[str] = []
for key, meta in EMAIL_TYPE_META.items():
if re.search(meta["patterns"], text, flags=re.IGNORECASE):
matched.append(key)
return matched
def instruction_matches(instructions: str, email: dict) -> tuple[bool, str]:
text = _email_text(email)
instructions = (instructions or "").strip()
if not instructions:
return True, "No handoff instructions were set, so this step passes."
keywords = re.findall(r"[a-z]{4,}", instructions.lower())
topic_hits = [word for word in keywords if word in text and len(word) > 4]
if topic_hits:
return True, f"Mentions topics related to your instructions ({', '.join(topic_hits[:3])})."
intent_patterns = [
(r"refund|chargeback", "This email appears to involve a refund or payment issue."),
(r"approval|authorize|sign off", "This email appears to require approval."),
(r"human|speak with|talk to|call me", "The sender appears to be asking for a person."),
(r"not confident|unsure|escalat", "Your instructions mention uncertain situations."),
(r"urgent|asap|immediately", "The message appears urgent."),
]
for pattern, reason in intent_patterns:
if re.search(pattern, instructions, flags=re.IGNORECASE) and re.search(pattern, text, flags=re.IGNORECASE):
return True, reason
if re.search(r"refund|approval|human|speak with|person|payment", instructions, flags=re.IGNORECASE):
if re.search(r"refund|approval|human|speak|person|payment|decision", text, flags=re.IGNORECASE):
return True, "The email situation matches your handoff instructions."
return False, "This email does not clearly match your handoff instructions."
def evaluate_handoff(
*,
email: dict,
labels: list[str],
email_types: list[str],
instructions: str,
) -> dict:
inferred_labels = infer_labels(email)
inferred_types = infer_email_types(email)
all_mail = ALL_MAIL_LABEL in labels
all_types = ALL_TYPES_ID in email_types
selected_labels = [label for label in labels if label != ALL_MAIL_LABEL]
selected_types = [item for item in email_types if item != ALL_TYPES_ID]
label_ok = all_mail or not selected_labels or any(
label in inferred_labels for label in selected_labels
)
matched_labels = (
[ALL_MAIL_LABEL]
if all_mail
else [label for label in selected_labels if label in inferred_labels]
)
type_ok = all_types or not selected_types or any(
item in inferred_types for item in selected_types
)
matched_types = (
[ALL_TYPES_ID]
if all_types
else [item for item in selected_types if item in inferred_types]
)
instruction_ok, instruction_reason = instruction_matches(instructions, email)
would_handoff = label_ok and type_ok and instruction_ok
return {
"label_matched": label_ok,
"matched_labels": matched_labels,
"type_matched": type_ok,
"matched_types": matched_types,
"instruction_matched": instruction_ok,
"instruction_reason": instruction_reason,
"would_handoff": would_handoff,
}