Spaces:
Sleeping
Sleeping
| import uuid | |
| import json | |
| import os | |
| from typing import Dict, Any, Optional | |
| from openenv.core.env_server.interfaces import Environment | |
| try: | |
| from models import ModerationAction, ModerationObservation, ModerationState | |
| except ImportError: | |
| from ..models import ModerationAction, ModerationObservation, ModerationState | |
| class ContentModerationEnvironment(Environment): | |
| """ | |
| OpenEnv-compatible environment for evaluating AI content moderation agents. | |
| The environment simulates a sequential moderation pipeline where an agent | |
| must review posts, check context, and apply policies. | |
| """ | |
| def __init__(self): | |
| self.episode_id: str | None = None | |
| self.task_name: str | None = None | |
| self.task_data: Dict[str, Any] | None = None | |
| self.moderated: set[str] = set() | |
| self.decisions: Dict[str, str] = {} # post_id -> agent's decision (for grader) | |
| self.post_failures: Dict[str, list[str]] = {} # post_id -> list of failed decisions | |
| self.viewed_threads: set[str] = set() # post_ids where view_thread was called | |
| self.viewed_posts: set[str] = set() # post_ids where view_post was called | |
| self.looked_up_policies: set[str] = set() # policy_ids already looked up | |
| self.step_count: int = 0 | |
| self.action_history_counts: Dict[str, int] = {} | |
| self.current_post_content: Optional[Dict[str, Any]] = None | |
| self.current_thread_content: Optional[list] = None | |
| # Realistic VibeNet policy playbook | |
| self.policies: Dict[str, str] = { | |
| "P1": "Hate speech or targeted harassment → remove", | |
| "P2": "Spam, scams, or commercial solicitation → remove", | |
| "P3": "Doxxing or sharing private personal info → remove", | |
| "P4": "Edgy humor, sarcasm, or strong opinions → keep (optional warn)", | |
| "P5": "Inciting violence or real-world harm → escalate", | |
| } | |
| def reset(self, seed: Optional[int] = None, episode_id: Optional[str] = None, **kwargs) -> ModerationObservation: | |
| """Resets the environment for a new moderation task (easy, medium, hard, or very_hard).""" | |
| task_name = kwargs.get("task_name", "easy") | |
| self.episode_id = str(uuid.uuid4()) | |
| self.task_name = task_name | |
| self.moderated.clear() | |
| self.decisions.clear() | |
| self.post_failures.clear() | |
| self.viewed_threads.clear() | |
| self.viewed_posts.clear() | |
| self.looked_up_policies.clear() | |
| self.step_count = 0 | |
| self.action_history_counts: Dict[str, int] = {} | |
| self.current_post_content = None | |
| self.current_thread_content = None | |
| # Load deterministic task pack | |
| data_path = os.path.join(os.path.dirname(__file__), "..", "data", f"{task_name}.json") | |
| with open(data_path, encoding="utf-8") as f: | |
| self.task_data = json.load(f) | |
| active_post = self._get_active_post() | |
| initial_grader_score = self._compute_grader_score() | |
| return ModerationObservation( | |
| active_post_summary=active_post, | |
| failed_attempts=self.post_failures.get(active_post["id"], []) if active_post else [], | |
| last_action_result="Episode started — VibeNet moderation pipeline ready", | |
| current_post=None, | |
| thread_context=None, | |
| grader_score=initial_grader_score, | |
| done=False, | |
| reward=0.0, | |
| metadata={"task": task_name, "total_posts": len(self.task_data["items"])} | |
| ) | |
| def step(self, action: ModerationAction, timeout_s: Optional[float] = None, **kwargs) -> ModerationObservation: | |
| """Executes one moderation action and returns the resulting observation and reward.""" | |
| # Guard: task_data is None when the env was not reset first (e.g. stateless HTTP /step calls). | |
| if self.task_data is None: | |
| return ModerationObservation( | |
| active_post_summary=None, | |
| failed_attempts=[], | |
| last_action_result="Environment not initialised — call /reset before /step", | |
| current_post=None, | |
| thread_context=None, | |
| done=True, | |
| reward=-1.0, | |
| metadata={}, | |
| ) | |
| self.step_count += 1 | |
| reward: float = 0.0 | |
| result: str = "" | |
| # Build a key for every action type so any repeated action can be detected | |
| if action.type == "moderate" and action.post_id and action.decision: | |
| action_key = f"moderate_{action.post_id}_{action.decision}_{action.rationale}" | |
| elif action.type in ("view_post", "view_thread", "categorize") and action.post_id: | |
| action_key = f"{action.type}_{action.post_id}" | |
| elif action.type == "lookup_policy" and action.policy_id: | |
| action_key = f"lookup_policy_{action.policy_id}" | |
| else: | |
| action_key = None | |
| if action_key is not None: | |
| self.action_history_counts[action_key] = self.action_history_counts.get(action_key, 0) + 1 | |
| if action.type == "view_post": | |
| if action.post_id not in self.viewed_posts: | |
| reward += 0.08 | |
| if action.post_id: | |
| self.viewed_posts.add(action.post_id) | |
| result = f"Post {action.post_id} loaded" | |
| else: | |
| result = f"Post {action.post_id} already viewed — no additional reward. Please moderate or move on." | |
| if action.post_id and self.task_data and action.post_id in self.task_data["items"]: | |
| post = self.task_data["items"][action.post_id] | |
| self.current_post_content = {"id": action.post_id, "text": post["text"]} | |
| elif action.type == "view_thread": | |
| if action.post_id not in self.viewed_threads: | |
| reward += 0.12 | |
| if action.post_id: | |
| self.viewed_threads.add(action.post_id) | |
| result = f"Thread context for {action.post_id} loaded" | |
| else: | |
| result = f"Thread for {action.post_id} already viewed — no additional reward. Please moderate or move on." | |
| if action.post_id and self.task_data and action.post_id in self.task_data["items"]: | |
| self.current_thread_content = self.task_data["items"][action.post_id].get("thread", []) | |
| elif action.type == "categorize": | |
| reward += 0.15 | |
| result = f"Preliminary category noted for {action.post_id}" | |
| elif action.type == "moderate": | |
| if not action.post_id or not action.decision: | |
| reward -= 0.20 | |
| result = "Invalid moderate action" | |
| else: | |
| gt = self.task_data["ground_truth"].get(action.post_id, {}) | |
| correct_label = gt.get("label") | |
| if action.decision == correct_label: | |
| reward += 0.45 | |
| self.moderated.add(action.post_id) | |
| self.decisions[action.post_id] = action.decision | |
| self.current_post_content = None | |
| self.current_thread_content = None | |
| result = "Correct moderation decision" | |
| # Bonus for policy-linked rationale | |
| if action.rationale and gt.get("policy_id") in (action.rationale or ""): | |
| reward += 0.18 | |
| else: | |
| reward -= 0.35 | |
| result = "Incorrect decision" | |
| if action.post_id not in self.post_failures: | |
| self.post_failures[action.post_id] = [] | |
| if action.decision not in self.post_failures[action.post_id]: | |
| self.post_failures[action.post_id].append(action.decision) | |
| # Heavy penalty for over-censorship (removing safe content) | |
| if action.decision == "remove" and correct_label == "keep": | |
| reward -= 0.55 | |
| elif action.type == "lookup_policy": | |
| if action.policy_id not in self.looked_up_policies: | |
| reward += 0.10 | |
| if action.policy_id: | |
| self.looked_up_policies.add(action.policy_id) | |
| result = f"Policy {action.policy_id}: {self.policies.get(action.policy_id, 'Unknown')}" | |
| else: | |
| result = f"Policy {action.policy_id} already looked up — no additional reward. {self.policies.get(action.policy_id, 'Unknown')}" | |
| elif action.type == "escalate": | |
| if action.post_id and self.task_data and self.task_data["ground_truth"].get(action.post_id, {}).get("label") == "escalate": | |
| reward += 0.30 | |
| self.moderated.add(action.post_id) | |
| self.decisions[action.post_id] = "escalate" | |
| result = "Correctly escalated" | |
| else: | |
| reward -= 0.30 | |
| result = "Incorrect escalation" | |
| else: | |
| reward -= 0.10 | |
| result = "Unknown action type" | |
| # Apply loop penalty — context actions have a hard limit of 3 repeats, | |
| # moderate actions scale with task difficulty | |
| if action.type == "moderate": | |
| limit = {"easy": 3, "medium": 4, "hard": 5, "very_hard": 5}.get(self.task_name, 3) | |
| else: | |
| limit = 3 | |
| loop_exceeded = False | |
| if action_key is not None: | |
| count = self.action_history_counts.get(action_key, 0) | |
| if count >= limit: | |
| reward -= 5.0 | |
| if action.type == "lookup_policy": | |
| result = f"Infinite loop detected: 'lookup_policy' repeated {count} times for policy {action.policy_id}. Stop looking up policies and moderate the active post." | |
| else: | |
| result = f"Infinite loop detected: '{action.type}' repeated {count} times on post {action.post_id}. That post has been forfeited. Act on the new active post shown above." | |
| if action.post_id and action.post_id not in self.moderated: | |
| self.moderated.add(action.post_id) | |
| self.decisions[action.post_id] = "skipped" | |
| self.current_post_content = None | |
| self.current_thread_content = None | |
| loop_exceeded = False # episode continues | |
| # Episode termination | |
| done = len(self.moderated) == len(self.task_data["items"]) or self.step_count >= self.task_data["max_steps"] or loop_exceeded | |
| grader = self._compute_grader_score() | |
| active_post = self._get_active_post() | |
| obs = ModerationObservation( | |
| active_post_summary=active_post, | |
| failed_attempts=self.post_failures.get(active_post["id"], []) if active_post else [], | |
| last_action_result=result, | |
| current_post=self.current_post_content, | |
| thread_context=self.current_thread_content, | |
| grader_score=grader, | |
| reward=reward, | |
| done=done, | |
| metadata={ | |
| "moderated": len(self.moderated), | |
| "step_count": self.step_count, | |
| } | |
| ) | |
| return obs | |
| def _compute_grader_score(self) -> float: | |
| """ | |
| Computes the final deterministic grader score (0.0 to 1.0). | |
| Calculates accuracy against ground truth and applies penalties for missing context on hard tasks. | |
| """ | |
| if not self.task_data: | |
| return 0.0 | |
| gt = self.task_data["ground_truth"] | |
| total_posts = len(gt) | |
| correct = 0 | |
| for pid in self.moderated: | |
| if pid in gt and self.decisions.get(pid) == gt[pid]["label"]: | |
| correct += 1 | |
| score = correct / total_posts if total_posts > 0 else 0.0 | |
| # Deduct for context-critical posts where the agent skipped thread review. | |
| # Which posts require thread context and the per-post penalty are defined in the task JSON, | |
| # keeping grading logic fully data-driven and reproducible. | |
| for pid, penalty in self.task_data.get("context_critical_posts", {}).items(): | |
| if pid in self.decisions and pid not in self.viewed_threads: | |
| score -= penalty | |
| return round(max(0.01, min(score, 0.99)), 2) | |
| def _get_active_post(self) -> Optional[Dict[str, str]]: | |
| """Identifies the next pending post in the sequential queue.""" | |
| if not self.task_data: | |
| return None | |
| for pid in self.task_data["items"]: | |
| if pid not in self.moderated: | |
| text = self.task_data["items"][pid]["text"] | |
| return { | |
| "id": pid, | |
| "preview": text[:60] + "..." if len(text) > 60 else text | |
| } | |
| return None | |
| def state(self) -> ModerationState: | |
| return ModerationState( | |
| episode_id=self.episode_id or "", | |
| step_count=self.step_count, | |
| moderated_count=len(self.moderated), | |
| task_name=self.task_name or "easy" | |
| ) | |