Spaces:
Sleeping
Sleeping
File size: 5,654 Bytes
2ade2c6 | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """Task-specific graders for the SupportDesk environment."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from models import SupportCaseProgress, SupportDeskObservation
def _as_case(obj) -> SupportCaseProgress:
"""Normalize observation/state/case to SupportCaseProgress."""
if isinstance(obj, SupportCaseProgress):
return obj
if hasattr(obj, "case"):
return obj.case # type: ignore[attr-defined]
raise TypeError(f"Unsupported object for grading: {type(obj)}")
@dataclass
class GradeBreakdown:
score: float
message: str
penalties: dict[str, float]
completed_milestones: list[str] = None
@property
def total_score(self) -> float:
return self.score
def __post_init__(self):
if self.completed_milestones is None:
object.__setattr__(self, "completed_milestones", [])
def _clamp(v: float) -> float:
return max(0.01, min(0.99, v))
def grade_task_id(task_id: str, observation: SupportDeskObservation | SupportCaseProgress) -> GradeBreakdown:
case = _as_case(observation)
if task_id == "billing_refund_easy":
return BillingRefundEasyGrader().score(case)
if task_id == "account_takeover_medium":
return AccountTakeoverMediumGrader().score(case)
if task_id == "api_incident_hard":
return ApiIncidentHardGrader().score(case)
if task_id == "regulated_export_exception_hard":
return RegulatedExportExceptionHardGrader().score(case)
return GradeBreakdown(0.01, "Unknown task", {"unknown_task": 1.0})
def grade_case(task_or_id, observation) -> GradeBreakdown:
"""Return a GradeBreakdown for the given task and case/observation."""
task_id = task_or_id.task_id if hasattr(task_or_id, "task_id") else str(task_or_id)
case = _as_case(observation)
return grade_task_id(task_id, case)
class BillingRefundEasyGrader:
def score(self, case: SupportCaseProgress) -> GradeBreakdown:
penalties: dict[str, float] = {}
score = 1.0
reply = (case.reply or "").lower()
note = (case.internal_note or "").lower()
if reply:
if "refund" not in reply:
penalties["missing_refund"] = 0.25
else:
penalties["no_reply"] = 0.55
if note:
if "duplicate" not in note:
penalties["note_missing_duplicate"] = 0.2
else:
penalties["no_note"] = 0.2
if case.status != "resolved":
penalties["status_not_resolved"] = 0.1
score -= sum(penalties.values())
score = round(score, 2)
return GradeBreakdown(_clamp(score), "Billing refund evaluation", penalties)
def grade(self, case: SupportCaseProgress) -> float:
return self.score(case).score
class AccountTakeoverMediumGrader:
def score(self, case: SupportCaseProgress) -> GradeBreakdown:
penalties: dict[str, float] = {}
score = 0.2
reply = (case.reply or "").lower()
if reply:
if "lock" not in reply:
penalties["missing_lock"] = 0.2
if "verify" not in reply:
penalties["missing_verify"] = 0.2
if "ownership" not in reply:
penalties["missing_ownership"] = 0.2
else:
penalties["no_reply"] = 0.4
if case.status not in ("escalated", "waiting_on_customer"):
penalties["wrong_status"] = 0.2
score -= sum(penalties.values())
score = round(score, 2)
return GradeBreakdown(_clamp(score), "Account takeover evaluation", penalties)
def grade(self, case: SupportCaseProgress) -> float:
return self.score(case).score
class ApiIncidentHardGrader:
def score(self, case: SupportCaseProgress) -> GradeBreakdown:
penalties: dict[str, float] = {}
score = 0.2
reply = (case.reply or "").lower()
if reply:
if "status" not in reply:
penalties["missing_status_page"] = 0.15
if "request" not in reply or "id" not in reply:
penalties["missing_request_ids"] = 0.2
if "escalat" not in reply:
penalties["missing_escalation"] = 0.2
else:
penalties["no_reply"] = 0.4
if case.queue != "platform_engineering":
penalties["wrong_queue"] = 0.15
score -= sum(penalties.values())
score = round(score, 2)
return GradeBreakdown(_clamp(score), "API incident evaluation", penalties)
def grade(self, case: SupportCaseProgress) -> float:
return self.score(case).score
class RegulatedExportExceptionHardGrader:
def score(self, case: SupportCaseProgress) -> GradeBreakdown:
penalties: dict[str, float] = {}
score = 0.2
reply = (case.reply or "").lower()
if reply:
if "compliance" not in reply:
penalties["missing_compliance"] = 0.2
if "cannot promise" not in reply and "not promise" not in reply:
penalties["missing_no_promise"] = 0.2
if "recipient" not in reply or "identity" not in reply:
penalties["missing_recipient"] = 0.15
else:
penalties["no_reply"] = 0.4
if case.status != "waiting_on_customer":
penalties["wrong_status"] = 0.15
score -= sum(penalties.values())
score = round(score, 2)
return GradeBreakdown(_clamp(score), "Regulated export evaluation", penalties)
def grade(self, case: SupportCaseProgress) -> float:
return self.score(case).score
|