"""Internal meta-agent group object.
Meta-agents are agents composed of other agents. Memberships are tracked by
the model's membership manager (``model.meta_agents``). This module holds the
internal group ``Agent`` subclass.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable
from types import MethodType
from typing import Any, cast
from modelly.core.agent import Agent
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(
modelly_agent_type: type[Agent] | tuple[type[Agent], ...] | None,
) -> tuple[type[Agent], ...]:
"""Normalize user-provided Modelly base classes for dynamic meta-agent classes."""
if modelly_agent_type is None:
return (Agent,)
if isinstance(modelly_agent_type, tuple):
return modelly_agent_type
return (modelly_agent_type,)
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 _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,
membership_api: Any,
) -> MetaAgent | None:
"""Find a compatible existing meta-agent among an agent's current groups."""
existing_meta_agents = []
for agent in agents:
for meta_agent in sorted(
membership_api.groups_of(agent), 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,
modelly_agent_type: tuple[type[Agent], ...],
) -> type[MetaAgent]:
"""Create a dynamic meta-agent class with the requested Modelly base types."""
agent_class = type(
new_agent_class,
(MetaAgent, *modelly_agent_type),
{
"unique_id": None,
},
)
return cast("type[MetaAgent]", agent_class)
def _resolve_meta_agent_class(
model: Any,
new_agent_class: str,
agent_bases: tuple[type[Agent], ...],
) -> type[MetaAgent]:
"""Return the model's class named ``new_agent_class``, or build a new one.
A class found by name is only constructible here if it is a ``MetaAgent``
subclass (other classes reject ``_membership_api``), hence the cast.
"""
existing_class = extract_class(model.agents_by_type, new_agent_class)
if existing_class is None:
return _build_meta_agent_class(new_agent_class, agent_bases)
return cast("type[MetaAgent]", existing_class)
def _reuse_meta_agent(
meta_agent: MetaAgent,
membership_api: Any,
meta_attributes: dict[str, Any] | None,
meta_methods: dict[str, Callable] | None,
) -> MetaAgent:
"""Rebind an existing meta-agent to the manager and apply attributes and methods."""
existing_api = getattr(meta_agent, "_membership_api", None)
if existing_api not in (None, membership_api):
raise RuntimeError("Meta-agent is bound to a different membership manager")
meta_agent._membership_api = membership_api
_apply_meta_attributes(meta_agent, meta_attributes)
_apply_meta_methods(meta_agent, meta_methods)
return meta_agent
def _create_meta_agent_instance(
model: Any,
new_agent_class: str,
agents: Iterable[Any],
modelly_agent_type: type[Agent] | tuple[type[Agent], ...] | None,
meta_attributes: dict[str, Any] | None = None,
meta_methods: dict[str, Callable] | None = None,
_membership_api: Any | None = None,
) -> MetaAgent:
"""Create or reuse a meta-agent instance without recording backend edges."""
if _membership_api is None:
raise RuntimeError("Use model.meta_agents.create() to create a meta-agent")
agents = _deduplicate_preserving_order(agents)
agent_bases = _normalize_agent_bases(modelly_agent_type)
meta_agent = _find_existing_meta_agent(agents, new_agent_class, _membership_api)
if meta_agent is not None:
return _reuse_meta_agent(
meta_agent, _membership_api, meta_attributes, meta_methods
)
agent_class = _resolve_meta_agent_class(model, new_agent_class, agent_bases)
meta_agent = agent_class(
model,
name=new_agent_class,
initial_attributes=meta_attributes,
_membership_api=_membership_api,
)
_apply_meta_attributes(meta_agent, meta_attributes)
_apply_meta_methods(meta_agent, meta_methods)
return meta_agent
class MetaAgent(Agent):
"""Internal agent subclass used as a live group object.
Construct instances through ``model.meta_agents.create``. Membership
reads and writes go through the membership manager, not this class.
"""
def __init__(
self,
model,
name: str = "MetaAgent",
initial_attributes: dict[str, Any] | None = None,
_membership_api: Any | None = None,
):
"""Create a meta-agent group bound to the model's membership manager."""
if _membership_api is None:
raise RuntimeError("Use model.meta_agents.create() to create a meta-agent")
installed_api = getattr(model, "meta_agents", None)
if installed_api is not _membership_api:
raise RuntimeError(
"Meta-agent must be created by its model's membership manager"
)
if initial_attributes:
for key, value in initial_attributes.items():
object.__setattr__(self, key, value)
super().__init__(model)
self._membership_api = _membership_api
self.name = name
def _remove_from_model(self) -> None:
"""Deregister this group without going through dissolve again."""
super().remove()
def remove(self) -> None:
"""Dissolve this group through the membership manager."""
self._membership_api.dissolve(self)
def step(self) -> None:
"""Default meta-agent behavior."""
__all__ = [
"extract_class",
]
