Spaces:
Sleeping
Sleeping
| # server/debate_environment.py | |
| import json | |
| import random | |
| import uuid | |
| from typing import Optional | |
| import sys | |
| import os | |
| _here = os.path.dirname(os.path.abspath(__file__)) | |
| sys.path.insert(0, os.path.dirname(_here)) | |
| from judge import judge_argument | |
| from models import DebateAction, DebateObservation, DebateState | |
| _data_path = os.path.join(_here, "debate_data.json") | |
| if not os.path.exists(_data_path): | |
| _data_path = os.path.join(os.path.dirname(_here), "debate_data.json") | |
| with open(_data_path) as f: | |
| DEBATE_DATA = json.load(f) | |
| class DebateEnvironment: | |
| def __init__(self, difficulty: int = 1): | |
| self._difficulty = difficulty | |
| self._current_topic = None | |
| self._attempts = 3 | |
| self._state = DebateState() | |
| def reset(self, difficulty: Optional[int] = None) -> DebateObservation: | |
| if difficulty: | |
| self._difficulty = difficulty | |
| available = [t for t in DEBATE_DATA if t["difficulty"] == self._difficulty] | |
| if not available: | |
| available = DEBATE_DATA # fallback to all topics | |
| self._current_topic = random.choice(available) | |
| self._attempts = 3 | |
| self._state = DebateState( | |
| episode_id=str(uuid.uuid4()), | |
| step_count=0, | |
| current_topic_id=self._current_topic["topic_id"], | |
| difficulty=self._difficulty, | |
| best_reward=0.0 | |
| ) | |
| return DebateObservation( | |
| done=False, | |
| reward=None, | |
| topic=self._current_topic["topic"], | |
| side=self._current_topic["side"], | |
| difficulty=self._difficulty, | |
| attempts_remaining=self._attempts, | |
| feedback=f"Argue {self._current_topic['side']} on: {self._current_topic['topic']}", | |
| ) | |
| def step(self, action: DebateAction) -> DebateObservation: | |
| if self._current_topic is None: | |
| raise RuntimeError("Call reset() before step()") | |
| self._state.step_count += 1 | |
| self._attempts -= 1 | |
| result = judge_argument( | |
| topic=self._current_topic["topic"], | |
| side=self._current_topic["side"], | |
| argument=action.argument, | |
| reference=self._current_topic["example_strong_argument"] | |
| ) | |
| reward = result["reward"] | |
| if reward > self._state.best_reward: | |
| self._state.best_reward = reward | |
| # stop if excellent argument or out of attempts | |
| done = self._attempts <= 0 or reward >= 0.9 | |
| return DebateObservation( | |
| done=done, | |
| reward=reward, | |
| topic=self._current_topic["topic"], | |
| side=self._current_topic["side"], | |
| difficulty=self._difficulty, | |
| attempts_remaining=self._attempts, | |
| feedback=result.get("feedback", ""), | |
| scores=result | |
| ) | |
| def state(self) -> DebateState: | |
| return self._state |