Skip to content

Constrained Markov Game (CMG)

masa.common.constraints.multi_agent.cmg.Budget dataclass 

Budget(amount: float, agents: tuple[str, ...], name: str | None = None)

Shared cumulative-cost budget over a subset of agents.

Parameters:

Name Type Description Default
amount float

Maximum allowed cumulative cost for this budget.

 required
agents tuple[str, ...]

Subset of agents from env.possible_agents covered by the budget.

 required
name str | None

Optional metric prefix. If omitted, a generated name is used.

 None
Notes

Agent memberships are deduplicated while preserving order. Budgets may overlap, so a single agent may contribute to more than one budget.

amount instance-attribute 

amount: float

agents instance-attribute 

agents: tuple[str, ...]

name class-attribute instance-attribute 

name: str | None = None

__post_init__ 

__post_init__()
Source code in masa/common/constraints/multi_agent/cmg.py 
56
57
58
59
60
61
def __post_init__(self):
    agents = tuple(dict.fromkeys(self.agents))
    if not agents:
        raise ValueError("Budget agents must be non-empty.")
    object.__setattr__(self, "amount", float(self.amount))
    object.__setattr__(self, "agents", agents)

masa.common.constraints.multi_agent.cmg.ConstrainedMarkovGame 

ConstrainedMarkovGame(possible_agents: Sequence[str], budgets: Sequence[Budget], cost_fn: CostFn = dummy_cost_fn)

Cumulative-cost monitor for a labelled parallel PettingZoo environment.

Source code in masa/common/constraints/multi_agent/cmg.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    possible_agents: Sequence[str],
    budgets: Sequence[Budget],
    cost_fn: CostFn = dummy_cost_fn,
):
    self.possible_agents = tuple(possible_agents)
    self._possible_agent_set = set(self.possible_agents)
    self.budgets = tuple(budgets)
    self.cost_fn = cost_fn

    if not self.possible_agents:
        raise ValueError("possible_agents must be non-empty.")
    if not self.budgets:
        raise ValueError("budgets must be non-empty.")

    self._budget_keys = []
    seen_budget_keys = set()
    for index, budget in enumerate(self.budgets):
        invalid_agents = set(budget.agents) - self._possible_agent_set
        if invalid_agents:
            raise ValueError(
                f"Budget agents must exist in possible_agents. Invalid agents: {sorted(invalid_agents)}"
            )
        key = budget.name or f"budget_{index}"
        if key in seen_budget_keys:
            raise ValueError(f"Budget names must be unique. Duplicate name: {key}")
        seen_budget_keys.add(key)
        self._budget_keys.append(key)
    self._budget_keys = tuple(self._budget_keys)

    self.reset()

possible_agents instance-attribute 

possible_agents = tuple(possible_agents)

_possible_agent_set instance-attribute 

_possible_agent_set = set(self.possible_agents)

budgets instance-attribute 

budgets = tuple(budgets)

cost_fn instance-attribute 

cost_fn = cost_fn

_budget_keys instance-attribute 

_budget_keys = tuple(self._budget_keys)

constraint_type property 

constraint_type: str

reset 

reset()

Reset per-agent and per-budget cumulative costs for a new episode.

Source code in masa/common/constraints/multi_agent/cmg.py 
100
101
102
103
104
105
def reset(self):
    """Reset per-agent and per-budget cumulative costs for a new episode."""
    self.agent_step_costs = {agent: 0.0 for agent in self.possible_agents}
    self.agent_totals = {agent: 0.0 for agent in self.possible_agents}
    self.budget_step_costs = {key: 0.0 for key in self._budget_keys}
    self.budget_totals = {key: 0.0 for key in self._budget_keys}

update 

update(labels_by_agent: Mapping[str, Iterable[str]])

Update the monitor from a mapping of agent ids to active labels.

