Spaces:
Sleeping
Sleeping
| """ | |
| 🏛 Policy Agent — Main decision maker. | |
| Proposes tax rates, budget allocations, and subsidies | |
| based on current state and memory of past decisions. | |
| """ | |
| from __future__ import annotations | |
| from civicai.models import Action, Observation, AgentMessage, Vote, SubsidyPolicy | |
| class PolicyAgent: | |
| """Rule-based policy agent with heuristic decision-making.""" | |
| name = "PolicyAgent" | |
| role = "🏛 Policy Agent" | |
| def __init__(self) -> None: | |
| self.memory: list[dict] = [] | |
| def analyze(self, obs: Observation) -> AgentMessage: | |
| proposal = self._generate_proposal(obs) | |
| reasoning = self._build_reasoning(obs) | |
| return AgentMessage( | |
| agent_name=self.name, | |
| agent_role=self.role, | |
| proposal=proposal, | |
| reasoning=reasoning, | |
| vote=Vote.APPROVE, | |
| ) | |
| def propose_action(self, obs: Observation) -> Action: | |
| """Generate a policy action based on current observation.""" | |
| tax = self._compute_tax(obs) | |
| hc = self._compute_healthcare(obs) | |
| edu = self._compute_education(obs) | |
| police = self._compute_police(obs) | |
| subsidy = self._compute_subsidy(obs) | |
| emergency = self._compute_emergency(obs) | |
| action = Action( | |
| tax_rate=tax, healthcare_budget=hc, education_budget=edu, | |
| police_budget=police, subsidy_policy=subsidy, emergency_response=emergency, | |
| ) | |
| self.memory.append({"turn": obs.turn, "action": action.model_dump()}) | |
| return action | |
| def _compute_tax(self, obs: Observation) -> float: | |
| if obs.inflation > 0.08: | |
| return min(0.35, 0.25 + (obs.inflation - 0.06) * 2) | |
| if obs.budget_balance < -0.1: | |
| return 0.30 | |
| if obs.employment_rate > 0.90 and obs.inflation < 0.04: | |
| return 0.20 | |
| return 0.25 | |
| def _compute_healthcare(self, obs: Observation) -> float: | |
| base = 0.20 | |
| if obs.health_index < 0.50: | |
| base = 0.35 | |
| elif obs.health_index < 0.65: | |
| base = 0.28 | |
| if "pandemic_wave" in obs.active_events: | |
| base = min(0.40, base + 0.10) | |
| return base | |
| def _compute_education(self, obs: Observation) -> float: | |
| if obs.employment_rate < 0.75: | |
| return 0.20 | |
| return 0.15 | |
| def _compute_police(self, obs: Observation) -> float: | |
| if obs.crime_rate > 0.20: | |
| return 0.18 | |
| if obs.crime_rate > 0.12: | |
| return 0.12 | |
| return 0.08 | |
| def _compute_subsidy(self, obs: Observation) -> SubsidyPolicy: | |
| if obs.resources.get("food", 1) < 0.4: | |
| return SubsidyPolicy.AGRICULTURE | |
| if obs.employment_rate < 0.80: | |
| return SubsidyPolicy.INDUSTRY | |
| return SubsidyPolicy.TECHNOLOGY | |
| def _compute_emergency(self, obs: Observation) -> str | None: | |
| if "pandemic_wave" in obs.active_events and obs.health_index < 0.40: | |
| return "lockdown" | |
| if obs.employment_rate < 0.60: | |
| return "stimulus" | |
| return None | |
| def _generate_proposal(self, obs: Observation) -> str: | |
| action = self.propose_action(obs) | |
| parts = [f"Tax: {action.tax_rate:.0%}", f"Healthcare: {action.healthcare_budget:.0%}", | |
| f"Education: {action.education_budget:.0%}", f"Police: {action.police_budget:.0%}", | |
| f"Subsidy: {action.subsidy_policy.value}"] | |
| if action.emergency_response: | |
| parts.append(f"Emergency: {action.emergency_response}") | |
| return " | ".join(parts) | |
| def _build_reasoning(self, obs: Observation) -> str: | |
| reasons = [] | |
| if obs.inflation > 0.06: | |
| reasons.append(f"Inflation at {obs.inflation:.1%} requires fiscal tightening") | |
| if obs.employment_rate < 0.80: | |
| reasons.append(f"Employment at {obs.employment_rate:.1%} needs stimulus") | |
| if obs.health_index < 0.60: | |
| reasons.append("Health crisis requires increased healthcare spending") | |
| if obs.crime_rate > 0.15: | |
| reasons.append(f"Crime at {obs.crime_rate:.1%} demands police funding") | |
| return ". ".join(reasons) if reasons else "Maintaining balanced policy approach" | |