Model

Core simulation model: population store, write flush, and run loop.

Write architecture (single source of truth = Polars agents_df)

• OOP path — Agent.__setattr__ queues into _pending_writes; _flush_pending_writes() applies them (uses ambr._id_index).

• Vectorized path — view API in ambr.sequences calls _set_frame() directly (scatter / set / column assign).

• Tensor path — ambr.tensor_lane borrow/commit also uses _set_frame().

Id-position caches (_id_pos_cache, _ids_arange_cache) are owned here but filled by ambr._id_index; _bump_id_version() invalidates them.

class ambr.model.Model(parameters: Dict[str, Any])

Bases: BaseModel

Base class for all simulation models (Polars-backed agent table).

add_agent(agent: Agent)

Add a single agent. Prefer add_agents() for bulk creation.

add_agents(n: int, *, agent_class: Type | None = None, **columns: Any)

Bulk-create n agents with columnar initial state:

self.add_agents(100, wealth=self.rng.integers(1, 10, 100),
                     status='S')

Scalar kwargs broadcast; list / np.ndarray / pl.Series values must have length n. Pass agent_class= to also spin up Python instances so AgentList.call and per-agent iteration work.

agent_reporters: List[str] | None = None

Declarative per-agent columns to snapshot each step into the (opt-in) 'agent_vars' long-format results frame. Empty / None = no cost. Override on the subclass; do not mutate the base attribute.

property agents_df: DataFrame

approve_fast_path(evidence: str) → Model

Allow this instance to use a private optimized GPU loop.

evidence must be a non-empty provenance label chosen by the caller (for example, a test report or experiment identifier). AMBER records only the explicit declaration; it does not validate the evidence. Without approval, run() uses the general native runner even when private _setup_gpu_fast / _run_gpu_fast hooks are present.

batch_update_agents(agent_ids: list, data: dict)

Deprecated: use agents.at[ids].set(**data) (or column assign).

Still equivalent to self.agents.at[agent_ids].set(**data) so multi-column updates stay atomic and contract-observed.

property contract_certificates: List[ContractCertificate]

cpu(mode: str | None = None) → Model

Place the next run() on CPU. Returns self for chaining.

Parameters:

mode – optional execution style — 'vectorized' (view API, default when omitted on the model) or 'oop' (per-agent objects). Same as model.cpu().run(mode=...) when passed here.

Examples:

model.cpu().run()
model.cpu(mode="vectorized").run()
model.cpu(mode="oop").run()

property device: str

Selected execution device for the next run() — 'cpu' or 'gpu'.

end()

property fast_path_approval: str | None

Caller-supplied evidence label, or None when not approved.

get_agent_data(agent_id: Any) → DataFrame

Return a 1-row DataFrame with the current state of agent_id.

Uses agents_df so pending buffered writes are flushed first.

gpu(mode: str | None = None) → Model

Place the next run() on GPU (device-resident columns). Returns self.

Parameters:

mode – optional execution style. GPU runs support the 'vectorized' view API; Python Agent objects use CPU OOP mode.

Examples:

model.gpu().run()
model.gpu(mode="vectorized").run()

property mode: str

Execution style — 'vectorized' (view API) or 'oop' (agent objects).

model_reporters: Dict[str, Any] | None = None

Declarative model-level metrics, evaluated once per step into the 'model' results frame. Maps a column name to a callable(model), the name of a model attribute/method, or a constant. Override on the subclass (do not mutate this class attribute – default is None to avoid shared mutable state). Complements imperative record_model.

params: Dict[str, Any] | None = None

Optional typed parameter schema {'name': (type, default)}. When set on the subclass, self.p.name is pre-coerced to type at init (missing -> default). Default None avoids shared mutable state.

record(key: str, value: Any)

Deprecated AgentPy alias for record_model() / model_reporters.

record_initial: bool = False

When True, capture a t=0 row of the reporters before the first step.

record_model(key: str, value: Any)

Record a model-level variable for the current step.

revoke_fast_path_approval() → Model

Remove this instance’s private-fast-path deployment approval.

property rng

Step RNG — device RNG during gpu() runs, NumPy otherwise.

run(steps: int | None = None, contract: str = 'off', mode: str | None = None, device: str | None = None, backend: str | None = None) → RunResults

Run the simulation.

