Update env.py

env.py

@@ -1,48 +1,71 @@
-import numpy as np
-import gymnasium as gym
-from gymnasium import spaces
-
-URGENCY_LABELS = ["General", "Billing", "Security Breach"]
-ROUTING_LABELS = ["AI Auto-Reply", "Tech Support", "Legal"]
-RESOLUTION_LABELS = ["Archive", "Draft Reply", "Escalate to Human"]
-
-class EmailTriageEnv(gym.Env):
-    def __init__(self, task="all"):
-        super().__init__()
-        self.full_dataset = [
-            {"difficulty": "easy", "description": "Spam promo", "correct_actions": (0, 0, 0)},
-            {"difficulty": "easy", "description": "Routine support", "correct_actions": (0, 1, 1)},
-            {"difficulty": "medium", "description": "Billing dispute", "correct_actions": (1, 2, 2)},
-            {"difficulty": "medium", "description": "Refund request", "correct_actions": (1, 2, 2)},
-            {"difficulty": "hard", "description": "IT password reset phish", "correct_actions": (2, 1, 2)},
-            {"difficulty": "hard", "description": "Ransomware threat", "correct_actions": (2, 2, 2)}
+from typing import List, Dict, Any, Tuple
+import random
+
+URGENCY_LABELS = ["low", "medium", "high"]
+ROUTING_LABELS = ["general", "support", "security"]
+RESOLUTION_LABELS = ["ignore", "respond", "escalate"]
+
+
+class EmailTriageEnv:
+    def __init__(self, task: str = "easy"):
+        self.task = task
+        self._queue: List[Dict] = []
+        self._index = 0
+        self._done = False
+
+    def _generate_emails(self) -> List[Dict]:
+        emails = [
+            {"description": "Password reset not working", "label": [2, 1, 2]},
+            {"description": "Billing refund request", "label": [1, 2, 2]},
+            {"description": "App is slow and buggy", "label": [0, 1, 1]},
         ]
-        
-        if task != "all":
-            self._queue = [e for e in self.full_dataset if e.get("difficulty") == task]
-        else:
-            self._queue = self.full_dataset
-            
-        self.action_space = spaces.MultiDiscrete([3, 3, 3])
-        self.observation_space = spaces.Box(low=0.0, high=1.0, shape=(10,), dtype=np.float32)
-        self._step_idx = 0
-
-    def reset(self, seed=None, options=None):
-        self._step_idx = 0
-        return np.zeros(10, dtype=np.float32), {}
-
-    def step(self, action):
-        if self._step_idx >= len(self._queue):
-            return np.zeros(10), 0.0, True, False, {}
-            
-        email = self._queue[self._step_idx]
-        correct = email["correct_actions"]
-        
-        reward = 1.0 if tuple(action) == tuple(correct) else 0.0
-        # Penalty for missing security threats
-        if correct[0] == 2 and action[0] != 2:
-            reward = -2.0
-
-        self._step_idx += 1
-        done = self._step_idx >= len(self._queue)
-        return np.zeros(10), float(reward), done, False, {"raw_reward": reward}
+
+        if self.task in ["medium", "hard"]:
+            emails += [
+                {"description": "Possible phishing attempt detected", "label": [2, 2, 2]},
+                {"description": "Invoice mismatch and payment issue", "label": [1, 2, 2]},
+            ]
+
+        if self.task == "hard":
+            emails += [
+                {"description": "Ransomware attack suspected on system", "label": [2, 2, 2]},
+                {"description": "User reports data breach and performance issues", "label": [2, 2, 2]},
+            ]
+
+        random.shuffle(emails)
+        return emails
+
+    def reset(self) -> Dict[str, Any]:
+        self._queue = self._generate_emails()
+        self._index = 0
+        self._done = False
+        return self.state()
+
+    def state(self) -> Dict[str, Any]:
+        if self._done:
+            return {"done": True}
+
+        current = self._queue[self._index]
+        return {
+            "description": current["description"],
+            "step": self._index,
+            "remaining": len(self._queue) - self._index,
+        }
+
+    def step(self, action: List[int]) -> Tuple[Dict, float, bool, Dict, Dict]:
+        if self._done:
+            return self.state(), 0.0, True, {}, {}
+
+        correct = self._queue[self._index]["label"]
+
+        reward = sum(
+            1.0 if a == b else 0.0
+            for a, b in zip(action, correct)
+        ) / 3.0
+
+        self._index += 1
+
+        if self._index >= len(self._queue):
+            self._done = True
+
+        return self.state(), reward, self._done, {}, {}