Update env.py
Browse files
env.py
CHANGED
|
@@ -2,7 +2,6 @@ import numpy as np
|
|
| 2 |
import gymnasium as gym
|
| 3 |
from gymnasium import spaces
|
| 4 |
|
| 5 |
-
# Labels as requested
|
| 6 |
URGENCY_LABELS = ["General", "Billing", "Security Breach"]
|
| 7 |
ROUTING_LABELS = ["AI Auto-Reply", "Tech Support", "Legal"]
|
| 8 |
RESOLUTION_LABELS = ["Archive", "Draft Reply", "Escalate to Human"]
|
|
@@ -10,8 +9,6 @@ RESOLUTION_LABELS = ["Archive", "Draft Reply", "Escalate to Human"]
|
|
| 10 |
class EmailTriageEnv(gym.Env):
|
| 11 |
def __init__(self, task="all"):
|
| 12 |
super().__init__()
|
| 13 |
-
|
| 14 |
-
# Hardcoded dataset to avoid import errors
|
| 15 |
self.full_dataset = [
|
| 16 |
{"difficulty": "easy", "description": "Spam promo", "correct_actions": (0, 0, 0)},
|
| 17 |
{"difficulty": "easy", "description": "Routine support", "correct_actions": (0, 1, 1)},
|
|
@@ -20,7 +17,7 @@ class EmailTriageEnv(gym.Env):
|
|
| 20 |
{"difficulty": "hard", "description": "IT password reset phish", "correct_actions": (2, 1, 2)},
|
| 21 |
{"difficulty": "hard", "description": "Ransomware threat", "correct_actions": (2, 2, 2)}
|
| 22 |
]
|
| 23 |
-
|
| 24 |
if task != "all":
|
| 25 |
self._queue = [e for e in self.full_dataset if e.get("difficulty") == task]
|
| 26 |
else:
|
|
@@ -31,7 +28,6 @@ class EmailTriageEnv(gym.Env):
|
|
| 31 |
self._step_idx = 0
|
| 32 |
|
| 33 |
def reset(self, seed=None, options=None):
|
| 34 |
-
super().reset(seed=seed)
|
| 35 |
self._step_idx = 0
|
| 36 |
return np.zeros(10, dtype=np.float32), {}
|
| 37 |
|
|
@@ -42,13 +38,11 @@ class EmailTriageEnv(gym.Env):
|
|
| 42 |
email = self._queue[self._step_idx]
|
| 43 |
correct = email["correct_actions"]
|
| 44 |
|
| 45 |
-
# Scoring logic
|
| 46 |
reward = 1.0 if tuple(action) == tuple(correct) else 0.0
|
|
|
|
| 47 |
if correct[0] == 2 and action[0] != 2:
|
| 48 |
-
reward = -2.0
|
| 49 |
|
| 50 |
self._step_idx += 1
|
| 51 |
done = self._step_idx >= len(self._queue)
|
| 52 |
-
|
| 53 |
-
info = {"raw_reward": reward, "correct_actions": correct}
|
| 54 |
-
return np.zeros(10), float(reward), done, False, info
|
|
|
|
| 2 |
import gymnasium as gym
|
| 3 |
from gymnasium import spaces
|
| 4 |
|
|
|
|
| 5 |
URGENCY_LABELS = ["General", "Billing", "Security Breach"]
|
| 6 |
ROUTING_LABELS = ["AI Auto-Reply", "Tech Support", "Legal"]
|
| 7 |
RESOLUTION_LABELS = ["Archive", "Draft Reply", "Escalate to Human"]
|
|
|
|
| 9 |
class EmailTriageEnv(gym.Env):
|
| 10 |
def __init__(self, task="all"):
|
| 11 |
super().__init__()
|
|
|
|
|
|
|
| 12 |
self.full_dataset = [
|
| 13 |
{"difficulty": "easy", "description": "Spam promo", "correct_actions": (0, 0, 0)},
|
| 14 |
{"difficulty": "easy", "description": "Routine support", "correct_actions": (0, 1, 1)},
|
|
|
|
| 17 |
{"difficulty": "hard", "description": "IT password reset phish", "correct_actions": (2, 1, 2)},
|
| 18 |
{"difficulty": "hard", "description": "Ransomware threat", "correct_actions": (2, 2, 2)}
|
| 19 |
]
|
| 20 |
+
|
| 21 |
if task != "all":
|
| 22 |
self._queue = [e for e in self.full_dataset if e.get("difficulty") == task]
|
| 23 |
else:
|
|
|
|
| 28 |
self._step_idx = 0
|
| 29 |
|
| 30 |
def reset(self, seed=None, options=None):
|
|
|
|
| 31 |
self._step_idx = 0
|
| 32 |
return np.zeros(10, dtype=np.float32), {}
|
| 33 |
|
|
|
|
| 38 |
email = self._queue[self._step_idx]
|
| 39 |
correct = email["correct_actions"]
|
| 40 |
|
|
|
|
| 41 |
reward = 1.0 if tuple(action) == tuple(correct) else 0.0
|
| 42 |
+
# Penalty for missing security threats
|
| 43 |
if correct[0] == 2 and action[0] != 2:
|
| 44 |
+
reward = -2.0
|
| 45 |
|
| 46 |
self._step_idx += 1
|
| 47 |
done = self._step_idx >= len(self._queue)
|
| 48 |
+
return np.zeros(10), float(reward), done, False, {"raw_reward": reward}
|
|
|
|
|
|