Spaces:
Sleeping
Sleeping
File size: 1,522 Bytes
1f213fe 410e78d 1f213fe | 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 | from typing import Any, Dict
from server.tasks.base_task import BaseTask
class DepUpdateTask(BaseTask):
def grade(self, final_state: Dict[str, Any], org_config: Dict[str, Any], scenario: Dict[str, Any]) -> float:
score = 0.0
expected = scenario["expected"]
msgs = final_state["chat"]["messages_posted_this_episode"]
tickets = final_state["ticketing"]["tickets"]
new_tickets = [t for t in tickets if t.get("created_this_episode")]
expected_channels = set(expected.get("channels", []))
expected_teams = set(expected.get("teams_to_notify", []))
notified = {m.get("channel") for m in msgs}
correct_notified = notified & expected_channels
wrong_notified = notified - expected_channels
if correct_notified:
score += 0.25 * (len(correct_notified) / max(len(expected_channels), 1))
assigned_teams = {t.get("assigned_team") for t in new_tickets if t.get("assigned_team")}
correct_teams = assigned_teams & expected_teams
if correct_teams:
score += 0.25 * (len(correct_teams) / max(len(expected_teams), 1))
if new_tickets:
score += 0.25
policy = expected.get("escalation_policy", "post-channel")
if policy == "post-channel" and correct_notified:
score += 0.25
elif policy in ("dm-manager", "page-oncall") and msgs:
score += 0.25
score -= 0.05 * len(wrong_notified)
return max(0.0, min(1.0, score))
|