Source code for decent_bench.metrics.runtime_library
"""Collection of pre-defined runtime metrics."""
from collections.abc import Sequence
from typing import TYPE_CHECKING
import decent_bench.utils.interoperability as iop
from decent_bench.metrics._runtime_metric import RuntimeMetric
if TYPE_CHECKING:
from decent_bench.agents import Agent
from decent_bench.benchmark import BenchmarkProblem
[docs]
class RuntimeLoss(RuntimeMetric):
r"""
Runtime loss metric.
Computes the average loss across all agents at each iteration.
This is useful for monitoring convergence and detecting issues early.
The loss is computed as:
.. math::
\text{loss} = \frac{1}{N} \sum_{i=1}^{N} f_i(\mathbf{x}_i)
where :math:`N` is the number of agents, :math:`f_i` is agent i's cost function,
and :math:`\mathbf{x}_i` is agent i's current optimization variable.
"""
description = "Loss"
x_log = False
y_log = False
[docs]
def compute(self, _: "BenchmarkProblem", agents: Sequence["Agent"], __: int) -> float: # noqa: D102
total_loss = sum(agent.cost.function(agent.x) for agent in agents)
return total_loss / len(agents)
[docs]
class RuntimeRegret(RuntimeMetric):
r"""
Runtime regret metric.
Requires a :class:`~decent_bench.benchmark.BenchmarkProblem` with `x_optimal` not None.
Regret is computed as:
.. math::
\text{regret} = \frac{1}{N} \sum_{i=1}^{N} f_i(\mathbf{x}_i) - \frac{1}{N} \sum_{i=1}^{N} f_i(\mathbf{x}^*)
where :math:`N` is the number of agents, :math:`f_i` is agent i's cost function, :math:`\mathbf{x}_i` is agent i's
current optimization variable, and :math:`\mathbf{x}^*` is the optimal solution.
"""
description = "Regret"
x_log = False
y_log = False
[docs]
def compute(self, problem: "BenchmarkProblem", agents: Sequence["Agent"], _: int) -> float: # noqa: D102
if getattr(problem, "x_optimal", None) is None:
return float("nan")
agent_cost = sum(agent.cost.function(agent.x) for agent in agents) / len(agents)
if hasattr(self, "_cached_optimal_cost"):
return agent_cost - self._cached_optimal_cost
self._cached_optimal_cost: float = sum(agent.cost.function(problem.x_optimal) for agent in agents) / len(agents) # type: ignore[arg-type,misc]
return agent_cost - self._cached_optimal_cost
[docs]
class RuntimeGradientNorm(RuntimeMetric):
r"""
Runtime gradient norm metric.
Computes the average gradient norm across all agents at each iteration.
This is useful for monitoring if the algorithm is making progress towards
a stationary point.
The gradient norm is computed as:
.. math::
\text{grad norm} = \frac{1}{N} \sum_{i=1}^{N} \|\nabla f_i(\mathbf{x}_i)\|
where :math:`N` is the number of agents, :math:`f_i` is agent i's cost function,
and :math:`\mathbf{x}_i` is agent i's current optimization variable.
"""
description = "Gradient Norm"
x_log = False
y_log = True
[docs]
def compute(self, _: "BenchmarkProblem", agents: Sequence["Agent"], __: int) -> float: # noqa: D102
grad_norms = sum(float(iop.norm(agent.cost.gradient(agent.x))) for agent in agents)
return grad_norms / len(agents)
[docs]
class RuntimeConsensusError(RuntimeMetric):
r"""
Monitors how well agents agree on their decision variables.
This is useful for diagnosing issues in decentralized algorithms where agents are supposed to reach consensus.
The consensus error is computed as:
.. math::
\text{consensus error} = \frac{1}{N} \sum_{i=1}^{N} \|\mathbf{x}_i - \bar{\mathbf{x}}\|
where :math:`N` is the number of agents, :math:`\mathbf{x}_i` is agent i's current optimization variable,
and :math:`\bar{\mathbf{x}}` is the average of all agents' optimization variables.
.. seealso::
:class:`~decent_bench.metrics.metric_library.ConsensusError`.
"""
description = "Consensus Error"
x_log = False
y_log = True
[docs]
def compute(self, _: "BenchmarkProblem", agents: Sequence["Agent"], __: int) -> float: # noqa: D102
# Compute average x across all agents
x_avg = iop.mean(iop.stack([agent.x for agent in agents]), dim=0)
# Compute average distance from the mean
errors = [float(iop.norm(agent.x - x_avg)) for agent in agents]
return sum(errors) / len(agents)