Source code in masa/common/constraints/multi_agent/cmg.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def update(self, labels_by_agent: Mapping[str, Iterable[str]]):
    """Update the monitor from a mapping of agent ids to active labels."""
    unknown_agents = set(labels_by_agent) - self._possible_agent_set
    if unknown_agents:
        raise ValueError(f"Unknown agents in labels_by_agent: {sorted(unknown_agents)}")

    for agent in self.possible_agents:
        labels = labels_by_agent.get(agent, set())
        if not isinstance(labels, (set, frozenset)):
            raise ValueError(
                f"Expected labels for agent '{agent}' to be a set of atomic propositions, "
                f"got {type(labels).__name__}"
            )
        step_cost = float(self.cost_fn(labels))
        self.agent_step_costs[agent] = step_cost
        self.agent_totals[agent] += step_cost

    for key, budget in zip(self._budget_keys, self.budgets):
        step_cost = float(sum(self.agent_step_costs[agent] for agent in budget.agents))
        self.budget_step_costs[key] = step_cost
        self.budget_totals[key] += step_cost

satisfied 

satisfied() -> bool

Return True when every budget remains within its cap.

Source code in masa/common/constraints/multi_agent/cmg.py 
129
130
131
132
133
134
def satisfied(self) -> bool:
    """Return ``True`` when every budget remains within its cap."""
    return all(
        self.budget_totals[key] <= budget.amount
        for key, budget in zip(self._budget_keys, self.budgets)
    )

step_metric 

step_metric() -> dict[str, float]

Return per-step metrics for agents and budgets.

Source code in masa/common/constraints/multi_agent/cmg.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
def step_metric(self) -> dict[str, float]:
    """Return per-step metrics for agents and budgets."""
    metrics: dict[str, float] = {}
    for agent in self.possible_agents:
        metrics[f"{agent}_cost"] = self.agent_step_costs[agent]
        metrics[f"{agent}_violation"] = float(self.agent_step_costs[agent] >= 0.5)
        metrics[f"{agent}_cum_cost"] = self.agent_totals[agent]
    for key, budget in zip(self._budget_keys, self.budgets):
        metrics[f"{key}_cost"] = self.budget_step_costs[key]
        metrics[f"{key}_cum_cost"] = self.budget_totals[key]
        metrics[f"{key}_satisfied"] = float(self.budget_totals[key] <= budget.amount)
    metrics["satisfied"] = float(self.satisfied())
    return metrics

episode_metric 

episode_metric() -> dict[str, float]

Return end-of-episode cumulative metrics for agents and budgets.

Source code in masa/common/constraints/multi_agent/cmg.py 
150
151
152
153
154
155
156
157
158
159
def episode_metric(self) -> dict[str, float]:
    """Return end-of-episode cumulative metrics for agents and budgets."""
    metrics: dict[str, float] = {}
    for agent in self.possible_agents:
        metrics[f"{agent}_cum_cost"] = self.agent_totals[agent]
    for key, budget in zip(self._budget_keys, self.budgets):
        metrics[f"{key}_cum_cost"] = self.budget_totals[key]
        metrics[f"{key}_satisfied"] = float(self.budget_totals[key] <= budget.amount)
    metrics["satisfied"] = float(self.satisfied())
    return metrics

masa.common.constraints.multi_agent.cmg.ConstrainedMarkovGameEnv 

ConstrainedMarkovGameEnv(env: ParallelEnv, budgets: Sequence[Budget], cost_fn: CostFn = dummy_cost_fn, **kw: Any)

Bases: ParallelEnv

PettingZoo parallel wrapper that updates a ConstrainedMarkovGame.

Source code in masa/common/constraints/multi_agent/cmg.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def __init__(
    self,
    env: ParallelEnv,
    budgets: Sequence[Budget],
    cost_fn: CostFn = dummy_cost_fn,
    **kw: Any,
):
    if not isinstance(env, LabelledParallelEnv):
        raise TypeError(
            f"{self.__class__.__name__} must wrap a LabelledParallelEnv, but got {type(env).__name__}."
        )
    self.env = env
    self.metadata = getattr(env, "metadata", {})
    self.possible_agents = tuple(env.possible_agents)
    self.agents = list(getattr(env, "agents", self.possible_agents))
    self.label_fn = getattr(env, "label_fn", None)
    self.cost_fn = cost_fn
    self._constraint = ConstrainedMarkovGame(
        possible_agents=self.possible_agents,
        budgets=budgets,
        cost_fn=cost_fn,
    )

env instance-attribute 

env = env

metadata instance-attribute 

metadata = getattr(env, 'metadata', {})

possible_agents instance-attribute 

possible_agents = tuple(env.possible_agents)

agents instance-attribute 

agents = list(getattr(env, 'agents', self.possible_agents))

label_fn instance-attribute 

label_fn = getattr(env, 'label_fn', None)

