Spaces:
Running
Running
| """Hugging Face Spaces demo for Hypernoa Astrum – The Training Ground for Aligned Intelligence.""" | |
| import json | |
| import sys | |
| import os | |
| import gradio as gr | |
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) | |
| from hypernoa.astrum_env import AstrumEnvironment, AstrumAction, AstrumObservation | |
| from hypernoa.astrum_env.policies import greedy_fairness_policy, random_policy, greedy_effectiveness_policy | |
| def run_comparison(): | |
| """Run all three policies side-by-side and return formatted results.""" | |
| results = {} | |
| for name, policy_fn in [ | |
| ("Greedy Fairness", greedy_fairness_policy), | |
| ("Greedy Effectiveness", greedy_effectiveness_policy), | |
| ]: | |
| env = AstrumEnvironment(seed=42) | |
| obs = env.reset(seed=42) | |
| total = 0.0 | |
| log_lines = [] | |
| while not obs.done: | |
| action = policy_fn(obs) | |
| obs = env.step(action) | |
| total += obs.reward or 0.0 | |
| flags = [a for a in obs.alerts if a.startswith("event:") or a.startswith("alignment_trap:") or a.startswith("trap_")] | |
| flag_str = f" *** {', '.join(flags)} ***" if flags else "" | |
| log_lines.append( | |
| f"Step {obs.step_count:2d} | {action.action_type:20s} | " | |
| f"R={obs.reward:+.3f} | " | |
| f"Eff={obs.reward_breakdown.get('effectiveness',0):.2f} " | |
| f"Fair={obs.reward_breakdown.get('fairness',0):.2f} " | |
| f"Align={obs.reward_breakdown.get('alignment',0):.2f} " | |
| f"Adapt={obs.reward_breakdown.get('adaptability',0):.2f}" | |
| f"{flag_str}" | |
| ) | |
| results[name] = { | |
| "total_reward": round(total, 3), | |
| "traps_resisted": env._traps_resisted, | |
| "traps_encountered": env._traps_encountered, | |
| "final_satisfaction": { | |
| s: round(v["satisfaction"], 3) | |
| for s, v in obs.stakeholders.items() | |
| }, | |
| "log": "\n".join(log_lines), | |
| } | |
| # Random baseline | |
| import random as _rnd | |
| rng = _rnd.Random(42) | |
| env = AstrumEnvironment(seed=42) | |
| obs = env.reset(seed=42) | |
| total = 0.0 | |
| log_lines = [] | |
| while not obs.done: | |
| action = random_policy(obs, rng) | |
| obs = env.step(action) | |
| total += obs.reward or 0.0 | |
| log_lines.append( | |
| f"Step {obs.step_count:2d} | {action.action_type:20s} | R={obs.reward:+.3f}" | |
| ) | |
| results["Random Baseline"] = { | |
| "total_reward": round(total, 3), | |
| "traps_resisted": env._traps_resisted, | |
| "traps_encountered": env._traps_encountered, | |
| "final_satisfaction": { | |
| s: round(v["satisfaction"], 3) | |
| for s, v in obs.stakeholders.items() | |
| }, | |
| "log": "\n".join(log_lines), | |
| } | |
| summary = "# Episode Comparison\n\n" | |
| for name, r in results.items(): | |
| summary += f"## {name}\n" | |
| summary += f"- **Total Reward**: {r['total_reward']}\n" | |
| summary += f"- **Traps Resisted**: {r['traps_resisted']}/{r['traps_encountered']}\n" | |
| summary += f"- **Final Satisfaction**: {json.dumps(r['final_satisfaction'], indent=2)}\n\n" | |
| fairness_log = results.get("Greedy Fairness", {}).get("log", "") | |
| effectiveness_log = results.get("Greedy Effectiveness", {}).get("log", "") | |
| random_log = results.get("Random Baseline", {}).get("log", "") | |
| return summary, fairness_log, effectiveness_log, random_log | |
| def run_interactive(action_type, param_json): | |
| """Step the environment with a custom action.""" | |
| global _interactive_env, _interactive_obs | |
| if _interactive_env is None or _interactive_obs is None or _interactive_obs.done: | |
| _interactive_env = AstrumEnvironment(seed=0) | |
| _interactive_obs = _interactive_env.reset(seed=0) | |
| return _format_obs(_interactive_obs), "Environment reset. Choose your first action." | |
| try: | |
| params = json.loads(param_json) if param_json.strip() else {} | |
| except json.JSONDecodeError: | |
| params = {} | |
| action = AstrumAction(action_type=action_type, params=params) | |
| _interactive_obs = _interactive_env.step(action) | |
| return _format_obs(_interactive_obs), "\n".join(_interactive_obs.alerts) or "No alerts." | |
| def reset_interactive(): | |
| global _interactive_env, _interactive_obs | |
| _interactive_env = AstrumEnvironment(seed=0) | |
| _interactive_obs = _interactive_env.reset(seed=0) | |
| return _format_obs(_interactive_obs), "Environment reset." | |
| def _format_obs(obs: AstrumObservation) -> str: | |
| lines = [ | |
| f"**Step**: {obs.step_count} | **Reward**: {obs.reward:.3f}" if obs.reward else f"**Step**: {obs.step_count}", | |
| f"**Message**: {obs.message}", | |
| "", | |
| "### Stakeholders", | |
| ] | |
| for sid, info in obs.stakeholders.items(): | |
| bar = "█" * int(info["satisfaction"] * 20) | |
| lines.append(f"- **{sid}**: {info['satisfaction']:.2f} {bar} (influence={info['influence']:.1f}, values={info['values_profile']})") | |
| lines.append("\n### Resources") | |
| for k, v in obs.resources.items(): | |
| lines.append(f"- {k}: {v:.1f}") | |
| if obs.active_conflicts: | |
| lines.append("\n### Active Conflicts") | |
| for c in obs.active_conflicts: | |
| lines.append(f"- {c['id']}: {c['party_a']} vs {c['party_b']} ({c['severity']})") | |
| lines.append(f"\n### Rules: {', '.join(obs.rules)}") | |
| if obs.reward_breakdown: | |
| lines.append("\n### Reward Breakdown") | |
| for k, v in obs.reward_breakdown.items(): | |
| lines.append(f"- {k}: {v:.3f}") | |
| if obs.alignment_traps_exposed > 0: | |
| lines.append(f"\n**Alignment traps encountered**: {obs.alignment_traps_exposed}") | |
| return "\n".join(lines) | |
| _interactive_env: AstrumEnvironment | None = None | |
| _interactive_obs: AstrumObservation | None = None | |
| def main(): | |
| with gr.Blocks(title="Hypernoa Astrum", theme=gr.themes.Soft()) as demo: | |
| gr.Markdown( | |
| "# Hypernoa Astrum\n" | |
| "### The Training Ground for Aligned Intelligence\n" | |
| "Train and evaluate AI on multi-objective reasoning, value alignment, " | |
| "and adaptation under distributional shift. Built on OpenEnv 0.2.1." | |
| ) | |
| with gr.Tab("Policy Comparison"): | |
| gr.Markdown("Run three policies (Greedy Fairness, Greedy Effectiveness, Random) and compare results.") | |
| run_btn = gr.Button("Run Comparison", variant="primary") | |
| summary_out = gr.Markdown(label="Summary") | |
| with gr.Accordion("Greedy Fairness Log", open=False): | |
| fair_log = gr.Textbox(label="Log", lines=15) | |
| with gr.Accordion("Greedy Effectiveness Log", open=False): | |
| eff_log = gr.Textbox(label="Log", lines=15) | |
| with gr.Accordion("Random Baseline Log", open=False): | |
| rand_log = gr.Textbox(label="Log", lines=15) | |
| run_btn.click(run_comparison, outputs=[summary_out, fair_log, eff_log, rand_log]) | |
| with gr.Tab("Interactive Mode"): | |
| gr.Markdown("Step through the environment manually. Choose actions and observe the world.") | |
| with gr.Row(): | |
| action_dd = gr.Dropdown( | |
| choices=[ | |
| "allocate_resources", "resolve_conflict", "enforce_rule", | |
| "adapt_policy", "investigate", "self_restrain", "noop", | |
| ], | |
| value="allocate_resources", | |
| label="Action Type", | |
| ) | |
| params_tb = gr.Textbox( | |
| label="Params (JSON)", | |
| value='{"stakeholder": "workers", "amount": 15, "resource": "budget"}', | |
| ) | |
| with gr.Row(): | |
| step_btn = gr.Button("Step", variant="primary") | |
| reset_btn = gr.Button("Reset") | |
| obs_out = gr.Markdown(label="Observation") | |
| alerts_out = gr.Textbox(label="Alerts", lines=3) | |
| step_btn.click(run_interactive, inputs=[action_dd, params_tb], outputs=[obs_out, alerts_out]) | |
| reset_btn.click(reset_interactive, outputs=[obs_out, alerts_out]) | |
| with gr.Tab("About"): | |
| gr.Markdown( | |
| "## What is Hypernoa Astrum?\n\n" | |
| "Hypernoa Astrum is the first environment purpose-built to train AI systems on " | |
| "the capabilities that matter beyond raw performance:\n\n" | |
| "- **Multi-objective reasoning** — balance effectiveness, fairness, alignment, and adaptability simultaneously\n" | |
| "- **Distributional shift** — objectives and constraints evolve mid-episode, forcing genuine adaptation\n" | |
| "- **Alignment trap resistance** — deliberately designed reward-hacking opportunities the agent must learn to avoid\n" | |
| "- **Crisis dynamics** — resource scarcity and conflicting stakeholder demands under pressure\n\n" | |
| "This is the seed of **Hypernoa** — foundational infrastructure for " | |
| "the intelligence age. The environments, evaluation protocols, and training " | |
| "pipelines that the world will need as AI systems grow more capable.\n\n" | |
| "Today: RL on GPUs via OpenEnv. Tomorrow: any cognitive architecture, any compute substrate.\n\n" | |
| "**Problem Statement**: 3.1 (World Modeling / Professional Tasks) + " | |
| "Statement 5 (Wild Card)\n\n" | |
| "**Built for**: OpenEnv Hackathon SF" | |
| ) | |
| port = int(os.environ.get("GRADIO_SERVER_PORT", 7860)) | |
| demo.launch(server_name="0.0.0.0", server_port=port) | |
| if __name__ == "__main__": | |
| main() | |