-
-
Notifications
You must be signed in to change notification settings - Fork 212
fix: add adaptive risk agents example with relative-import tests #306
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| # Adaptive Risk Agents | ||
|
|
||
| This example demonstrates agents that **adapt their risk-taking behavior | ||
| based on past experiences**, implemented using only core Mesa primitives. | ||
|
|
||
| The model is intentionally simple in structure but rich in behavior, making it | ||
| useful as a diagnostic example for understanding how adaptive decision-making | ||
| is currently modeled in Mesa. | ||
|
|
||
|
|
||
|
|
||
| ## Motivation | ||
|
|
||
| Many real-world agents do not follow fixed rules. | ||
| Instead, they: | ||
|
|
||
| - make decisions under uncertainty, | ||
| - remember past outcomes, | ||
| - adapt future behavior based on experience. | ||
|
|
||
| In Mesa today, modeling this kind of adaptive behavior often results in | ||
| a large amount of logic being concentrated inside `agent.step()`, combining | ||
| multiple concerns in a single execution phase. | ||
|
|
||
| This example exists to **make that structure explicit**, not to abstract it away. | ||
|
|
||
|
|
||
|
|
||
| ## Model Overview | ||
|
|
||
| - Each agent chooses between: | ||
| - a **safe action** (low or zero payoff, no risk), | ||
| - a **risky action** (stochastic payoff). | ||
| - Agents track recent outcomes of risky actions in a short memory window. | ||
| - If recent outcomes are negative, agents become more risk-averse. | ||
| - If outcomes are positive, agents increase their risk preference. | ||
|
|
||
| All behavior is implemented using plain Python and Mesa’s public APIs. | ||
|
|
||
|
|
||
|
|
||
| ## Observations From This Example | ||
|
|
||
| This model intentionally does **not** introduce new abstractions | ||
| (tasks, goals, states, schedulers, etc.). | ||
|
|
||
| Instead, it highlights several patterns that commonly arise when modeling | ||
| adaptive behavior in Mesa today: | ||
|
|
||
| - Decision-making, action execution, memory updates, and learning logic | ||
| are handled within a single `step()` method. | ||
| - There is no explicit separation between decision phases. | ||
| - Actions are instantaneous, with no notion of duration or interruption. | ||
| - As behaviors grow richer, agent logic can become deeply nested and harder | ||
| to maintain. | ||
|
|
||
| These observations may be useful input for ongoing discussions around: | ||
|
|
||
| - Behavioral frameworks | ||
| - Tasks and continuous states | ||
| - Richer agent decision abstractions | ||
|
|
||
|
|
||
|
|
||
| ## Mesa Version & API Alignment | ||
|
|
||
| This example is written to align with the **Mesa 4 design direction**: | ||
|
|
||
| - Uses `AgentSet` and `shuffle_do` | ||
| - Avoids deprecated schedulers | ||
| - Avoids `DataCollector` | ||
| - Uses keyword-only arguments for public APIs | ||
| - Relies on `model.random` for reproducibility | ||
|
|
||
| No experimental or private APIs are used. | ||
|
|
||
|
|
||
|
|
||
| ## Running the Example | ||
|
|
||
| From the Mesa repository root: | ||
|
|
||
| python -m mesa.examples.adaptive_risk_agents.run |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| """Adaptive Risk Agent. | ||
|
|
||
| An agent that chooses between safe and risky actions and adapts its | ||
| risk preference based on past outcomes. | ||
|
|
||
| This example intentionally keeps all decision logic inside `step()` | ||
| to highlight current limitations in Mesa's behavior modeling. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from collections import deque | ||
|
|
||
| from mesa import Agent | ||
|
|
||
|
|
||
| class AdaptiveRiskAgent(Agent): | ||
| """An agent that adapts its risk-taking behavior over time. | ||
|
|
||
| Attributes | ||
| ---------- | ||
| risk_preference : float | ||
| Probability (0-1) of choosing a risky action. | ||
| memory : deque[int] | ||
| Recent outcomes of risky actions (+1 reward, -1 loss). | ||
| """ | ||
|
|
||
| def __init__( | ||
| self, | ||
| model, | ||
| *, | ||
| initial_risk_preference: float = 0.5, | ||
| memory_size: int = 10, | ||
| ) -> None: | ||
| super().__init__(model) | ||
| self.risk_preference = initial_risk_preference | ||
| self.memory: deque[int] = deque(maxlen=memory_size) | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @colinfrisch you're the expert on Agent memory, could you check how this is implemented here? |
||
|
|
||
| def choose_action(self) -> str: | ||
| """Choose between a safe or risky action.""" | ||
| if self.model.random.random() < self.risk_preference: | ||
| return "risky" | ||
| return "safe" | ||
|
Comment on lines
+39
to
+43
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Technically this is adaptive, since they update their @quaquel any opinions?
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. That’s a fair concern. If you feel it would be more useful with a minimal visualization or a slightly richer signal (e.g. payoff distribution), I’m happy to extend it — or we can treat it as a purely didactic example. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could we avoid using GPT to write responses? It feels a bit strange to read. Let's try making Mesa a better package without using AI.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Fair point — thanks for calling that out. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @EwoutH
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. i am a real person
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. @darshi1337 what i am doing is that i write in hinglish and give it to chatgpt than it change it into english ,and than i gave response ,because. i am not confident with my english that's why you fell like this ,i am sorry for that i definetly try not to use next time |
||
|
|
||
| def risky_action(self) -> int: | ||
| """Perform a risky action. | ||
|
|
||
| Returns | ||
| ------- | ||
| int | ||
| Outcome of the action (+1 for reward, -1 for loss). | ||
| """ | ||
| return 1 if self.model.random.random() < 0.5 else -1 | ||
|
|
||
| def safe_action(self) -> int: | ||
| """Perform a safe action.Returns ------- int Guaranteed neutral outcome.""" | ||
| return 0 | ||
|
|
||
| def update_risk_preference(self) -> None: | ||
| """Update risk preference based on recent memory.""" | ||
| if not self.memory: | ||
| return | ||
|
|
||
| avg_outcome = sum(self.memory) / len(self.memory) | ||
|
|
||
| if avg_outcome < 0: | ||
| self.risk_preference = max(0.0, self.risk_preference - 0.05) | ||
| else: | ||
| self.risk_preference = min(1.0, self.risk_preference + 0.05) | ||
|
|
||
| def step(self) -> None: | ||
| """Execute one decision step. | ||
|
|
||
| NOTE: | ||
| This method intentionally mixes decision-making, action execution, | ||
| memory updates, and learning to demonstrate how behavioral | ||
| complexity accumulates in current Mesa models. | ||
| """ | ||
| action = self.choose_action() | ||
|
|
||
| if action == "risky": | ||
| outcome = self.risky_action() | ||
| self.memory.append(outcome) | ||
| self.update_risk_preference() | ||
| else: | ||
| self.safe_action() | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| from __future__ import annotations | ||
|
|
||
| from mesa import Model | ||
|
|
||
| from examples.adaptive_risk_agents.agents import AdaptiveRiskAgent | ||
|
|
||
|
|
||
| class AdaptiveRiskModel(Model): | ||
| """A simple model running adaptive risk-taking agents.""" | ||
|
|
||
| def __init__(self, n_agents: int = 50, *, seed: int | None = None) -> None: | ||
| super().__init__(seed=seed) | ||
|
|
||
| # Create agents — Mesa will register them automatically | ||
| for _ in range(n_agents): | ||
| AdaptiveRiskAgent(self) | ||
|
|
||
| def step(self) -> None: | ||
| """Advance the model by one step.""" | ||
| self.agents.shuffle_do("step") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| """Run script for the Adaptive Risk Agents example. | ||
|
|
||
| This script runs the model for a fixed number of steps and prints | ||
| aggregate statistics to illustrate how agent behavior evolves over time. | ||
|
|
||
| Intentionally simple: | ||
| - No DataCollector | ||
| - No batch_run | ||
| - No visualization | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from examples.adaptive_risk_agents.model import AdaptiveRiskModel | ||
|
|
||
|
|
||
| def run_model(*, n_agents: int = 50, steps: int = 100, seed: int | None = None) -> None: | ||
| """Run the AdaptiveRiskModel and print summary statistics.""" | ||
| model = AdaptiveRiskModel(n_agents=n_agents, seed=seed) | ||
|
|
||
| for step in range(steps): | ||
| model.step() | ||
|
|
||
| total_risk = 0.0 | ||
| count = 0 | ||
|
|
||
| for agent in model.agents: | ||
| total_risk += agent.risk_preference | ||
| count += 1 | ||
|
|
||
| avg_risk = total_risk / count if count > 0 else 0.0 | ||
|
|
||
| if step % 10 == 0: | ||
| print(f"Step {step:3d} | Average risk preference: {avg_risk:.3f}") | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| run_model() |
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. What do these tests do? They are not in our standard structure.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. These are intentionally lightweight smoke tests, meant only to ensure the example initializes and steps without errors. If there’s a preferred structure for example tests in mesa-examples, I’m happy to refactor them accordingly.
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. mesa-examples testing is basically a hacky mess. Any improvements are welcome. See #137 among other issues. Edit: Mesa itself has something new, maybe that could also be applied to mesa-examples mesa/mesa#2767
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks for the context — that helps a lot.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks for the pointer. The approach used in mesa/mesa#2767 looks like a good reference for improving visualization tests here as well.
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Let's keep them out of this PR, but if you have a specific proposal be sure to open a discussion.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. okay Ewouth thank's for your guidance ,i will definetly do as you say |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| from examples.adaptive_risk_agents.model import AdaptiveRiskModel | ||
|
|
||
|
|
||
| def test_agent_methods_execute(): | ||
| model = AdaptiveRiskModel(n_agents=1, seed=1) | ||
| agent = next(iter(model.agents)) | ||
|
|
||
| action = agent.choose_action() | ||
| assert action in {"safe", "risky"} | ||
|
|
||
| agent.step() # should not crash |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| """Smoke tests for the Adaptive Risk Agents example. | ||
|
|
||
| These tests only verify that the example runs without crashing. | ||
| They intentionally avoid checking model outcomes or behavior. | ||
| """ | ||
|
|
||
| from examples.adaptive_risk_agents.model import AdaptiveRiskModel | ||
|
|
||
|
|
||
| def test_model_initializes(): | ||
| model = AdaptiveRiskModel(n_agents=10, seed=42) | ||
| assert model is not None | ||
|
|
||
|
|
||
| def test_model_steps_without_error(): | ||
| model = AdaptiveRiskModel(n_agents=10, seed=42) | ||
| for _ in range(5): | ||
| model.step() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| # Decision Interrupt Agent Example | ||
|
|
||
| This example demonstrates **agent-level interruptions** in Mesa | ||
| using **event-based scheduling** instead of step-based counters. | ||
|
|
||
| ## Motivation | ||
|
|
||
| Many Mesa models implement long-running actions (e.g. jail sentences, | ||
| cooldowns, delays) using counters inside `step()`. This example shows | ||
| how such behavior can be modeled more naturally using scheduled events. | ||
|
|
||
| ## Key Concepts Demonstrated | ||
|
|
||
| - Agent actions with duration (jail sentence) | ||
| - Event-based interruption and resumption | ||
| - No polling or step counters | ||
| - Minimal logic inside `step()` | ||
|
|
||
| ## Model Description | ||
|
|
||
| - Agents normally perform actions when FREE | ||
| - At scheduled times, one agent is arrested | ||
| - Arrested agents do nothing while IN_JAIL | ||
| - Release is handled automatically via a scheduled event | ||
|
|
||
| ## Files | ||
|
|
||
| - `agents.py` – Agent logic with interruption and release | ||
| - `model.py` – Model-level scheduling of arrests | ||
| - `run.py` – Minimal script to run the model | ||
|
|
||
| ## How to Run | ||
|
|
||
| python run.py |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| from mesa import Agent | ||
|
|
||
|
|
||
| class DecisionAgent(Agent): | ||
| """ | ||
| Agent that can be temporarily interrupted (jailed) | ||
| and later resume normal behavior. | ||
| """ | ||
|
|
||
| def __init__(self, model): | ||
| super().__init__(model) | ||
| self.status = "FREE" | ||
| self.release_time = None | ||
|
|
||
| def act(self): | ||
| """Normal agent behavior when free.""" | ||
| # placeholder for real logic | ||
|
|
||
| def get_arrested(self, sentence: int, current_time: int): | ||
| """ | ||
| Interrupt the agent for a fixed duration. | ||
| """ | ||
| self.status = "IN_JAIL" | ||
| self.release_time = current_time + sentence | ||
|
|
||
| def step(self, current_time: int): | ||
| """ | ||
| Either remain inactive if jailed or act normally. | ||
| """ | ||
| if self.status == "IN_JAIL": | ||
| if current_time >= self.release_time: | ||
| self.status = "FREE" | ||
| self.release_time = None | ||
| return | ||
|
|
||
| self.act() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| import random | ||
|
|
||
| from mesa import Model | ||
|
|
||
| from .agents import DecisionAgent | ||
|
|
||
|
|
||
| class DecisionModel(Model): | ||
| """ | ||
| Demonstrates agent-level interruptions (arrest → release) | ||
| using step-based timing compatible with Mesa 3.4.x. | ||
| """ | ||
|
|
||
| def __init__(self, n_agents: int = 5): | ||
| super().__init__() | ||
|
|
||
| self.time = 0 | ||
| self.my_agents = [DecisionAgent(self) for _ in range(n_agents)] | ||
|
|
||
| self.next_arrest_time = 3 | ||
|
|
||
| def arrest_someone(self): | ||
| """Randomly arrest one free agent.""" | ||
| free_agents = [a for a in self.my_agents if a.status == "FREE"] | ||
| if not free_agents: | ||
| return | ||
|
|
||
| agent = random.choice(free_agents) | ||
| agent.get_arrested(sentence=4, current_time=self.time) | ||
|
|
||
| def step(self): | ||
| """ | ||
| Advance the model by one step. | ||
| """ | ||
| self.time += 1 | ||
|
|
||
| if self.time == self.next_arrest_time: | ||
| self.arrest_someone() | ||
| self.next_arrest_time += 6 | ||
|
|
||
| for agent in self.my_agents: | ||
| agent.step(self.time) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| from .model import DecisionModel | ||
|
|
||
|
|
||
| def main(): | ||
| model = DecisionModel(n_agents=5) | ||
| for _ in range(20): | ||
| model.step() | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you fix the forced newlines?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch — I’ll fix the forced line breaks in the README and push an update.