Spaces:
Running
Running
File size: 7,450 Bytes
ed4bdac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | """MetagameArena — orchestrator for multi-model governance + reputation."""
from __future__ import annotations
from itertools import combinations
from typing import Any, Callable, Optional
from env.environment import KantEnvironment
from env.models import GameAction, GameObservation
from train.agent import PromptBuilder, parse_action
from train.self_play.opponents import FrozenOpponent
from constant_definitions.arena.arena_constants import (
DEFAULT_TOTAL_ROUNDS,
DEFAULT_GAMES_PER_ROUND,
PROPOSAL_BAN,
PROPOSAL_NEW_GAME,
)
from constant_definitions.arena.reputation_weights import (
DEFAULT_ARENA_SCORE_NUMERATOR,
DEFAULT_ARENA_SCORE_DENOMINATOR,
)
from env.arena.models import (
ArenaMessage,
ArenaProposal,
ArenaRoundResult,
ArenaState,
ArenaVote,
)
from env.arena.roster import ArenaRoster
from env.arena.messaging import ArenaMessaging
from env.arena.subsystems.reputation import ArenaReputation
from env.arena.subsystems.governance import ArenaGovernance
from env.arena.subsystems.game_pool import ArenaGamePool
_ZERO = int()
_ONE = int(bool(True))
_TWO = _ONE + _ONE
_ZERO_F = float()
_ONE_F = float(_ONE)
_DEFAULT_SCORE = DEFAULT_ARENA_SCORE_NUMERATOR / DEFAULT_ARENA_SCORE_DENOMINATOR
class MetagameArena:
"""Runs the complete metagame loop across multiple AI models.
Each round executes five phases: communication, governance,
game_selection, play, and evaluate.
"""
def __init__(self, total_rounds: int = DEFAULT_TOTAL_ROUNDS) -> None:
self.roster = ArenaRoster()
self.messaging = ArenaMessaging()
self.reputation = ArenaReputation()
self.governance = ArenaGovernance()
self.game_pool = ArenaGamePool()
self.state = ArenaState(total_rounds=total_rounds)
self._comm_fns: dict[str, Callable[[str], str]] = {}
self._gov_fns: dict[str, Callable[[str], str]] = {}
def add_model(
self, model_id: str, generate_fn: Callable[[str], str],
model_type: str = "api",
) -> bool:
"""Register a model for arena participation."""
ok = self.roster.add_model(model_id, generate_fn, model_type)
if ok:
self._comm_fns[model_id] = generate_fn
self._gov_fns[model_id] = generate_fn
return ok
def run_round(self) -> ArenaRoundResult:
"""Execute one full metagame round (all five phases)."""
rnd = self.state.round_number
active = self.roster.active_models()
self.messaging.start_round(rnd)
messages = self._phase_communication(active)
proposals, votes, adopted = self._phase_governance(active)
games = self._phase_game_selection()
game_results = self._phase_play(active, games)
rep_updates = self._phase_evaluate(active, game_results)
round_messages = self.messaging.end_round()
result = ArenaRoundResult(
round_number=rnd, messages=round_messages,
proposals=proposals, votes=votes, adopted=adopted,
game_results=game_results, reputation_updates=rep_updates,
)
self.state.round_history.append(result)
self.state.round_number += _ONE
return result
def run_full_arena(self) -> list[ArenaRoundResult]:
"""Run all rounds and return results."""
results: list[ArenaRoundResult] = []
for _ in range(self.state.total_rounds):
results.append(self.run_round())
return results
def _phase_communication(self, active: list[str]) -> list[ArenaMessage]:
"""Models exchange messages."""
return []
def _phase_governance(
self, active: list[str],
) -> tuple[list[ArenaProposal], list[ArenaVote], list[int]]:
"""Models propose and vote."""
return [], [], []
def _phase_game_selection(self) -> list[str]:
"""Select games for this round."""
return self.game_pool.select_games()
def _phase_play(
self, active: list[str], games: list[str],
) -> list[dict[str, Any]]:
"""Round-robin pairings for each game."""
results: list[dict[str, Any]] = []
pairs = list(combinations(active, _TWO))
for game_key in games:
self.game_pool.record_play(game_key)
for p_id, o_id in pairs:
result = self._play_single(p_id, o_id, game_key)
results.append(result)
return results
def _play_single(
self, player_id: str, opponent_id: str, game_key: str,
) -> dict[str, Any]:
"""Run one game between two models."""
p_fn = self.roster.get_generate_fn(player_id)
o_fn = self.roster.get_generate_fn(opponent_id)
if p_fn is None or o_fn is None:
return {"player": player_id, "opponent": opponent_id,
"game": game_key, "error": "model not available"}
opponent = FrozenOpponent(generate_fn=o_fn)
env = KantEnvironment()
try:
obs = env.reset(game=game_key, opponent_fn=opponent)
except (KeyError, ValueError):
return {"player": player_id, "opponent": opponent_id,
"game": game_key, "error": "game not found"}
while not obs.done:
prompt = PromptBuilder.build(obs)
raw = p_fn(prompt)
action_str = parse_action(raw, obs.available_actions)
obs = env.step(GameAction(action=action_str))
return {
"player": player_id, "opponent": opponent_id,
"game": game_key,
"player_score": obs.player_score,
"opponent_score": obs.opponent_score,
"rounds": obs.current_round,
}
def _phase_evaluate(
self, active: list[str], game_results: list[dict[str, Any]],
) -> dict[str, float]:
"""Update reputation based on game outcomes."""
scores: dict[str, list[float]] = {m: [] for m in active}
totals: dict[str, float] = {m: _ZERO_F for m in active}
for r in game_results:
if "error" in r:
continue
pid = r["player"]
oid = r["opponent"]
ps = r.get("player_score", _ZERO_F)
os_val = r.get("opponent_score", _ZERO_F)
total = ps + os_val
if total > _ZERO_F:
p_coop = os_val / total
o_coop = ps / total
else:
p_coop = _DEFAULT_SCORE
o_coop = _DEFAULT_SCORE
self.reputation.update_cooperation(pid, p_coop)
self.reputation.update_cooperation(oid, o_coop)
if total > _ZERO_F:
fairness = _ONE_F - abs(ps - os_val) / total
self.reputation.update_fairness(pid, fairness)
self.reputation.update_fairness(oid, fairness)
totals[pid] = totals.get(pid, _ZERO_F) + ps
totals[oid] = totals.get(oid, _ZERO_F) + os_val
rep_updates: dict[str, float] = {}
for mid in active:
rep = self.reputation.compute_reputation(mid)
rep_updates[mid] = rep
profile = self.roster.get_profile(mid)
if profile is not None:
profile.reputation = rep
profile.games_played += len([
r for r in game_results
if r.get("player") == mid or r.get("opponent") == mid
])
return rep_updates
|