cost_fn instance-attribute 

cost_fn = cost_fn

_constraint instance-attribute 

_constraint = ConstrainedMarkovGame(possible_agents=self.possible_agents, budgets=budgets, cost_fn=cost_fn)

budgets property 

budgets: tuple[Budget, ...]

Return the immutable CMG budget definitions used by this wrapper.

constraint_type property 

constraint_type: str

__getattr__ 

__getattr__(name: str)
Source code in masa/common/constraints/multi_agent/cmg.py 
192
193
def __getattr__(self, name: str):
    return getattr(self.env, name)

reset 

reset(seed: int | None = None, options: dict[str, Any] | None = None)

Reset the wrapped env and seed the constraint from initial agent labels.

Source code in masa/common/constraints/multi_agent/cmg.py 
195
196
197
198
199
200
201
def reset(self, seed: int | None = None, options: dict[str, Any] | None = None):
    """Reset the wrapped env and seed the constraint from initial agent labels."""
    obs, infos = self.env.reset(seed=seed, options=options)
    self.agents = list(getattr(self.env, "agents", self.possible_agents))
    self._constraint.reset()
    self._constraint.update(self._labels_by_agent(infos))
    return obs, infos

step 

step(actions)

Step the wrapped env and update the constraint from per-agent labels.

Source code in masa/common/constraints/multi_agent/cmg.py 
203
204
205
206
207
208
def step(self, actions):
    """Step the wrapped env and update the constraint from per-agent labels."""
    obs, rewards, terminations, truncations, infos = self.env.step(actions)
    self.agents = list(getattr(self.env, "agents", self.possible_agents))
    self._constraint.update(self._labels_by_agent(infos))
    return obs, rewards, terminations, truncations, infos

state 

state()
Source code in masa/common/constraints/multi_agent/cmg.py 
210
211
def state(self):
    return self.env.state()

render 

render()
Source code in masa/common/constraints/multi_agent/cmg.py 
213
214
def render(self):
    return self.env.render()

close 

close()
Source code in masa/common/constraints/multi_agent/cmg.py 
216
217
def close(self):
    return self.env.close()

observation_space 

observation_space(agent)
Source code in masa/common/constraints/multi_agent/cmg.py 
219
220
def observation_space(self, agent):
    return self.env.observation_space(agent)

action_space 

action_space(agent)
Source code in masa/common/constraints/multi_agent/cmg.py 
222
223
def action_space(self, agent):
    return self.env.action_space(agent)

constraint_step_metrics 

constraint_step_metrics() -> dict[str, float]
Source code in masa/common/constraints/multi_agent/cmg.py 
225
226
def constraint_step_metrics(self) -> dict[str, float]:
    return self._constraint.step_metric()

constraint_episode_metrics 

constraint_episode_metrics() -> dict[str, float]
Source code in masa/common/constraints/multi_agent/cmg.py 
228
229
def constraint_episode_metrics(self) -> dict[str, float]:
    return self._constraint.episode_metric()

_labels_by_agent 

_labels_by_agent(infos: Mapping[str, Mapping[str, Any] | None]) -> dict[str, set[str] | frozenset[str]]

Extract and validate infos[agent]['labels'] for all possible agents.

Source code in masa/common/constraints/multi_agent/cmg.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def _labels_by_agent(self, infos: Mapping[str, Mapping[str, Any] | None]) -> dict[str, set[str] | frozenset[str]]:
    """Extract and validate ``infos[agent]['labels']`` for all possible agents."""
    unknown_agents = set(infos) - set(self.possible_agents)
    if unknown_agents:
        raise ValueError(f"Unknown agents in infos: {sorted(unknown_agents)}")

    labels_by_agent: dict[str, set[str] | frozenset[str]] = {}
    for agent in self.possible_agents:
        agent_info = infos.get(agent, {})
        if agent_info is None:
            agent_info = {}
        if not isinstance(agent_info, Mapping):
            raise ValueError(
                f"Expected info for agent '{agent}' to be a mapping, got {type(agent_info).__name__}"
            )
        labels = agent_info.get("labels", set())
        if not isinstance(labels, (set, frozenset)):
            raise ValueError(
                f"Expected labels for agent '{agent}' to be a set of atomic propositions, "
                f"got {type(labels).__name__}"
            )
        labels_by_agent[agent] = labels
    return labels_by_agent