Skip to content

Step-wise Probabilistic Constraint

masa.common.constraints.prob.ProbabilisticSafety 

ProbabilisticSafety(cost_fn: CostFn, alpha: float)

Bases: Constraint

Undiscounted probabilistic constraint based on unsafe-step frequency.

Parameters:

Name Type Description Default
cost_fn CostFn

Mapping from a label set to a scalar cost.

 required
alpha float

Allowed maximum fraction of unsafe steps in an episode.

 required

Attributes:

Name Type Description
total

Number of steps observed so far.
total_unsafe

Number of steps considered unsafe so far.
step_cost

Most recent cost.
Source code in masa/common/constraints/prob.py 
54
55
56
def __init__(self, cost_fn: CostFn, alpha: float):
    self.cost_fn = cost_fn
    self.alpha = alpha

cost_fn instance-attribute 

cost_fn = cost_fn

alpha instance-attribute 

alpha = alpha

constraint_type property 

constraint_type: str

Stable identifier string: "PROB".

reset 

reset()

Reset episode counters.

Source code in masa/common/constraints/prob.py 
58
59
60
61
62
def reset(self):
    """Reset episode counters."""
    self.total = 0
    self.total_unsafe = 0.0
    self.step_cost = 0.0

update 

update(labels: Iterable[str])

Update counters from the current label set.

Parameters:

Name Type Description Default
labels Iterable[str]

Iterable of atomic propositions for the current step.

 required
Source code in masa/common/constraints/prob.py 
64
65
66
67
68
69
70
71
72
def update(self, labels: Iterable[str]):
    """Update counters from the current label set.

    Args:
        labels: Iterable of atomic propositions for the current step.
    """
    self.step_cost = self.cost_fn(labels)
    self.total_unsafe += float(self.step_cost >= 0.5)
    self.total += 1

prob_unsafe 

prob_unsafe() -> float

Return the empirical fraction of unsafe steps.

Returns:

Type Description
float

total_unsafe / total.
Source code in masa/common/constraints/prob.py 
74
75
76
77
78
79
80
81
82
83
def prob_unsafe(self) -> float:
    """Return the empirical fraction of unsafe steps.

    Returns:
        ``total_unsafe / total``.
    """
    if not self.total:
        return 0.0
    else:
        return self.total_unsafe / self.total

satisfied 

satisfied() -> bool

Check whether the unsafe fraction is within the threshold.

Source code in masa/common/constraints/prob.py 
85
86
87
def satisfied(self) -> bool:
    """Check whether the unsafe fraction is within the threshold."""
    return self.prob_unsafe() <= self.alpha

episode_metric 

episode_metric() -> Dict[str, float]

End-of-episode metrics.

Returns:

Type Description
Dict[str, float]

Dict containing:
Dict[str, float]
  • "cum_unsafe": count of unsafe steps,
Dict[str, float]
  • "p_unsafe": proportion of unsafe states in the current trace,
Dict[str, float]
  • "satisfied": 1.0 if p_unsafe <= self.alpha else 0.0.
Source code in masa/common/constraints/prob.py 
89
90
91
92
93
94
95
96
97
98
99
def episode_metric(self) -> Dict[str, float]:
    """End-of-episode metrics.

    Returns:
        Dict containing:

        - ``"cum_unsafe"``: count of unsafe steps,
        - ``"p_unsafe"``: proportion of unsafe states in the current trace,
        - ``"satisfied"``: 1.0 if p_unsafe <= self.alpha else 0.0.
    """
    return {"cum_unsafe": float(self.total_unsafe), "p_unsafe": self.prob_unsafe(), "satisfied": float(self.satisfied())}

step_metric 

step_metric() -> Dict[str, float]

Per-step metrics.

Returns:

Type Description
Dict[str, float]

Dict containing:
Dict[str, float]
  • "cost": current step cost,
Dict[str, float]
  • "violation": 1.0 if cost >= 0.5 else 0.0.
Source code in masa/common/constraints/prob.py 
101
102
103
104
105
106
107
108
109
110
def step_metric(self) -> Dict[str, float]:
    """Per-step metrics.

    Returns:
        Dict containing:

        - ``"cost"``: current step cost,
        - ``"violation"``: 1.0 if ``cost >= 0.5`` else 0.0.
    """
    return {"cost": self.step_cost, "violation": float(self.step_cost >= 0.5)}

masa.common.constraints.prob.ProbabilisticSafetyEnv 

ProbabilisticSafetyEnv(env: Env, cost_fn: CostFn = dummy_cost_fn, alpha: float = 0.01, **kw)

Bases: BaseConstraintEnv

Gymnasium wrapper for ProbabilisticSafety.

Parameters:

Name Type Description Default
env Env

Base environment (must be a LabelledEnv).

 required
cost_fn CostFn

Cost function mapping labels to a scalar.

 cost_fn
alpha float

Allowed maximum unsafe-step fraction.

 0.01
**kw

Extra keyword arguments forwarded to BaseConstraintEnv.

 {}
Source code in masa/common/constraints/prob.py 
128
129
def __init__(self, env: gym.Env, cost_fn: CostFn = dummy_cost_fn, alpha: float = 0.01, **kw):
    super().__init__(env, ProbabilisticSafety(cost_fn=cost_fn, alpha=alpha), **kw)