SR-Forge

Hooks¶

Hooks are the supported way to inject behaviour into the training loop — logging, checkpointing, gradient clipping, progress bars, custom loss terms. They replace the legacy Observer / EventBus system, which is scheduled for removal in v0.16.0. This page covers what hooks are, how to write one, and how to port an existing Observer to a Hook.

For where hooks plug into the runner / trainer machinery, see Trainers & Runners.

The model in one paragraph¶

A Hook is a class with methods. Each method declares (via the @hooks_into("on_<x>") decorator) which HookPoint on the runner / trainer it should fire at. At construction the trainer walks the hook, asks each method "where do you bind?", and registers it with the matching HookPoint. When the runner reaches that stage and calls HookPoint.fire(ctx), every registered method runs with the mutable Context.

That's it. No event bus, no scope tag at construction time, no name-derived event lookup.

A first hook¶

from srforge.training.hooks import Hook, hooks_into
from srforge.training.context import Context


class EpochBanner(Hook):
    """Print a banner at the start of every training epoch."""

    @hooks_into("on_epoch_start")
    def print_banner(self, ctx: Context) -> None:
        print(f"▶ epoch {ctx.epoch}")

Wire it up:

trainer = PyTorchTrainer(
    ...,
    hooks=[EpochBanner()],
)

Or in YAML:

trainer:
  _target: srforge.training.trainers.PyTorchTrainer
  params:
    hooks:
      - _target: my_module.EpochBanner

The three binding sources¶

Hook.bind_to resolves each method's target HookPoint from three sources, in this order:

self.bindings — instance- or class-level dict {method_name: target_point_name | [target_point_names] | None}. Set via constructor kwarg, class attribute, or YAML. None / [] explicitly disables a default binding.
@hooks_into("on_<x>", ...) — decorator metadata on the method itself. Stack the decorator or pass multiple names to bind one method to several points.
Nothing — a method without a decorator and without a bindings entry is not a handler. Naming a method on_<x> is not enough; sr-forge does not introspect names.

Multiple targets¶

Bind one method to several HookPoints by passing the names together:

class EpochBanner(Hook):
    @hooks_into("on_epoch_start", "on_iteration_start")
    def print_banner(self, ctx):
        ...

Or stack decorators — equivalent:

    @hooks_into("on_epoch_start")
    @hooks_into("on_iteration_start")
    def print_banner(self, ctx):
        ...

Remapping at the call site¶

If a hook ships bound to on_trainer_epoch_finished but your custom runner exposes a on_iteration_complete point instead, override at construction:

from srforge.training.hooks import LossLogger

logger = LossLogger(
    tracker=...,
    bindings={"flush_epoch_metrics": "on_iteration_complete"},
)

In YAML, the same thing:

hooks:
  - _target: srforge.training.hooks.LossLogger
    params:
      tracker: ${ref:tracker}
      bindings:
        flush_epoch_metrics: on_iteration_complete

Disabling a default binding¶

Pass None to suppress a method's default binding without subclassing:

hooks:
  - _target: my_module.CheckpointSaver
    params:
      path: ${ref:out}/best.pt
      bindings:
        restore_best_loss: null     # skip the restore step

Attaching hooks: directly, not via cascade¶

Pass each hook to the component it should fire from. There is no distribution layer — a hook attached to a runner fires from that runner's HookPoints; a hook attached to the trainer fires from the trainer's HookPoints; nothing more.

train_runner = TrainingEpochRunner(
    optimizer=opt,
    hooks=[ProgressBar(name="T"), GradientClip(max_norm=1.0)],
)
val_runner = ValidationEpochRunner(
    hooks=[ProgressBar(name="V")],
)
trainer = PyTorchTrainer(
    training_epoch_runner=train_runner,
    validation_epoch_runner=val_runner,
    hooks=[LossLogger(tracker=tracker)],   # fires only from the Trainer
    ...,
)

If you need the same hook instance to fire from several components, put the same object in each hooks=[...] list — hook instances are plain Python objects, nothing prevents reuse:

progress = ProgressBar(name="P")
train_runner = TrainingEpochRunner(..., hooks=[progress])
val_runner   = ValidationEpochRunner(hooks=[progress])

In YAML, declare the hook once and reference it (or use a YAML anchor):

hooks:
  progress: &progress
    _target: srforge.training.hooks.ProgressBar
    params: {name: "P"}

training_runner:
  _target: srforge.training.runners.TrainingEpochRunner
  params:
    hooks: [*progress]