Returns a RunResults mapping (dict subclass). Use results['agents'] or results.agents interchangeably.

Device placement is Keras-style: call cpu() or gpu() on the model (or pass device= / legacy backend= here). Mode can be set on those fluent methods (model.cpu(mode="vectorized")) or here via mode=. On CPU, mode='vectorized' (default) dispatches step_vectorized(); mode='oop' dispatches step_oop() and expects tracked per-agent objects. Models that only implement step() retain backwards-compatible fallback behavior.

Parameters:

• steps – number of steps to run (defaults to self.p['steps'] or 100).

• contract – snapshot-view conformance checking mode. One of 'off' (default, zero overhead), 'check' (record a per-step ContractCertificate in self.contract_certificates and the 'contract' results key), 'warn' (also emit a warning per violation), or 'raise' (raise ContractViolationError on the first step with an error-severity violation).

• mode – 'vectorized' (default) or 'oop'. Overrides cpu() / gpu() mode= and parameters['mode'].

• device – 'cpu' or 'gpu'. Overrides cpu() / gpu(). Legacy alias: backend=.

run_step() → None

Execute one simulation step. The first call also runs setup.

When a contract mode other than 'off' is active (set via run()), the step body is bracketed by the snapshot-view conformance checker and a ContractCertificate is appended to self.contract_certificates.

setup()

step()

step_oop()

Execute one object-oriented step.

Models with tracked Agent objects can override this hook. The default preserves backwards compatibility for models that have only a single step() implementation.

step_vectorized()

Execute one vectorized step.

Models with a distinct vectorized implementation can override this hook. The default preserves the pre-mode API, where step() is the model’s single implementation.

update()

Per-step hook, called after step() with t already advanced.

Override to record model metrics imperatively (self.record_model(...)) or for any post-step bookkeeping. This is a pure hook: overriding it no longer requires super().update() – the step counter and step-data lifecycle are owned by run_step(). Calling super().update() remains legal (it is a no-op) so existing models keep working.

update_agent_data(agent_id: int, data: Dict[str, Any])

Deprecated: use agent.<col> = value or agents.at[id].set(...).

Still routes through _queue_write() so the snapshot-view contract can witness the writes.

property xp

NumPy or CuPy array module for the active run (CPU when idle).

The Model class is the foundation of any AMBER simulation. It provides the framework for:

• Managing simulation time and execution

• Storing and updating agent data

• Recording model-level metrics

• Coordinating agent behaviors

Basic Usage

import ambr as am

class MyModel(am.Model):
    def setup(self):
        # Initialize agents and environment
        pass

    def step_vectorized(self):
        # Columnar / array-native implementation
        pass

    def step_oop(self):
        # Optional tracked-Agent implementation
        pass

# Run the model (fluent placement, 0.4.4)
model = MyModel({'steps': 100, 'seed': 42})
results = model.cpu(mode="vectorized").run()
# results = model.gpu().run()   # vectorized lane on GPU

Key Methods

Lifecycle Methods:

• setup() - Called once at the beginning to initialize the model

• step_vectorized() - Called for vectorized CPU/GPU runs

• step_oop() - Called for CPU OOP runs with tracked Agent objects

• step() - Backwards-compatible fallback when a lane hook is not defined

• update() - Called after step() to update model state

• end() - Called once at the end of the simulation

Data Management:

• add_agent(agent) - Add a new agent to the model

• update_agent_data(agent_id, data) - Update data for a specific agent

• get_agent_data(agent_id) - Retrieve data for a specific agent

• record_model(name, value) - Record a model-level metric

Execution / placement (0.4.4):

• cpu(mode=None) - Place the next run on CPU (optional mode='vectorized'|'oop'); returns self for chaining

• gpu(mode=None) - Place the next run on GPU with device-resident columns; GPU runs are vectorized-only

• approve_fast_path(evidence) - Explicitly allow a private optimized GPU loop on this model instance and retain the caller-supplied evidence label; AMBER does not verify that label (requires contract="off")

• revoke_fast_path_approval() - Return the instance to the general runner

• run(...) - Execute the full simulation and return results. Accepts device=, mode=, contract=; legacy backend= still works but is deprecated

• run_step() - Execute a single time step

Mode defaults to vectorized. Fluent placement and run(mode=...) / run(device=...) compose: kwargs to run override what cpu() / gpu() set when both are used.