decent_bench.agents

class decent_bench.agents.Agent(*_: object, _id: UUID | None = None, **__: object)

Bases: object

Agent with local cost function, activation scheme, state snapshot period, and optional data.

At initialization, the agent is assigned a unique id (accessible via Agent.id) which serves as its hash. The agent can also be assigned an index (Agent.index). This assignment is performed when initializing a network, and is useful to index arrays by Agent or for user-friendly representation of Agents.

Parameters:

• cost – local cost function; once assigned, it should not be modified

• activation – activation scheme to model synchrony/asynchrony; defaults to synchrony (activate at all iterations)

• state_snapshot_period – how often to record the agent’s state when executing an algorithm

• data – dictionary for arbitrary agent data

Raises:

ValueError – if state_snapshot_period is not a positive int

property id: UUID

Unique id for the agent.

property index: int

Agent index within a network, -1 if unassigned.

property cost: Cost

Local cost function.

Alias: f, loss

property f: Cost

Local cost function.

Alias: f, loss

property loss: Cost

Local cost function.

Alias: f, loss

property x: Array

Local optimization variable x.

Raises:

RuntimeError – if x is retrieved before being set or initialized

property state_snapshot_period: int

Number of iterations between snapshots of the agent’s state.

messages(channel: str = 'default') → Mapping[Agent, Array]

Received messages with channel, keyed by sender.

message(sender: Agent, channel: str = 'default') → Array

Received message from sender with channel.

property aux_vars: dict[str, Any]

Auxiliary optimization variables used by algorithms that require more variables than x.

initialize(*, x: Array | None = None, aux_vars: dict[str, Any] | None = None) → None

Initialize local variables and messages before running an algorithm.

Parameters:

• x – initial x

• aux_vars – initial auxiliary variables

Raises:

ValueError – if initialized x has incorrect shape

static no_count(agents: Sequence[Agent]) → Iterator[None]

Context manager that disables call counting for a sequence of agents.

Use this when computing metrics or other operations that should not be counted as algorithm function/gradient calls.

Parameters:

agents – sequence of agents to disable call counting for

Example:

with Agent.no_count(agents):
    value = metric.compute(problem, agents, iteration)

class decent_bench.agents.ReceivedMessages

Bases: object

Container for received messages keyed by channel and sender.

put(sender: Agent, msg: Array, channel: str = 'default') → None

Store or overwrite a message from sender under channel.

get(sender: Agent, channel: str = 'default') → Array

Return the message from sender under channel.

has(sender: Agent, channel: str = 'default') → bool

Return True if a message from sender exists under channel.

by_channel(channel: str = 'default') → Mapping[Agent, Array]

Return a read-only sender->message mapping for channel.

clear(sender: Agent | Sequence[Agent] | None = None, channel: str | None = None) → None

Clear messages with optional sender/channel scoping.

class decent_bench.agents.AgentHistory

Bases: object

Ordered history of an agent’s optimization variable x, indexed by algorithm iteration.

Snapshots are stored sparsely — only iterations at which Agent._snapshot was called are recorded. Lookups for iterations between snapshots fall back to the nearest preceding snapshot.

Internally, snapshots are kept in a dict for O(1) exact lookup and a parallel sorted list for O(log n) predecessor search via bisect.

max() → int

Return the latest iteration for which a snapshot exists.

Raises:

ValueError – if no snapshots have been recorded yet.

min() → int

Return the earliest iteration for which a snapshot exists.

Raises:

ValueError – if no snapshots have been recorded yet.

items() → Iterator[tuple[int, Array]]

Yield (iteration, x) pairs for every snapshot, in ascending iteration order.

values() → Iterator[Array]

Yield the x snapshot for every recorded iteration, in ascending iteration order.

keys() → list[int]

Return a sorted list of all iterations for which a snapshot has been recorded.

set_x(iteration: int, x: Array) → None

Record x at iteration, replacing any existing snapshot at that iteration.

Also available as history[iteration] = x.

Raises:

ValueError – if iteration is negative.

get_x(iteration: int) → Array

Return x at iteration, falling back to the nearest preceding snapshot if needed.

Snapshots are not necessarily recorded at every iteration (controlled by Agent.state_snapshot_period). When the exact iteration is not found, the closest snapshot with an iteration number <= iteration is returned instead. For example, if snapshots exist at iterations 0, 10, 20 and iteration 23 is requested, the snapshot from iteration 20 is returned.

Also available as value = history[iteration].

Parameters:

iteration – The algorithm iteration to retrieve x for.

Raises:

ValueError – if iteration is before the first recorded snapshot or if iteration < -1.