validation_runner:
  _target: srforge.training.runners.ValidationEpochRunner
  params:
    hooks: [*progress]

Persistent state¶

If a hook carries state that should survive a checkpoint reload, override state_dict() and load_state_dict():

class StepCounter(Hook):
    def __init__(self):
        self.steps = 0

    @hooks_into("on_post_step")
    def tick(self, ctx):
        self.steps += 1

    def state_dict(self):
        return {"steps": self.steps}

    def load_state_dict(self, state):
        self.steps = state["steps"]

The trainer collects each hook's state_dict into the overall training state.

Porting an `Observer` to a `Hook`¶

The migration is mechanical. Most hooks land in 5–10 minutes.

Side-by-side example¶

Before — Observer:

import srforge.events.trainer
from srforge.observers.base import Observer


class LossLogger(Observer):
    EVENTS = [srforge.events.trainer.TrainerEpochFinished]

    def __init__(self, tracker=None, **kwargs):
        super().__init__(**kwargs)              # forwards `scope` kwarg
        self.tracker = tracker

    def on_epoch_finished(
        self,
        event: srforge.events.trainer.TrainerEpochFinished,
    ) -> None:
        self.tracker.log_metrics(
            event.train_loss.mean_weighted(), step=event.epoch
        )

After — Hook:

from srforge.training.hooks import Hook, hooks_into
from srforge.training.context import Context


class LossLogger(Hook):
    def __init__(self, tracker=None):
        self.tracker = tracker

    @hooks_into("on_trainer_epoch_finished")
    def flush_epoch_metrics(self, ctx: Context) -> None:
        self.tracker.log_metrics(
            ctx.train_loss.mean_weighted(), step=ctx.epoch
        )

Step-by-step¶

Replace the base class. class Foo(Observer) → class Foo(Hook). Drop the EVENTS = [...] line.
Drop the super().__init__(**kwargs) boilerplate unless you have your own kwargs to forward. Hook.__init__ only accepts bindings= and scope=.
Rename methods to describe what they do, not which event they handle (on_epoch_finished → flush_epoch_metrics, save_checkpoint, update_ema, …). The method name is documentation now; the binding is metadata.
Add @hooks_into("on_<x>") above each handler. Map old on_<event_name> methods to the corresponding HookPoint name:
on_runner_epoch_started → @hooks_into("on_epoch_start")
on_runner_batch_finished → @hooks_into("on_post_step")
on_runner_epoch_finished → @hooks_into("on_epoch_end")
on_training_began → @hooks_into("on_training_began") (same)
on_trainer_epoch_finished → @hooks_into("on_trainer_epoch_finished") (same)
Swap event: SomeEvent for ctx: Context in the signature. The fields you read off the event almost all live on ctx — event.epoch becomes ctx.epoch, event.train_loss becomes ctx.train_loss, and so on. If a field is missing, check srforge/training/context.py.
Drop the EventBus plumbing. No more bus.subscribe(observer); pass the hook to Trainer(hooks=[...]) or EpochRunner(hooks=[...]) directly — to the specific component it should fire from.
If the old observer used scope="train" etc., simply pass the hook to that runner's hooks=[...] and don't pass it to the others. There is no scope filter on Hook — direct attachment is the contract.

Field mapping cheat-sheet¶

Event field	Context attribute
`event.epoch`	`ctx.epoch`
`event.batch`	`ctx.batch_idx`
`event.entry`	`ctx.entry`
`event.batch_scores`	`ctx.scores`
`event.epoch_scores`	`ctx.epoch_scores`
`event.train_loss`	`ctx.train_loss`
`event.val_loss`	`ctx.val_loss`
`event.lr_scheduler`	`ctx.lr_scheduler`
`event.runner_state`	`ctx.runner_state`
`event.total_epochs`	`ctx.total_epochs`
`event.initial_epoch`	`ctx.initial_epoch`
`event.best_losses`	`ctx.best_losses`

Inspecting a hook's bindings¶

Call Hook.list_bindings() to see the resolved (method, target_point) pairs — useful when porting and when something doesn't fire:

>>> LossLogger(tracker=...).list_bindings()
[('flush_epoch_metrics', 'on_trainer_epoch_finished')]

Where to go next¶

Trainers & Runners — the full lifecycle, including which HookPoints each runner exposes.
Observers & Events — the legacy system, kept for reference until v0.16.0.
API reference: srforge.training.hooks (built-in hooks), srforge.training.hookpoint (the HookPoint / Handle primitives).