Spaces:
Sleeping
Sleeping
File size: 3,427 Bytes
4f129c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """Reusable policy helpers for local baselines and training examples."""
from __future__ import annotations
from models import SupportDeskAction, SupportDeskObservation
from tasks import get_task
def default_reply(task_id: str) -> str:
"""Return a task-specific high-signal customer reply."""
if task_id == "billing_refund_easy":
return (
"Thanks for flagging the duplicate charge. I have started the refund for the extra "
"charge, and the funds usually appear within 5-7 business days."
)
if task_id == "account_takeover_medium":
return (
"We have escalated this to our trust team. Please reset your password, scan your "
"device for malware, and reply with your workspace_id, last successful login time, "
"and billing email so we can verify the account safely."
)
if task_id == "regulated_export_exception_hard":
return (
"We cannot provide a bypass or temporary unlock yet. Our compliance team is running "
"a compliance review, and we need your tenant_region, dpa_amendment_id, and "
"legal_contact_email to continue that review."
)
return (
"We are treating this as an active incident and our on-call engineering team is engaged. "
"Please send the affected request IDs, UTC timestamps, and the impacted region so we can "
"speed up the investigation."
)
def default_note(task_id: str) -> str:
"""Return a task-specific internal note."""
if task_id == "billing_refund_easy":
return "Duplicate charge confirmed from attached invoice; refund approved."
if task_id == "account_takeover_medium":
return "Suspicious login alert reported and customer is locked out."
if task_id == "regulated_export_exception_hard":
return (
"Audit-driven export exception request tied to an EU residency policy block; "
"customer asked for a manual bypass before legal approval."
)
return "EU data residency rollout hit intermittent HTTP 500s and the customer launches tonight."
def heuristic_action(observation: SupportDeskObservation) -> SupportDeskAction:
"""Deterministic high-performing policy used by the baseline."""
task = get_task(observation.task_id)
case = observation.case
if case.queue is None or case.priority is None or case.issue_type is None:
return SupportDeskAction(
operation="classify",
queue=task.gold_queue,
priority=task.gold_priority,
issue_type=task.gold_issue_type,
)
if task.required_requested_fields and sorted(case.requested_fields) != sorted(task.required_requested_fields):
return SupportDeskAction(
operation="request_info",
requested_fields=list(task.required_requested_fields),
)
if case.customer_follow_up.status == "pending":
return SupportDeskAction(operation="wait")
if not case.reply:
return SupportDeskAction(operation="draft_reply", reply=default_reply(observation.task_id))
if not case.internal_note:
return SupportDeskAction(operation="add_internal_note", internal_note=default_note(observation.task_id))
return SupportDeskAction(
operation="submit",
status=task.gold_status,
resolution_code=task.gold_resolution_code,
)
|