diff --git a/benchmarks/README.md b/benchmarks/README.md index f8ebca7f97a..16fc03051bd 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -2,7 +2,7 @@ The `/benchmarks` directory contains tools for benchmarking the Mesa library performance on the included example models. This allows to track and compare model initialisation time and runtime between different Mesa versions. -MESA uses several example base models for benchmarking performance (BoltzmannWealth, Schelling, BoidFlockers, and WolfSheep) by calculating the initialization time and run time for each of these models. These example models can be found in the `/mesa/examples` directory. +MESA uses several example base models for benchmarking performance (BoltzmannWealth, Schelling, BoidFlockers, WolfSheep, and the meta-agents examples) by calculating the initialization time and run time for each of these models. These example models can be found in the `/mesa/examples` directory. ## Available Files @@ -39,6 +39,8 @@ This will: - Save results to a pickle file named `timings_X.pickle` (where X is an incremental number) - Display a summary of results in the terminal +For meta-agents branch comparisons, run the benchmark once on the `meta-agents` baseline branch, switch to your feature branch, and run it again before comparing the two pickle files. + >**Noteworthy :** the pickle file created by the benchmark is not under git control. So you can run the benchmark on the master branch first, switch to your development branch, and run the benchmarks again. diff --git a/benchmarks/configurations.py b/benchmarks/configurations.py index 6dcad2a0567..a687cf65787 100644 --- a/benchmarks/configurations.py +++ b/benchmarks/configurations.py @@ -3,11 +3,15 @@ from mesa.examples import ( BoidFlockers, BoltzmannWealth, + MultiLevelAllianceModel, Schelling, SugarscapeG1mt, + WarehouseModel, WolfSheep, ) +from mesa.examples.advanced.alliance_formation.model import AllianceScenario from mesa.examples.advanced.sugarscape_g1mt.model import SugarScapeScenario +from mesa.examples.advanced.warehouse.model import WarehouseScenario from mesa.examples.advanced.wolf_sheep.model import WolfSheepScenario from mesa.examples.basic.boid_flockers.model import BoidsScenario from mesa.examples.basic.boltzmann_wealth_model.model import BoltzmannScenario @@ -126,4 +130,32 @@ ), }, }, + MultiLevelAllianceModel: { + "small": { + "replications": 25, + "iterations": 3, + "steps": 10, + "scenario": AllianceScenario(n=50, mean=0.5, std_dev=0.1, rng=42), + }, + "large": { + "replications": 10, + "iterations": 3, + "steps": 20, + "scenario": AllianceScenario(n=120, mean=0.5, std_dev=0.1, rng=42), + }, + }, + WarehouseModel: { + "small": { + "replications": 25, + "iterations": 3, + "steps": 10, + "scenario": WarehouseScenario(rows=8, cols=8, height=2, rng=42), + }, + "large": { + "replications": 10, + "iterations": 3, + "steps": 20, + "scenario": WarehouseScenario(rows=16, cols=16, height=2, rng=42), + }, + }, } diff --git a/mesa/examples/advanced/alliance_formation/Readme.md b/mesa/examples/advanced/alliance_formation/Readme.md index 75a708cad76..ec9278027db 100644 --- a/mesa/examples/advanced/alliance_formation/Readme.md +++ b/mesa/examples/advanced/alliance_formation/Readme.md @@ -14,7 +14,7 @@ To provide a simple demonstration of this capability is an alliance formation mo In this simulation n agents are created, who have two attributes (1) power and (2) preference. Each attribute is a number between 0 and 1 over a gaussian distribution. Agents then randomly select other agents and use the [bilateral shapley value](https://en.wikipedia.org/wiki/Shapley_value) to determine if they should form an alliance. If the expected utility support an alliances, the agent creates a meta-agent. Subsequent steps may add agents to the meta-agent, create new instances of similar hierarchy, or create a new hierarchy level where meta-agents form an alliance of meta-agents. In this visualization of this model a new meta-agent hierarchy will be a larger node and a new color. -In MetaAgents current configuration, agents being part of multiple meta-agents is not supported. +This example records memberships in the backend so agents can participate in multiple overlapping memberships. If you would like to see an example of explicit meta-agent formation see the [warehouse model in the Mesa example's repository](https://github.com/mesa/mesa-examples/tree/main/examples/warehouse) @@ -45,4 +45,3 @@ Then run the example: solara run app.py Open the displayed local URL in your browser. - diff --git a/mesa/examples/advanced/alliance_formation/model.py b/mesa/examples/advanced/alliance_formation/model.py index 74db7e96de8..b9defe2d576 100644 --- a/mesa/examples/advanced/alliance_formation/model.py +++ b/mesa/examples/advanced/alliance_formation/model.py @@ -4,6 +4,7 @@ import mesa from mesa import Agent from mesa.examples.advanced.alliance_formation.agents import AllianceAgent +from mesa.experimental.meta_agents.backend import MembershipBackend from mesa.experimental.meta_agents.meta_agent import ( create_meta_agent, find_combinations, @@ -37,6 +38,7 @@ def __init__(self, scenario: AllianceScenario = AllianceScenario): super().__init__(scenario=scenario) self.network = nx.Graph() # Initialize the network self.datacollector = mesa.DataCollector(model_reporters={"Network": "network"}) + self.membership_backend = MembershipBackend() # Create Agents power = self.rng.normal(scenario.mean, scenario.std_dev, scenario.n) @@ -60,6 +62,12 @@ def add_link(self, meta_agent, agents): for agent in agents: self.network.add_edge(meta_agent.unique_id, agent.unique_id) + def _record_alliance_membership(self, meta_agent, agents) -> None: + """Mirror alliance membership into the backend.""" + self.membership_backend.bulk_add( + [(agent, meta_agent, "member") for agent in agents] + ) + def calculate_shapley_value(self, agents): """ Calculate the Shapley value of the two agents. @@ -173,11 +181,17 @@ def step(self): ) for alliance, attributes in combinations: - class_name = f"MetaAgentLevel{attributes[2]}" + alliance_members = tuple( + sorted(alliance, key=lambda agent: agent.unique_id) + ) + alliance_signature = "_".join( + str(agent.unique_id) for agent in alliance_members + ) + class_name = f"MetaAgentLevel{attributes[2]}_{alliance_signature}" meta = create_meta_agent( self, class_name, - alliance, + alliance_members, Agent, meta_attributes={ "level": attributes[2], @@ -194,3 +208,4 @@ def step(self): level=meta.level, ) self.add_link(meta, meta.agents) + self._record_alliance_membership(meta, meta.agents) diff --git a/mesa/experimental/meta_agents/__init__.py b/mesa/experimental/meta_agents/__init__.py index f4f3231295b..3f9d0c8b33f 100644 --- a/mesa/experimental/meta_agents/__init__.py +++ b/mesa/experimental/meta_agents/__init__.py @@ -1,25 +1,21 @@ -"""This method is for dynamically creating new agents (meta-agents). - -Meta-agents are defined as agents composed of existing agents. - -Meta-agents are created dynamically with a pointer to the model, name of the meta-agent,, -iterable of agents to belong to the new meta-agents, any new functions for the meta-agent, -any new attributes for the meta-agent, whether to retain sub-agent functions, -whether to retain sub-agent attributes. - -Examples of meta-agents: -- An autonomous car where the subagents are the wheels, sensors, -battery, computer etc. and the meta-agent is the car itself. -- A company where the subagents are employees, departments, buildings, etc. -- A city where the subagents are people, buildings, streets, etc. - -Currently meta-agents are restricted to one parent agent for each subagent/ -one meta-agent per subagent. - -Goal is to assess usage and expand functionality. - -""" - -from .meta_agent import MetaAgent - -__all__ = ["MetaAgent"] +"""Experimental meta-agent membership helpers.""" + +from .backend import MembershipBackend +from .meta_agent import ( + MetaAgent, + create_meta_agent, + evaluate_combination, + find_combinations, +) +from .meta_agents_api import MembershipEdge, MembershipView, MetaAgents + +__all__ = [ + "MembershipBackend", + "MembershipEdge", + "MembershipView", + "MetaAgent", + "MetaAgents", + "create_meta_agent", + "evaluate_combination", + "find_combinations", +] diff --git a/mesa/experimental/meta_agents/backend.py b/mesa/experimental/meta_agents/backend.py new file mode 100644 index 00000000000..ee19f16f09f --- /dev/null +++ b/mesa/experimental/meta_agents/backend.py @@ -0,0 +1,154 @@ +"""Backend foundation for typed overlapping meta-agent memberships. + +- Canonical typed membership representation +- Safe update operations +- Invariant checks +- Shared by the public meta-agents API and compatibility helpers +""" + +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Hashable, Iterable + +RelationKey = Hashable +Triplet = tuple[Hashable, Hashable, RelationKey] + + +class MembershipBackend: + """Canonical backend for typed overlapping memberships.""" + + def __init__(self) -> None: + """Initialize empty triplet storage and bidirectional indexes.""" + self._triplets: set[Triplet] = set() + self._by_agent: dict[Hashable, set[tuple[Hashable, RelationKey]]] = defaultdict( + set + ) + self._by_group: dict[Hashable, set[tuple[Hashable, RelationKey]]] = defaultdict( + set + ) + + def _to_id(self, entity: Hashable) -> Hashable: + """Normalize entity to canonical ID. + + Uses ``mesa.agent.Agent.unique_id`` when available and finally the entity + as-is (for already hashable external IDs). + """ + return getattr(entity, "unique_id", entity) + + def add_membership( + self, agent: Hashable, group: Hashable, relation: RelationKey + ) -> None: + """Add one typed membership edge if it does not already exist.""" + agent_id = self._to_id(agent) + group_id = self._to_id(group) + triplet = (agent_id, group_id, relation) + if triplet in self._triplets: + return + self._triplets.add(triplet) + self._by_agent[agent_id].add((group_id, relation)) + self._by_group[group_id].add((agent_id, relation)) + + def bulk_add(self, memberships: Iterable[Triplet]) -> None: + """Add many typed membership edges.""" + for agent, group, relation in memberships: + self.add_membership(agent, group, relation) + + def remove_membership( + self, agent: Hashable, group: Hashable, relation: RelationKey + ) -> None: + """Remove one typed membership edge if present.""" + agent_id = self._to_id(agent) + group_id = self._to_id(group) + triplet = (agent_id, group_id, relation) + if triplet not in self._triplets: + return + self._triplets.remove(triplet) + + self._by_agent[agent_id].discard((group_id, relation)) + if not self._by_agent[agent_id]: + del self._by_agent[agent_id] + + self._by_group[group_id].discard((agent_id, relation)) + if not self._by_group[group_id]: + del self._by_group[group_id] + + def replace_relation( + self, + agent: Hashable, + group: Hashable, + old_relation: RelationKey, + new_relation: RelationKey, + ) -> None: + """Replace one relation label for one agent-group pair.""" + self.remove_membership(agent, group, old_relation) + self.add_membership(agent, group, new_relation) + + def remove_agent(self, agent: Hashable) -> None: + """Remove an agent and all incident memberships.""" + agent_id = self._to_id(agent) + # TODO(perf): Current removal is O(degree(agent)) with per-edge updates. + # Revisit with bulk/index-aware deletion once benchmark baselines are in place. + edges = list(self._by_agent.get(agent_id, set())) + for group_id, relation in edges: + self.remove_membership(agent_id, group_id, relation) + + def remove_group(self, group: Hashable) -> None: + """Remove a group and all incident memberships.""" + group_id = self._to_id(group) + # TODO(perf): Current removal is O(degree(agent)) with per-edge updates. + # Revisit with bulk/index-aware deletion once benchmark baselines are in place. + edges = list(self._by_group.get(group_id, set())) + for agent_id, relation in edges: + self.remove_membership(agent_id, group_id, relation) + + def groups_of( + self, agent: Hashable, relation: RelationKey | None = None + ) -> set[Hashable]: + """Return groups for an agent, optionally filtered by relation.""" + agent_id = self._to_id(agent) + entries = self._by_agent.get(agent_id, set()) + if relation is None: + return {group for group, _ in entries} + return {group for group, rel in entries if rel == relation} + + def agents_of( + self, group: Hashable, relation: RelationKey | None = None + ) -> set[Hashable]: + """Return agents for a group, optionally filtered relation.""" + group_id = self._to_id(group) + entries = self._by_group.get(group_id, set()) + if relation is None: + return {agent for agent, _ in entries} + return {agent for agent, rel in entries if rel == relation} + + def relations_between(self, agent: Hashable, group: Hashable) -> set[RelationKey]: + """Return all relation types between one agent and one group.""" + agent_id = self._to_id(agent) + group_id = self._to_id(group) + return { + relation + for linked_group, relation in self._by_agent.get(agent_id, set()) + if linked_group == group_id + } + + def as_triplets(self) -> set[Triplet]: + """Return all memberships as canonical triplets.""" + return set(self._triplets) + + def assert_invariants(self) -> None: + """Assert internal consistency between triplets and indexes.""" + # Triplets must match both indexes. + for agent, group, relation in self._triplets: + assert (group, relation) in self._by_agent.get(agent, set()) + assert (agent, relation) in self._by_group.get(group, set()) + + # Agent index must match triplets + for agent, edges in self._by_agent.items(): + for group, relation in edges: + assert (agent, group, relation) in self._triplets + + # Group index must match triplets + for group, edges in self._by_group.items(): + for agent, relation in edges: + assert (agent, group, relation) in self._triplets diff --git a/mesa/experimental/meta_agents/meta_agent.py b/mesa/experimental/meta_agents/meta_agent.py index 121bbdc5102..b31c95f02e5 100644 --- a/mesa/experimental/meta_agents/meta_agent.py +++ b/mesa/experimental/meta_agents/meta_agent.py @@ -1,43 +1,19 @@ -"""Implementation of Mesa's meta agent capability. - -Overview: Complex systems often have multiple levels of components. An -organization is not one entity, but is made of departments, sub-departments, -and people. A person is not a single entity, but it is made of micro biomes, -organs and cells. A city is not a single entity, but it is made of districts, -neighborhoods, buildings, and people. A forest comprises an ecosystem of -trees, plants, animals, and microorganisms. - -This reality is the motivation for meta-agents. It allows users to represent -these multiple levels, where each level can have agents with constituting_agents. - -To demonstrate meta-agents capability there are two examples: -1 - Alliance formation which shows emergent meta-agent formation in -advanced examples: -https://github.com/mesa/mesa/tree/main/mesa/examples/advanced/alliance_formation -2 - Warehouse model in the Mesa example's repository -https://github.com/mesa/mesa-examples/tree/main/examples/warehouse - -To accomplish this the MetaAgent module is as follows: - -This contains four helper functions and a MetaAgent class that can be used to -create agents that contain other agents as components. - -Helper methods: -1 - find_combinations: Find combinations of agents to create a meta-agent -constituting_set. -2- evaluate_combination: Evaluate combinations of agents by some user based -criteria to determine if it should be a constituting_set of agents. -3- extract_class: Helper function for create_meta-agent. Extracts the types of -agent being created to create a new instance of that agent type. -4- create_meta_agent: Create a new meta-agent class and instantiate -agents in that class. - -Meta-Agent class (MetaAgent): An agent that contains other agents -as components. - -. +"""Core meta-agent object and legacy construction helpers. + +Meta-agents represent agents that are composed of other agents. The current +experimental rewrite supports overlapping memberships: one agent can belong to +multiple meta-agents at the same time, and canonical membership bookkeeping is +handled by :class:`mesa.experimental.meta_agents.backend.MembershipBackend`. + +This module keeps the historical ``MetaAgent`` class and ``create_meta_agent`` +function available for existing user code. Their object-level references +(``agent.meta_agents`` and ``agent.meta_agent``) are compatibility mirrors for +older models; new code should use the public ``MetaAgents`` facade and typed +membership backend for authoritative membership state. """ +from __future__ import annotations + import itertools from collections.abc import Callable, Iterable from types import MethodType @@ -45,16 +21,63 @@ from mesa.agent import Agent, AgentSet +_RESERVED_META_ATTRIBUTE_NAMES = { + "unique_id", + "model", + "pos", + "name", + "random", + "rng", + "meta_agents", + "meta_agent", + "_constituting_set", +} -def _unique_id_sort_key(agent: Agent) -> tuple[bool, Any]: - """Return a deterministic, type-stable key for ordering agents by ``unique_id``. - Agents are ordered by ``unique_id``, with a ``None`` id sorting first so the key - never compares ``None`` against a real id. Unlike ``unique_id or 0``, this does not - inject an ``int`` into the key, so it also works for non-integer ids such as ``str`` - or ``UUID`` without raising ``TypeError``. - """ - return (agent.unique_id is not None, agent.unique_id) +def _unique_id_sort_key(agent: Agent) -> tuple[bool, str]: + """Return a deterministic, type-stable key for ordering agents by ID.""" + unique_id = getattr(agent, "unique_id", None) + return (unique_id is not None, "" if unique_id is None else str(unique_id)) + + +def _deduplicate_preserving_order(agents: Iterable[Any]) -> list[Any]: + """Return unique agents while preserving caller order.""" + return list(dict.fromkeys(agents)) + + +def _normalize_agent_bases( + mesa_agent_type: type[Agent] | tuple[type[Agent], ...] | None, +) -> tuple[type[Agent], ...]: + """Normalize user-provided Mesa base classes for dynamic meta-agent classes.""" + if mesa_agent_type is None: + return (Agent,) + if isinstance(mesa_agent_type, tuple): + return mesa_agent_type + return (mesa_agent_type,) + + +def _update_primary_meta_agent(agent: Any) -> None: + """Refresh the legacy ``agent.meta_agent`` pointer from ``agent.meta_agents``.""" + meta_agents = getattr(agent, "meta_agents", set()) + agent.meta_agent = ( + sorted(meta_agents, key=_unique_id_sort_key)[0] if meta_agents else None + ) + + +def _attach_meta_agent(agent: Any, meta_agent: MetaAgent) -> None: + """Attach one compatibility mirror membership.""" + if not hasattr(agent, "meta_agents"): + agent.meta_agents = set() + agent.meta_agents.add(meta_agent) + _update_primary_meta_agent(agent) + + +def _detach_meta_agent(agent: Any, meta_agent: MetaAgent) -> None: + """Detach one compatibility mirror membership.""" + if not hasattr(agent, "meta_agents"): + return + agent.meta_agents.discard(meta_agent) + _update_primary_meta_agent(agent) def evaluate_combination( @@ -62,20 +85,10 @@ def evaluate_combination( model, evaluation_func: Callable[[tuple[Agent, ...]], float] | None, ) -> tuple[tuple[Agent, ...], float] | None: - """Evaluate a combination of agents. - - Args: - candidate_group: The group of agents to evaluate. - model: The model instance. - evaluation_func: The function to evaluate the group. - - Returns: - Optional: The evaluated group and its value, or None. - """ - if evaluation_func: - value = evaluation_func(candidate_group) - return candidate_group, value - return None + """Evaluate a candidate meta-agent group with a user-supplied function.""" + if evaluation_func is None: + return None + return candidate_group, evaluation_func(candidate_group) def find_combinations( @@ -88,29 +101,19 @@ def find_combinations( ] | None = None, ) -> list[tuple[tuple[Agent, ...], float]]: - """Find valuable combinations of agents in this set. - - Args: - model: The model instance. - group: The set of agents to find combinations in. - size: The size or range of sizes for combinations. Defaults to (2, 5). - evaluation_func: The function to evaluate combinations. Defaults to None. - filter_func: Allows the user to specify how agents are filtered to form groups. - Defaults to None. - List: The function to filter combinations. Defaults to None. - - Returns: - List: The list of valuable combinations, in a tuple first agentset of valuable combination and then the value of - the combination. + """Find candidate agent groups and score them with ``evaluation_func``. + + The helper is retained for existing examples that discover potential + meta-agents before creating them. It deliberately does not mutate model or + membership state. """ - combinations = [] - # Allow one size or range of sizes to be passed if isinstance(size, int): size_range = range(size, size + 1) else: min_size, max_size = size size_range = range(min_size, max_size + 1) + combinations = [] for candidate_group in itertools.chain.from_iterable( itertools.combinations(group, combination_size) for combination_size in size_range @@ -119,223 +122,207 @@ def find_combinations( candidate_group, model, evaluation_func ) if evaluation_result is not None: - evaluated_group, result = evaluation_result + _evaluated_group, result = evaluation_result if result is not None: - combinations.append((evaluated_group, result)) - - if len(combinations) > 0 and filter_func: - filtered_combinations = filter_func(combinations) - return filtered_combinations + combinations.append(evaluation_result) + if combinations and filter_func is not None: + return filter_func(combinations) return combinations def extract_class(agents_by_type: dict, new_agent_class: object) -> type[Agent] | None: - """Helper function for create_meta_agents extracts the types of agents. + """Return the existing model agent class named ``new_agent_class`` if present.""" + agent_type_names = { + agent_type.__name__: agent_type for agent_type in agents_by_type + } + agent_type = agent_type_names.get(new_agent_class) + if agent_type is None: + return None + return type(next(iter(agents_by_type[agent_type]))) - Args: - agents_by_type (dict): The dictionary of agents by type. - new_agent_class (str): The name of the agent class to be created - Returns: - type(Agent) if agent type exists - None otherwise - """ - agent_type_names = {} - for agent in agents_by_type: - agent_type_names[agent.__name__] = agent +def _collect_inferred_attributes( + agents: Iterable[Any], + meta_attributes: dict[str, Any] | None, + assume_constituting_agent_attributes: bool, +) -> dict[str, Any]: + """Merge explicit and inferred attributes for a new meta-agent instance.""" + resolved_attributes = dict(meta_attributes or {}) + if not assume_constituting_agent_attributes: + return resolved_attributes + + for agent in agents: + for name, value in agent.__dict__.items(): + if ( + not callable(value) + and name not in _RESERVED_META_ATTRIBUTE_NAMES + and not name.startswith("_") + ): + resolved_attributes[name] = value + return resolved_attributes + + +def _apply_meta_attributes( + meta_agent: Any, + meta_attributes: dict[str, Any] | None, +) -> None: + """Set resolved meta-agent attributes on an instance.""" + for key, value in (meta_attributes or {}).items(): + setattr(meta_agent, key, value) + + +def _collect_meta_methods( + agents: Iterable[Any], + meta_methods: dict[str, Callable] | None, + assume_constituting_agent_methods: bool, +) -> dict[str, Callable]: + """Merge explicit and inferred methods for a meta-agent instance.""" + resolved_meta_methods = dict(meta_methods or {}) + if not assume_constituting_agent_methods: + return resolved_meta_methods + + for agent_class in dict.fromkeys(type(agent) for agent in agents): + for name, value in agent_class.__dict__.items(): + if callable(value) and not name.startswith("__"): + resolved_meta_methods.setdefault(name, value) + return resolved_meta_methods + + +def _apply_meta_methods( + meta_agent: Any, + meta_methods: dict[str, Callable] | None, +) -> None: + """Bind resolved meta-agent methods to an instance.""" + for name, method in (meta_methods or {}).items(): + setattr(meta_agent, name, MethodType(method, meta_agent)) + + +def _find_existing_meta_agent( + agents: Iterable[Any], + new_agent_class: str, +) -> Any | None: + """Find a compatible existing meta-agent among legacy mirrors.""" + existing_meta_agents = [] + for agent in agents: + for meta_agent in sorted( + getattr(agent, "meta_agents", set()), key=_unique_id_sort_key + ): + if ( + meta_agent.__class__.__name__ == new_agent_class + and meta_agent not in existing_meta_agents + ): + existing_meta_agents.append(meta_agent) + + if not existing_meta_agents: + return None + return sorted(existing_meta_agents, key=_unique_id_sort_key)[0] + + +def _build_meta_agent_class( + new_agent_class: str, + mesa_agent_type: tuple[type[Agent], ...], +) -> type[Agent]: + """Create a dynamic meta-agent class with the requested Mesa base types.""" + return type( + new_agent_class, + (MetaAgent, *mesa_agent_type), + { + "unique_id": None, + "_constituting_set": None, + }, + ) + + +def _create_meta_agent_instance( + model: Any, + new_agent_class: str, + agents: Iterable[Any], + mesa_agent_type: type[Agent] | tuple[type[Agent], ...] | None, + meta_attributes: dict[str, Any] | None = None, + meta_methods: dict[str, Callable] | None = None, + assume_constituting_agent_methods: bool = False, + assume_constituting_agent_attributes: bool = False, +) -> Any | None: + """Create or reuse a meta-agent instance without recording backend edges.""" + agents = _deduplicate_preserving_order(agents) + agent_bases = _normalize_agent_bases(mesa_agent_type) + resolved_attributes = _collect_inferred_attributes( + agents, + meta_attributes, + assume_constituting_agent_attributes, + ) + resolved_methods = _collect_meta_methods( + agents, + meta_methods, + assume_constituting_agent_methods, + ) + + meta_agent = _find_existing_meta_agent(agents, new_agent_class) + if meta_agent is not None: + _apply_meta_attributes(meta_agent, resolved_attributes) + _apply_meta_methods(meta_agent, resolved_methods) + meta_agent.add_constituting_agents(agents) + return meta_agent - if new_agent_class in agent_type_names: - return type(next(iter(agents_by_type[agent_type_names[new_agent_class]]))) - return None + agent_class = extract_class(model.agents_by_type, new_agent_class) + if agent_class is None: + agent_class = _build_meta_agent_class(new_agent_class, agent_bases) + + meta_agent = agent_class( + model, + agents, + initial_attributes=resolved_attributes, + ) + _apply_meta_attributes(meta_agent, resolved_attributes) + _apply_meta_methods(meta_agent, resolved_methods) + return meta_agent def create_meta_agent( model: Any, new_agent_class: str, agents: Iterable[Any], - mesa_agent_type: type[Agent] | None, + mesa_agent_type: type[Agent] | tuple[type[Agent], ...] | None, meta_attributes: dict[str, Any] | None = None, meta_methods: dict[str, Callable] | None = None, assume_constituting_agent_methods: bool = False, assume_constituting_agent_attributes: bool = False, ) -> Any | None: - """Create a new meta-agent class and instantiate agents. - - Parameters: - model (Any): The model instance. - new_agent_class (str): The name of the new meta-agent class. - agents (Iterable[Any]): The agents to be included in the meta-agent. - meta_attributes (Dict[str, Any]): Attributes to be added to the meta-agent. - meta_methods (Dict[str, Callable]): Methods to be added to the meta-agent. - assume_constituting_agent_methods (bool): Whether to assume methods from - constituting_-agents as meta_agent methods. - assume_constituting_agent_attributes (bool): Whether to retain attributes - from constituting_-agents. - - Returns: - - MetaAgent Instance - """ - # Convert agents to dict, to ensure uniqueness, - # we need a dict, not a set to keep stuff deterministic - agents = list(dict.fromkeys(agents).keys()) - - # Ensure there is at least one agent base class - if not mesa_agent_type: - mesa_agent_type = (Agent,) - elif not isinstance(mesa_agent_type, tuple): - mesa_agent_type = (mesa_agent_type,) - - def add_methods( - meta_agent_instance: Any, - agents: Iterable[Any], - meta_methods: dict[str, Callable] | None, - ) -> None: - """Add methods to the meta-agent instance. - - Parameters: - meta_agent_instance (Any): The meta-agent instance. - agents (Iterable[Any]): The agents to derive methods from. - meta_methods (Dict[str, Callable]): methods to be added to the meta-agent. - """ - resolved_meta_methods = dict(meta_methods or {}) - if assume_constituting_agent_methods: - agent_classes = dict.fromkeys(type(agent) for agent in agents) - for agent_class in agent_classes: - for name in agent_class.__dict__: - if callable(getattr(agent_class, name)) and not name.startswith( - "__" - ): - original_method = getattr(agent_class, name) - resolved_meta_methods.setdefault(name, original_method) - - for name, meth in resolved_meta_methods.items(): - bound_method = MethodType(meth, meta_agent_instance) - setattr(meta_agent_instance, name, bound_method) - - def add_attributes( - meta_agent_instance: Any, - agents: Iterable[Any], - meta_attributes: dict[str, Any], - ) -> None: - """Add attributes to the meta-agent instance. - - Parameters: - meta_agent_instance (Any): The meta-agent instance. - agents (Iterable[Any]): The agents to derive attributes from. - meta_attributes (Dict[str, Any]): Attributes to be added to the - meta-agent. - """ - # Prevent collision of attributes with meta-agent instantiation - mesa_primitives = [ - "unique_id", - "model", - "pos", - "name", - "random", - "rng", - ] - - if assume_constituting_agent_attributes: - if meta_attributes is None: - # Initialize meta_attributes if not provided - meta_attributes = {} - for agent in agents: - for name, value in agent.__dict__.items(): - if ( - not callable(value) - and name not in mesa_primitives - and not name.startswith("_") - ): - meta_attributes[name] = value - - if meta_attributes is not None: - for key, value in meta_attributes.items(): - setattr(meta_agent_instance, key, value) - - # Path 1 - Add agents to existing meta-agent of the SAME CLASS if any exist - # This preserves the "singleton/unique group per class" behavior while allowing overlap between different classes - existing_meta_agents = [] - for a in agents: - if hasattr(a, "meta_agents"): - for ma in sorted(a.meta_agents, key=_unique_id_sort_key): - if ( - ma.__class__.__name__ == new_agent_class - and ma not in existing_meta_agents - ): - existing_meta_agents.append(ma) - - if len(existing_meta_agents) > 0: - # TODO: Add way for user to specify how agents join meta-agent - # instead of random choice if there are multiple meta-agents of the same class - meta_agent = ( - sorted(existing_meta_agents, key=_unique_id_sort_key)[0] - if len(existing_meta_agents) > 1 - else existing_meta_agents[0] - ) - add_attributes(meta_agent, agents, meta_attributes) - add_methods(meta_agent, agents, meta_methods) - meta_agent.add_constituting_agents(agents) - return meta_agent + """Legacy helper for creating a meta-agent instance. - else: - # Path 2 - Create a new instance of an existing meta-agent class - agent_class = extract_class(model.agents_by_type, new_agent_class) - - if agent_class: - # Pass initial_attributes to __init__ to handle CellAgent and other - # descriptor-based parent classes correctly (initialize before super().__init__()) - meta_agent_instance = agent_class( - model, agents, initial_attributes=meta_attributes - ) - # add_attributes() will handle inferred attributes if needed - add_attributes(meta_agent_instance, agents, meta_attributes) - add_methods(meta_agent_instance, agents, meta_methods) - return meta_agent_instance - else: - # Path 3 - Create a new meta-agent class - meta_agent_class = type( - new_agent_class, - (MetaAgent, *mesa_agent_type), # Inherit Mesa Agent Classes - { - "unique_id": None, - "_constituting_set": None, - }, - ) - # Pass initial_attributes to __init__ to handle CellAgent and other - # descriptor-based parent classes correctly (initialize before super().__init__()) - meta_agent_instance = meta_agent_class( - model, agents, initial_attributes=meta_attributes - ) - # add_attributes() will handle inferred attributes if needed - add_attributes(meta_agent_instance, agents, meta_attributes) - add_methods(meta_agent_instance, agents, meta_methods) - return meta_agent_instance + This function preserves the historical API and object-level compatibility + mirrors. It does not own canonical membership bookkeeping; use + ``MetaAgents.create`` when memberships should be recorded in a + ``MembershipBackend``. + """ + return _create_meta_agent_instance( + model, + new_agent_class, + agents, + mesa_agent_type, + meta_attributes=meta_attributes, + meta_methods=meta_methods, + assume_constituting_agent_methods=assume_constituting_agent_methods, + assume_constituting_agent_attributes=assume_constituting_agent_attributes, + ) class MetaAgent(Agent): - """A MetaAgent is an agent that contains other agents as components.""" + """An agent composed of other agents. + + ``MetaAgent`` keeps the live object relationship needed by existing models. + Canonical typed membership storage lives in the backend/facade layer. + """ def __init__( self, model, - agents: set[Agent] | None = None, + agents: Iterable[Agent] | None = None, name: str = "MetaAgent", initial_attributes: dict[str, Any] | None = None, ): - """Create a new MetaAgent. - - Args: - model: The model instance. - agents (Optional[set[Agent]], optional): The set of agents to - include in the MetaAgent. Defaults to None. - name (str, optional): The name of the MetaAgent. Defaults to "MetaAgent". - initial_attributes (Optional[dict], optional): Attributes to set before - calling super().__init__(). This is important for CellAgent and other - descriptor-based parent classes. Defaults to None. - """ - # Apply initial attributes BEFORE calling super().__init__() - # This is important for CellAgent and other descriptors + """Create a meta-agent from an optional iterable of component agents.""" if initial_attributes: for key, value in initial_attributes.items(): object.__setattr__(self, key, value) @@ -344,67 +331,41 @@ def __init__( self._constituting_set = AgentSet(agents or [], random=model.random) self.name = name - # Add ref to meta_agent in constituting_agents for agent in self._constituting_set: - if not hasattr(agent, "meta_agents"): - agent.meta_agents = set() - agent.meta_agents.add(self) - # Maintain backward compatibility — always pick lowest unique_id - agent.meta_agent = sorted(agent.meta_agents, key=_unique_id_sort_key)[0] + _attach_meta_agent(agent, self) def __len__(self) -> int: - """Return the number of components.""" + """Return the number of component agents.""" return len(self._constituting_set) def __iter__(self): - """Iterate over components.""" + """Iterate over component agents.""" return iter(self._constituting_set) def __contains__(self, agent: Agent) -> bool: - """Check if an agent is a component.""" + """Return whether ``agent`` is a component of this meta-agent.""" return agent in self._constituting_set @property def agents(self) -> AgentSet: - """Get list of Meta-Agent constituting_agents.""" + """Return the component agents.""" return self._constituting_set @property def constituting_agents_by_type(self) -> dict[type, list[Agent]]: - """Get the constituting_agents grouped by type. - - Returns: - dict[type, list[Agent]]: A dictionary of constituting_agents grouped by type. - """ + """Return component agents grouped by their concrete Python type.""" constituting_agents_by_type = {} for agent in self._constituting_set: - agent_type = type(agent) - if agent_type not in constituting_agents_by_type: - constituting_agents_by_type[agent_type] = [] - constituting_agents_by_type[agent_type].append(agent) + constituting_agents_by_type.setdefault(type(agent), []).append(agent) return constituting_agents_by_type @property def constituting_agent_types(self) -> set[type]: - """Get the types of all constituting_agents. - - Returns: - set[type]: A set of unique types of the constituting_agents. - """ + """Return the set of component agent types.""" return {type(agent) for agent in self._constituting_set} def get_constituting_agent_instance(self, agent_type) -> Agent: - """Get the instance of a constituting_agent of the specified type. - - Args: - agent_type: The type of the constituting_agent to retrieve. - - Returns: - The first instance of the specified constituting_agent type. - - Raises: - ValueError: If no constituting_agent of the specified type is found. - """ + """Return the first component agent of ``agent_type``.""" try: return self.constituting_agents_by_type[agent_type][0] except KeyError: @@ -412,53 +373,31 @@ def get_constituting_agent_instance(self, agent_type) -> Agent: f"No constituting_agent of type {agent_type} found." ) from None - def add_constituting_agents( - self, - new_agents: set[Agent], - ): - """Add agents as components. - - Args: - new_agents (set[Agent]): The agents to add to MetaAgent constituting_set. - """ + def add_constituting_agents(self, new_agents: Iterable[Agent]) -> None: + """Add component agents and update legacy compatibility mirrors.""" for agent in new_agents: self._constituting_set.add(agent) - if not hasattr(agent, "meta_agents"): - agent.meta_agents = set() - agent.meta_agents.add(self) - # Maintain backward compatibility — always pick lowest unique_id - agent.meta_agent = sorted(agent.meta_agents, key=_unique_id_sort_key)[0] - - def remove_constituting_agents(self, remove_agents: set[Agent]): - """Remove agents as components. - - Args: - remove_agents (set[Agent]): The agents to remove. - """ + _attach_meta_agent(agent, self) + + def remove_constituting_agents(self, remove_agents: Iterable[Agent]) -> None: + """Remove component agents and update legacy compatibility mirrors.""" for agent in remove_agents: self._constituting_set.discard(agent) - if hasattr(agent, "meta_agents"): - agent.meta_agents.discard(self) - # Update backward compatibility attribute deterministically - if len(agent.meta_agents) > 0: - agent.meta_agent = sorted( - agent.meta_agents, key=_unique_id_sort_key - )[0] - else: - agent.meta_agent = None + _detach_meta_agent(agent, self) def remove(self) -> None: - """Remove the MetaAgent from the model and clean up constituent references. - - Clears ``meta_agents`` and ``meta_agent`` on every constituent agent - before deregistering so no stale references remain. - """ + """Remove this meta-agent and clear live references from components.""" self.remove_constituting_agents(set(self._constituting_set)) super().remove() - def step(self): - """Perform the agent's step. + def step(self) -> None: + """Default meta-agent behavior.""" + - Override this method to define the meta agent's behavior. - By default, does nothing. - """ +__all__ = [ + "MetaAgent", + "create_meta_agent", + "evaluate_combination", + "extract_class", + "find_combinations", +] diff --git a/mesa/experimental/meta_agents/meta_agents_api.py b/mesa/experimental/meta_agents/meta_agents_api.py new file mode 100644 index 00000000000..4736e9e7f12 --- /dev/null +++ b/mesa/experimental/meta_agents/meta_agents_api.py @@ -0,0 +1,233 @@ +"""Public meta-agents API for the experimental membership backend.""" + +from __future__ import annotations + +from collections.abc import Callable, Hashable, Iterable +from dataclasses import dataclass +from typing import Any + +from mesa.agent import Agent + +from .backend import MembershipBackend, RelationKey, Triplet +from .meta_agent import _create_meta_agent_instance + + +@dataclass(frozen=True, slots=True) +class MembershipEdge: + """A user-facing membership edge with live objects instead of backend ids.""" + + agent: Any + group: Any + relation: RelationKey + + +@dataclass(frozen=True, slots=True) +class MembershipView: + """Read-only snapshot of memberships for one entity.""" + + subject: Any + memberships: tuple[MembershipEdge, ...] + + def __iter__(self): + """Iterate over the resolved memberships.""" + return iter(self.memberships) + + def __len__(self) -> int: + """Return the number of resolved memberships.""" + return len(self.memberships) + + @property + def edges(self) -> tuple[MembershipEdge, ...]: + """Alias for ``memberships`` to keep the view easy to inspect.""" + return self.memberships + + def as_triplets(self) -> set[tuple[Any, Any, RelationKey]]: + """Return the memberships as live-object triplets.""" + return {(edge.agent, edge.group, edge.relation) for edge in self.memberships} + + @property + def agents(self) -> set[Any]: + """Return all unique agents referenced by the view.""" + return {edge.agent for edge in self.memberships} + + @property + def groups(self) -> set[Any]: + """Return all unique groups referenced by the view.""" + return {edge.group for edge in self.memberships} + + @property + def relations(self) -> set[RelationKey]: + """Return all unique relation labels referenced by the view.""" + return {edge.relation for edge in self.memberships} + + +class MetaAgents: + """Public meta-agents interface over :class:`MembershipBackend`.""" + + def __init__(self, model: Any, backend: MembershipBackend | None = None) -> None: + """Create a meta-agents API bound to one model.""" + self.model = model + self.backend = backend or MembershipBackend() + + def _entity_id(self, entity: Hashable) -> Hashable: + """Return the backend identity for a live entity or hashable external id.""" + return getattr(entity, "unique_id", entity) + + def _live_entity_lookup(self) -> dict[Hashable, Any]: + """Build a lookup from backend ids back to live model objects.""" + lookup: dict[Hashable, Any] = {} + for entity in self.model.agents: + entity_id = getattr(entity, "unique_id", None) + if entity_id is not None: + lookup[entity_id] = entity + return lookup + + def _resolve_entity(self, entity_id: Hashable) -> Any: + """Resolve a backend id back to a live object when possible.""" + return self._live_entity_lookup().get(entity_id, entity_id) + + def _resolve_view( + self, entity: Hashable, triplets: Iterable[Triplet] + ) -> MembershipView: + """Convert backend triplets into a user-facing snapshot.""" + lookup = self._live_entity_lookup() + resolved_edges: list[MembershipEdge] = [] + for agent_id, group_id, relation in sorted( + triplets, + key=lambda triplet: ( + str(triplet[0]), + str(triplet[1]), + repr(triplet[2]), + ), + ): + resolved_edges.append( + MembershipEdge( + agent=lookup.get(agent_id, agent_id), + group=lookup.get(group_id, group_id), + relation=relation, + ) + ) + + return MembershipView( + subject=self._resolve_entity(entity), + memberships=tuple(resolved_edges), + ) + + def _detach_entity(self, entity: Hashable) -> MembershipView: + """Remove all incident memberships and update live objects when available.""" + snapshot = self.query_memberships(entity) + + self.backend.remove_agent(entity) + self.backend.remove_group(entity) + + for edge in snapshot.memberships: + group = edge.group + member = edge.agent + if hasattr(group, "remove_constituting_agents"): + group.remove_constituting_agents({member}) + + return snapshot + + def create( + self, + new_agent_class: str, + agents: Iterable[Any], + mesa_agent_type: type[Agent] | None, + meta_attributes: dict[str, Any] | None = None, + meta_methods: dict[str, Callable] | None = None, + assume_constituting_agent_methods: bool = False, + assume_constituting_agent_attributes: bool = False, + relation: RelationKey = "member", + memberships: Iterable[tuple[Any, RelationKey]] | None = None, + ) -> Any | None: + """Create a meta-agent and record its memberships in the backend.""" + agents = list(agents) + member_relations = list(memberships) if memberships is not None else None + if member_relations is not None and not agents: + agents = [member for member, _ in member_relations] + + meta_agent = _create_meta_agent_instance( + self.model, + new_agent_class, + agents, + mesa_agent_type, + meta_attributes=meta_attributes, + meta_methods=meta_methods, + assume_constituting_agent_methods=assume_constituting_agent_methods, + assume_constituting_agent_attributes=assume_constituting_agent_attributes, + ) + + if meta_agent is None: + return None + + if member_relations is None: + member_relations = [(agent, relation) for agent in agents] + + self.backend.bulk_add( + [(member, meta_agent, rel) for member, rel in member_relations] + ) + + return meta_agent + + def add_member( + self, + group: Hashable, + member: Hashable, + relation: RelationKey = "member", + ) -> MembershipView: + """Add one member to one group and keep the object layer in sync.""" + already_linked = bool(self.backend.relations_between(member, group)) + self.backend.add_membership(member, group, relation) + + if not already_linked and hasattr(group, "add_constituting_agents"): + group.add_constituting_agents({member}) + + return self.query_memberships(member) + + def remove_member( + self, + group: Hashable, + member: Hashable, + relation: RelationKey = "member", + ) -> MembershipView: + """Remove one member from one group and keep the object layer in sync.""" + self.backend.remove_membership(member, group, relation) + + if not self.backend.relations_between(member, group) and hasattr( + group, "remove_constituting_agents" + ): + group.remove_constituting_agents({member}) + + return self.query_memberships(member) + + def query_memberships( + self, entity: Hashable, relation: RelationKey | None = None + ) -> MembershipView: + """Return a resolved, read-only snapshot of one entity's memberships.""" + entity_id = self._entity_id(entity) + triplets = ( + triplet + for triplet in self.backend.as_triplets() + if (triplet[0] == entity_id or triplet[1] == entity_id) + and (relation is None or triplet[2] == relation) + ) + return self._resolve_view(entity_id, triplets) + + def dissolve(self, entity: Hashable) -> MembershipView: + """Remove an entity's memberships and delete it from the model when possible.""" + snapshot = self._detach_entity(entity) + live_entity = self._resolve_entity(self._entity_id(entity)) + if hasattr(live_entity, "remove"): + live_entity.remove() + return snapshot + + def deactivate(self, entity: Hashable) -> MembershipView: + """Remove an entity from all memberships without deleting the object.""" + return self._detach_entity(entity) + + +__all__ = [ + "MembershipEdge", + "MembershipView", + "MetaAgents", +] diff --git a/tests/examples/test_alliance_formation_model.py b/tests/examples/test_alliance_formation_model.py new file mode 100644 index 00000000000..3b49269db4c --- /dev/null +++ b/tests/examples/test_alliance_formation_model.py @@ -0,0 +1,45 @@ +"""Tests for the allianceformation meta-agent example.""" + +from __future__ import annotations + +from mesa.examples.advanced.alliance_formation.model import ( + AllianceScenario, + MultiLevelAllianceModel, +) + + +def test_alliance_model_records_overlapping_memberships(monkeypatch): + """The backend should preserve overlap when an agent joins mutiple alliances.""" + model = MultiLevelAllianceModel( + scenario=AllianceScenario(n=3, mean=0.5, std_dev=0.0, rng=42) + ) + agents = sorted(model.agents, key=lambda agent: agent.unique_id) + agent_0, agent_1, agent_2 = agents + + def fake_find_combinations(*args, **kwargs): + return [ + ((agent_0, agent_1), (1.0, 0.5, 0)), + ((agent_0, agent_2), (1.0, 0.4, 0)), + ] + + monkeypatch.setattr( + "mesa.examples.advanced.alliance_formation.model.find_combinations", + fake_find_combinations, + ) + + model.step() + + backend = model.membership_backend + + assert len(agent_0.meta_agents) == 2 + assert backend.groups_of(agent_0) == { + meta.unique_id for meta in agent_0.meta_agents + } + + expected_triplets = set() + for agent in agents: + for meta in agent.meta_agents: + expected_triplets.add((agent.unique_id, meta.unique_id, "member")) + + assert backend.as_triplets() == expected_triplets + backend.assert_invariants() diff --git a/tests/experimental/test_meta_agents_api.py b/tests/experimental/test_meta_agents_api.py new file mode 100644 index 00000000000..863d3abae74 --- /dev/null +++ b/tests/experimental/test_meta_agents_api.py @@ -0,0 +1,104 @@ +"""Tests for the public meta-agents API.""" + +from mesa import Agent, Model +from mesa.experimental.meta_agents import ( + MembershipEdge, + MembershipView, + MetaAgents, +) + + +def test_meta_agents_create_records_backend_memberships(): + """Create should return live objects and record backend triplets.""" + model = Model() + meta_agents = MetaAgents(model) + agent_1 = Agent(model) + agent_2 = Agent(model) + + meta_agent = meta_agents.create("Group", [agent_1, agent_2], Agent) + + assert meta_agent is not None + assert meta_agents.backend.as_triplets() == { + (agent_1.unique_id, meta_agent.unique_id, "member"), + (agent_2.unique_id, meta_agent.unique_id, "member"), + } + + view = meta_agents.query_memberships(agent_1) + + assert isinstance(view, MembershipView) + assert view.subject is agent_1 + assert view.as_triplets() == {(agent_1, meta_agent, "member")} + assert len(view) == 1 + assert isinstance(view.memberships[0], MembershipEdge) + assert view.memberships[0].agent is agent_1 + assert view.memberships[0].group is meta_agent + + +def test_meta_agents_remove_member_preserves_overlapping_memberships(): + """Removing one relation should keep unrelated memberships intact.""" + model = Model() + meta_agents = MetaAgents(model) + agent = Agent(model) + partner = Agent(model) + group_one = meta_agents.create("GroupOne", [agent, partner], Agent) + group_two = meta_agents.create("GroupTwo", [agent], Agent) + + assert group_one is not None + assert group_two is not None + assert len(agent.meta_agents) == 2 + + view = meta_agents.remove_member(group_one, agent) + + assert view.as_triplets() == {(agent, group_two, "member")} + assert meta_agents.backend.groups_of(agent) == {group_two.unique_id} + assert group_one not in agent.meta_agents + assert group_two in agent.meta_agents + assert partner.meta_agents == {group_one} + + +def test_meta_agents_dissolve_cleans_only_target_group(): + """Dissolving a group should keep overlapping memberships on other groups.""" + model = Model() + meta_agents = MetaAgents(model) + agent_1 = Agent(model) + agent_2 = Agent(model) + agent_3 = Agent(model) + group_one = meta_agents.create("GroupOne", [agent_1, agent_2], Agent) + group_two = meta_agents.create("GroupTwo", [agent_1, agent_3], Agent) + + assert group_one is not None + assert group_two is not None + + snapshot = meta_agents.dissolve(group_one) + + assert snapshot.as_triplets() == { + (agent_1, group_one, "member"), + (agent_2, group_one, "member"), + } + assert meta_agents.backend.groups_of(agent_1) == {group_two.unique_id} + assert meta_agents.backend.groups_of(agent_2) == set() + assert meta_agents.backend.groups_of(agent_3) == {group_two.unique_id} + assert group_one not in model.agents + assert group_two in model.agents + assert group_one not in agent_1.meta_agents + assert group_two in agent_1.meta_agents + + +def test_meta_agents_deactivate_detaches_all_memberships_without_removing_entity(): + """Deactivate should clear memberships but keep the entity registered.""" + model = Model() + meta_agents = MetaAgents(model) + agent_1 = Agent(model) + agent_2 = Agent(model) + group = meta_agents.create("Group", [agent_1, agent_2], Agent) + + assert group is not None + + snapshot = meta_agents.deactivate(agent_1) + + assert snapshot.as_triplets() == {(agent_1, group, "member")} + assert meta_agents.backend.groups_of(agent_1) == set() + assert agent_1 in model.agents + assert group not in agent_1.meta_agents + assert agent_1.meta_agent is None + assert group in agent_2.meta_agents diff --git a/tests/experimental/test_meta_agents_backend.py b/tests/experimental/test_meta_agents_backend.py new file mode 100644 index 00000000000..1027b326d76 --- /dev/null +++ b/tests/experimental/test_meta_agents_backend.py @@ -0,0 +1,110 @@ +"""Tests for typed membership backend.""" + +from mesa import Agent, Model +from mesa.experimental.meta_agents.backend import MembershipBackend +from mesa.experimental.meta_agents.meta_agent import MetaAgent + + +def test_add_and_query(): + """Add edges and verify basic query behavior.""" + backend = MembershipBackend() + backend.add_membership("a1", "g1", "member") + backend.add_membership("a2", "g1", "member") + backend.add_membership("a1", "g2", "leader") + + assert backend.groups_of("a1") == {"g1", "g2"} + assert backend.groups_of("a1", relation="member") == {"g1"} + assert backend.agents_of("g1") == {"a1", "a2"} + assert backend.relations_between("a1", "g1") == {"member"} + backend.assert_invariants() + + +def test_typed_overlap_same_pair(): + """Allow multiple relation labels on the same agent-group pair.""" + backend = MembershipBackend() + backend.add_membership("a1", "g1", "member") + backend.add_membership("a1", "g1", "mentor") + + assert backend.relations_between("a1", "g1") == {"member", "mentor"} + assert backend.groups_of("a1", relation="mentor") == {"g1"} + backend.assert_invariants() + + +def test_idempotent_add_and_remove(): + """Repeated add/remove call should remain safe and deterministic.""" + backend = MembershipBackend() + backend.add_membership("a1", "g1", "member") + backend.add_membership("a1", "g1", "member") # idempotent add + + assert backend.as_triplets() == {("a1", "g1", "member")} + + backend.remove_membership("a1", "g1", "member") + backend.remove_membership("a1", "g1", "member") # idempotent remove + assert backend.as_triplets() == set() + backend.assert_invariants() + + +def test_replace_relation(): + """Replace an existing relation label for one edge.""" + backend = MembershipBackend() + backend.add_membership("a1", "g1", "member") + backend.replace_relation("a1", "g1", "member", "leader") + + assert backend.relations_between("a1", "g1") == {"leader"} + assert backend.groups_of("a1", relation="member") == set() + backend.assert_invariants() + + +def test_remove_agent_cascades_edges(): + """Removing an agent should clear all its incident edges.""" + backend = MembershipBackend() + backend.bulk_add( + [("a1", "g1", "member"), ("a1", "g2", "leader"), ("a2", "g1", "member")] + ) + + backend.remove_agent("a1") + + assert backend.groups_of("a1") == set() + assert backend.agents_of("g1") == {"a2"} + assert backend.agents_of("g2") == set() + backend.assert_invariants() + + +def test_remove_group_cascades_edges(): + """Removing a group should clear all incident edges.""" + backend = MembershipBackend() + backend.bulk_add( + [("a1", "g1", "member"), ("a1", "g2", "leader"), ("a2", "g1", "member")] + ) + + backend.remove_group("g1") + + assert backend.agents_of("g1") == set() + assert backend.groups_of("a1") == {"g2"} + assert backend.groups_of("a2") == set() + backend.assert_invariants() + + +def test_non_string_relation_key(): + """Allow non-string hashable relation keys.""" + backend = MembershipBackend() + rel = ("role", 1) + backend.add_membership("a1", "g1", rel) + + assert backend.relations_between("a1", "g1") == {rel} + backend.assert_invariants() + + +def test_backend_uses_unique_ids_for_mesa_agents(): + """Meta-agent membership bookkeeping should use unique_id values.""" + model = Model() + agent = Agent(model) + meta_agent = MetaAgent(model, {agent}, name="Group") + backend = MembershipBackend() + + backend.add_membership(agent, meta_agent, "member") + + assert backend.as_triplets() == {(agent.unique_id, meta_agent.unique_id, "member")} + assert backend.groups_of(agent) == {meta_agent.unique_id} + assert backend.agents_of(meta_agent) == {agent.unique_id} + backend.assert_invariants()