latents.callbacks

Callbacks for model fitting.

Callbacks allow custom actions at specific points during model fitting. Implement any subset of the callback methods you need (duck typing).

Callback Methods:

on_fit_start(ctx)

Called when fit() begins, after initialization.

on_fit_end(ctx, reason)

Called when fit() completes. reason is one of: “converged”, “max_iter”, “no_latents”.

on_iteration_end(ctx, iteration, lb, lb_prev)

Called at the end of each EM iteration.

on_flag_changed(ctx, flag, value, iteration)

Called when a boolean flag changes (converged, decreasing_lb, private_var_floor).

on_x_dim_pruned(ctx, n_removed, x_dim_remaining, iteration)

Called when latent dimensions are pruned.

Iteration Numbering:

Callbacks receive 0-indexed iteration values from the fit loop (for array indexing). User-facing output (logs, filenames) uses 1-indexed iterations for clarity.

Examples:

>>> from latents.callbacks import ProgressCallback, CheckpointCallback
>>>
>>> model.fit(Y, callbacks=[
...     ProgressCallback(),
...     CheckpointCallback(save_dir="./checkpoints"),
... ])

Callbacks

ProgressCallback	Display tqdm progress bar during fitting.
LoggingCallback	Emit log events to the 'latents' logger during fitting.
CheckpointCallback	Save model checkpoints during fitting.

Utilities

invoke_callbacks

Call a method on all callbacks that implement it.

---

class ProgressCallback(desc: str = 'Fitting')

Display tqdm progress bar during fitting.

Parameters:

descstr, default “Fitting”

Description shown next to the progress bar.

Examples

>>> model.fit(Y, callbacks=[ProgressCallback()])

Custom description

>>> model.fit(Y, callbacks=[ProgressCallback(desc="Training GFA")])

class LoggingCallback

Emit log events to the ‘latents’ logger during fitting.

Configure Python logging to see output:

import logging
logging.basicConfig(level=logging.INFO)

Or to log to a file:

import logging
logging.basicConfig(level=logging.INFO, filename="fit.log")

Examples

>>> import logging
>>> logging.basicConfig(level=logging.INFO)
>>> model.fit(Y, callbacks=[LoggingCallback()])

class CheckpointCallback(

save_dir: str | Path,

every_n_iter: int = 5000,

save_initial: bool = True,

save_final: bool = True,

save_on_interrupt: bool = True,

max_checkpoints: int = 3,

prefix: str = '',

)

Save model checkpoints during fitting.

Checkpoints are saved in safetensors format and can be loaded via GFAModel.load().

Parameters:

save_dirstr or Path

Directory for checkpoint files. Created if it doesn’t exist.

every_n_iterint, default 5000

Save every N iterations. Set to 0 to disable periodic saves.

save_initialbool, default True

If True, save immediately after initialization (before iteration 0).

save_finalbool, default True

If True, save after fit completes.

save_on_interruptbool, default True

If True, save checkpoint when Ctrl+C is pressed. Only works in the main process; in parallel workers, rely on periodic checkpoints.

max_checkpointsint, default 3

Maximum periodic checkpoints to keep. Older ones are deleted. Set to 0 to keep all. Does not affect initial, final, or interrupt checkpoints.

prefixstr, default “”

Optional prefix for checkpoint filenames.

Examples

>>> callback = CheckpointCallback(
...     save_dir="./checkpoints",
...     every_n_iter=5000,
...     prefix="experiment1",
... )
>>> model.fit(Y, callbacks=[callback])

invoke_callbacks(callbacks: list, method: str, **kwargs: Any) → None

Call a method on all callbacks that implement it.

Parameters:

callbackslist

List of callback objects.

methodstr

Method name to call.

**kwargsAny

Arguments passed to the method.