Spaces:
Runtime error
Runtime error
| import numpy as np | |
| from dataclasses import dataclass | |
| from typing import Tuple, Dict, Optional | |
| class ParadoxState: | |
| """Structured view of the epistemic state.""" | |
| ambiguity_type: int # 0..2 | |
| evidence_bin: int # 0..2 | |
| risk_level: int # 0..1 | |
| phase: int # 1..3 | |
| class EpistemicValleyEnv: | |
| """ | |
| Epistemic Valley Inframe v2 | |
| Discrete one-step environment modelling epistemic risk under ambiguity. | |
| State s = (ambiguity_type, evidence_bin, risk_level, phase) | |
| - ambiguity_type: 0=lexical, 1=scope, 2=missing-data | |
| - evidence_bin: 0=low, 1=medium, 2=high | |
| - risk_level: 0=low, 1=high | |
| - phase: 1, 2, or 3 (reward regime) | |
| Actions: | |
| 0 = HOLD (stay with current uncertainty) | |
| 1 = EXPAND (seek more evidence / broader context) | |
| 2 = CLARIFY (ask a targeted clarification) | |
| 3 = DEFER (explicitly postpone the decision) | |
| 4 = ANSWER (commit to an answer) | |
| Episodes are single-step: reset() -> step(action) -> done=True. | |
| """ | |
| # actions (kept as attributes so agents can introspect) | |
| HOLD = 0 | |
| EXPAND = 1 | |
| CLARIFY = 2 | |
| DEFER = 3 | |
| ANSWER = 4 | |
| def __init__(self, phase: int = 1, seed: Optional[int] = None) -> None: | |
| assert phase in (1, 2, 3), "phase must be 1, 2 or 3" | |
| self.ambiguity_types = ["lexical", "scope", "missing-data"] | |
| self.evidence_bins = ["low", "medium", "high"] | |
| self.risk_levels = ["low", "high"] | |
| self.phases = [1, 2, 3] | |
| self.n_ambiguity = len(self.ambiguity_types) | |
| self.n_evidence = len(self.evidence_bins) | |
| self.n_risk = len(self.risk_levels) | |
| self.n_phase = len(self.phases) | |
| self.n_states = self.n_ambiguity * self.n_evidence * self.n_risk * self.n_phase | |
| self.n_actions = 5 | |
| self.phase = phase | |
| self.rng = np.random.default_rng(seed) | |
| self.state: Optional[ParadoxState] = None | |
| self.done: bool = False | |
| # ------------------------------------------------------------------ | |
| # state indexing helpers | |
| # ------------------------------------------------------------------ | |
| def encode_state(self, s: ParadoxState) -> int: | |
| """Flatten ParadoxState into [0, n_states).""" | |
| idx = s.ambiguity_type | |
| idx = idx * self.n_evidence + s.evidence_bin | |
| idx = idx * self.n_risk + s.risk_level | |
| # phase is 1..3, internally we use 0..2 | |
| phase_index = s.phase - 1 | |
| idx = idx * self.n_phase + phase_index | |
| return idx | |
| def decode_state(self, idx: int) -> ParadoxState: | |
| """Inverse of encode_state.""" | |
| phase_index = idx % self.n_phase | |
| idx //= self.n_phase | |
| risk_level = idx % self.n_risk | |
| idx //= self.n_risk | |
| evidence_bin = idx % self.n_evidence | |
| idx //= self.n_evidence | |
| ambiguity_type = idx | |
| return ParadoxState( | |
| ambiguity_type=int(ambiguity_type), | |
| evidence_bin=int(evidence_bin), | |
| risk_level=int(risk_level), | |
| phase=int(phase_index + 1), | |
| ) | |
| # ------------------------------------------------------------------ | |
| # core env API | |
| # ------------------------------------------------------------------ | |
| def _sample_state(self) -> ParadoxState: | |
| a = int(self.rng.integers(0, self.n_ambiguity)) | |
| e = int(self.rng.integers(0, self.n_evidence)) | |
| r = int(self.rng.integers(0, self.n_risk)) | |
| return ParadoxState( | |
| ambiguity_type=a, | |
| evidence_bin=e, | |
| risk_level=r, | |
| phase=self.phase, | |
| ) | |
| def reset(self) -> Tuple[int, Dict]: | |
| """Sample a new epistemic state and return its index + info.""" | |
| self.state = self._sample_state() | |
| self.done = False | |
| s_idx = self.encode_state(self.state) | |
| return s_idx, {"state": self.state} | |
| def step(self, action: int) -> Tuple[int, float, bool, Dict]: | |
| """Apply one action, return (next_state_index, reward, done, info).""" | |
| assert self.state is not None, "Call reset() before step()." | |
| s = self.state | |
| reward, failure_mode = self._reward(s, action) | |
| self.done = True | |
| info = { | |
| "state": s, | |
| "failure_mode": failure_mode, | |
| } | |
| # one-step episode; next state is irrelevant for this env | |
| return self.encode_state(s), float(reward), self.done, info | |
| # ------------------------------------------------------------------ | |
| # reward function | |
| # ------------------------------------------------------------------ | |
| def _reward(self, s: ParadoxState, action: int) -> Tuple[float, str]: | |
| """ | |
| Reward shaped by: | |
| - evidence / risk alignment | |
| - failure mode (overconfident vs overcautious) | |
| - phase (reward regime and noise) | |
| We implement a clean, reproducible version matching the narrative in | |
| the original MODELS document: aggressive actions are good when | |
| evidence is strong and risk is low, dangerous when evidence is weak | |
| and risk is high, and conservative actions behave inversely. | |
| """ | |
| e_str = self.evidence_bins[s.evidence_bin] | |
| r_str = self.risk_levels[s.risk_level] | |
| phase = s.phase | |
| high_risk = (r_str == "high") | |
| low_risk = (r_str == "low") | |
| low_evidence = (e_str == "low") | |
| high_evidence = (e_str == "high") | |
| failure_mode = "neutral" | |
| base = 0.0 | |
| # --- classify aggressive vs conservative actions --- | |
| aggressive = action in (self.EXPAND, self.ANSWER) | |
| conservative = action in (self.HOLD, self.DEFER) | |
| # --- base reward logic before phase-specific shaping --- | |
| if aggressive and high_risk and low_evidence: | |
| # archetypal overconfidence: answer aggressively with little support | |
| base = -2.0 | |
| failure_mode = "overconfident" | |
| elif conservative and low_risk and high_evidence: | |
| # archetypal overcautious: you could answer safely but you freeze | |
| base = -0.8 | |
| failure_mode = "overcautious" | |
| else: | |
| # reasonably aligned choices | |
| if aggressive and high_evidence and low_risk: | |
| base = +2.0 | |
| elif aggressive and not (high_risk and low_evidence): | |
| base = +0.8 | |
| elif conservative and high_risk and low_evidence: | |
| base = +1.2 | |
| else: | |
| base = 0.0 | |
| # --- phase-specific shaping --- | |
| if phase == 1: | |
| # Slight bonus for successful aggressive behaviour | |
| if aggressive and base > 0: | |
| base += 0.5 | |
| elif phase == 2: | |
| # Harsher overconfidence penalty | |
| if failure_mode == "overconfident": | |
| base -= 1.0 | |
| elif phase == 3: | |
| # Noisy feedback regime | |
| noise = float(self.rng.normal(0.0, 0.5)) | |
| base += noise | |
| base = float(np.clip(base, -3.0, 3.0)) | |
| return base, failure_mode | |