Skip to content

Constrained Markov Decision Process (CMDP)

masa.common.constraints.cmdp.CumulativeCost 

CumulativeCost(cost_fn: CostFn, budget: float)

Bases: Constraint

CMDP-style cumulative cost constraint with a fixed budget.

The monitor keeps:

  • step_cost: the instantaneous cost \(c_t\),
  • total: the accumulated cost \(C_T\).

Parameters:

Name Type Description Default
cost_fn CostFn

Mapping from a label set to a scalar cost.

 required
budget float

Episode budget \(B\). The episode is satisfied if total <= budget.

 required

Attributes:

Name Type Description
cost_fn

The cost function labels -> float.
budget

Maximum allowed cumulative cost.
total

Running cumulative cost for the current episode.
step_cost

Cost at the most recent update.
Source code in masa/common/constraints/cmdp.py 
62
63
64
def __init__(self, cost_fn: CostFn, budget: float):
    self.cost_fn = cost_fn
    self.budget = budget

cost_fn instance-attribute 

cost_fn = cost_fn

budget instance-attribute 

budget = budget

constraint_type property 

constraint_type: str

Stable identifier string: "CMDP".

reset 

reset()

Reset episode counters.

Source code in masa/common/constraints/cmdp.py 
66
67
68
69
def reset(self):
    """Reset episode counters."""
    self.total = 0.0
    self.step_cost = 0.0

update 

update(labels: Iterable[str])

Update costs from the current label set.

Parameters:

Name Type Description Default
labels Iterable[str]

Iterable of atomic proposition strings for the current step.

 required
Source code in masa/common/constraints/cmdp.py 
71
72
73
74
75
76
77
78
def update(self, labels: Iterable[str]):
    """Update costs from the current label set.

    Args:
        labels: Iterable of atomic proposition strings for the current step.
    """
    self.step_cost = self.cost_fn(labels)
    self.total += self.step_cost

satisfied 

satisfied() -> bool

Check whether the episode remains within budget.

Returns:

Type Description
bool

True iff total <= budget.
Source code in masa/common/constraints/cmdp.py 
80
81
82
83
84
85
86
def satisfied(self) -> bool:
    """Check whether the episode remains within budget.

    Returns:
        ``True`` iff ``total <= budget``.
    """
    return self.total <= self.budget

episode_metric 

episode_metric() -> Dict[str, float]

End-of-episode metrics.

Returns:

Type Description
Dict[str, float]

A dict containing:
Dict[str, float]
  • "cum_cost": cumulative cost over the episode,
Dict[str, float]
  • "satisfied": 1.0 if within budget else 0.0.
Source code in masa/common/constraints/cmdp.py 
88
89
90
91
92
93
94
95
96
97
def episode_metric(self) -> Dict[str, float]:
    """End-of-episode metrics.

    Returns:
        A dict containing:

        - ``"cum_cost"``: cumulative cost over the episode,
        - ``"satisfied"``: ``1.0`` if within budget else ``0.0``.
    """
    return {"cum_cost": self.total, "satisfied": float(self.satisfied())}

step_metric 

step_metric() -> Dict[str, float]

Per-step metrics.

Returns:

Type Description
Dict[str, float]

A dict containing:
Dict[str, float]
  • "cost": instantaneous cost,
Dict[str, float]
  • "violation": 1.0 if the instantaneous cost is considered unsafe under the local convention cost >= 0.5,
Dict[str, float]
  • "cum_cost": running total.
Source code in masa/common/constraints/cmdp.py 
 99
100
101
102
103
104
105
106
107
108
109
110
def step_metric(self) -> Dict[str, float]:
    """Per-step metrics.

    Returns:
        A dict containing:

        - ``"cost"``: instantaneous cost,
        - ``"violation"``: 1.0 if the instantaneous cost is considered unsafe
          under the local convention ``cost >= 0.5``,
        - ``"cum_cost"``: running total.
    """
    return {"cost": self.step_cost, "violation": float(self.step_cost >= 0.5), "cum_cost": self.total}

masa.common.constraints.cmdp.CumulativeCostEnv 

CumulativeCostEnv(env: Env, cost_fn: CostFn = dummy_cost_fn, budget: float = 20.0, **kw)

Bases: BaseConstraintEnv

Gymnasium wrapper that attaches CumulativeCost to an environment.

Parameters:

Name Type Description Default
env Env

Base environment (must be a LabelledEnv).

 required
cost_fn CostFn

Cost function mapping label sets to float cost.

 cost_fn
budget float

Cumulative cost budget \(B\).

 20.0
**kw

Extra keyword arguments forwarded to BaseConstraintEnv.

 {}
Source code in masa/common/constraints/cmdp.py 
128
129
def __init__(self, env: gym.Env, cost_fn: CostFn = dummy_cost_fn, budget: float = 20.0, **kw):
    super().__init__(env, CumulativeCost(cost_fn=cost_fn, budget=budget), **kw)