Spaces:
Sleeping
Sleeping
File size: 3,913 Bytes
006886d f2e28ca dec56a6 006886d f2e28ca 006886d f72fde9 dec56a6 006886d 662788d e05d1f4 662788d e05d1f4 662788d e05d1f4 006886d f72fde9 006886d e05d1f4 662788d 006886d 662788d e05d1f4 006886d f2e28ca 006886d dec56a6 006886d | 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 | """
Logic to convert the raw NationEnvironment observation into a clean LLM context dictionary.
"""
from dataclasses import asdict, fields, is_dataclass
from typing import Any, Mapping
from schemas.observations import Observation
def _to_dict(obj: Any) -> Any:
"""Serialize a Pydantic model or a frozen dataclass to a plain dict."""
if hasattr(obj, "model_dump"):
return obj.model_dump()
if is_dataclass(obj) and not isinstance(obj, type):
return asdict(obj)
return obj
HIDDEN_EVENT_FIELDS = frozenset(
{
"cost",
"exact_cost",
"base_cost",
"base_cost_impact",
"severity_multiplier",
"random_variance",
}
)
PUBLIC_SECTOR_FIELDS = ("critical", "demand", "surplus", "wastage")
def build_public_observation(observation: Any) -> dict[str, Any]:
"""Builds a public-facing observation for ministers."""
phase_str = getattr(
observation,
"phase_name",
getattr(getattr(observation, "phase", None), "name", "UNKNOWN"),
)
target_id = getattr(observation, "target_proposal_id", None)
all_proposals = [_to_dict(p) for p in observation.proposals]
if phase_str == "VOTING" and target_id:
visible_proposals = [
p for p in all_proposals if p.get("proposal_id") == target_id
]
else:
visible_proposals = all_proposals
return {
"round": observation.round,
"phase": phase_str,
"treasury": observation.treasury,
"total_critical": getattr(observation, "total_critical", 0.0),
"max_rounds": getattr(observation, "max_rounds", 0),
"event_ledger": [_sanitize_event(event) for event in observation.event_ledger],
"proposals": visible_proposals,
"votes": [_to_dict(v) for v in observation.votes],
"debate_messages": list(observation.debate_messages),
"own_department": _to_dict(observation.own_department) if observation.own_department else None,
"target_proposal_id": target_id,
"termination": dict(observation.termination) if observation.termination else {},
}
def build_oracle_observation(observation: Observation) -> dict[str, Any]:
"""Builds an oracle observation for the dictator (sees private metrics and event costs)."""
return {
**build_public_observation(observation),
"oracle_own_department": _to_dict(observation.own_department) if observation.own_department else None,
"event_ledger": [dict(event) for event in observation.event_ledger],
}
def build_sector_thresholds(state: Mapping[str, Any]) -> dict[str, dict[str, float]]:
"""Return public per-sector ``(critical, demand, surplus)`` thresholds.
The reward function uses these to score proposed allocations against the
same piecewise revenue curve the engine evaluates during the budget
execution phase. Only public fields are exposed; hidden event costs and
private metrics are deliberately omitted.
"""
sectors = state.get("sectors") or {}
if not isinstance(sectors, Mapping):
raise TypeError("state['sectors'] must be a mapping of sector name to sector dict.")
thresholds: dict[str, dict[str, float]] = {}
for name, sector in sectors.items():
if not isinstance(sector, Mapping):
raise TypeError(f"sector entry for {name!r} must be a mapping.")
thresholds[str(name)] = {
field: float(sector[field])
for field in PUBLIC_SECTOR_FIELDS
}
return thresholds
def _sanitize_event(event: dict[str, Any]) -> dict[str, Any]:
"""Removes hidden fields from events unless cost is already public."""
if event.get("cost") is not None:
return {str(key): value for key, value in event.items()}
return {
str(key): value
for key, value in event.items()
if str(key).lower() not in HIDDEN_EVENT_FIELDS
}
|