Events and Subscriptions

Radiate provides an event system that allows you to monitor and react to the evolution process in real-time. This is great for:

• Tracking the progress of evolution
• Collecting metrics and statistics
• Implementing custom logging or logic based on the state of evolution
• Visualizing the evolution process

Overview

The event system in radiate is built around the concept of event handlers or subscribers that can be attached to the GeneticEngine. These subscribers receive events at key points during the evolution process, allowing you to monitor and react to changes in the environment in real-time. The event system is designed to be flexible and extensible, allowing you to create custom event handlers that can perform various actions based on the evolution state.

The GeneticEngine offloads nearly all of a subscriber's compute cost onto the handler itself — so be mindful of this when implementing your handlers; expensive work here can slow the whole run.

Threading Behavior

Currently, the rust implementation is multi-threaded (if multi-threaded executors are used), meaning if you have multiple subscribers, there is no guarantee of the order in which they will be called. For python, regardless of if you are using a free-threaded interpreter (3.13t/3.14t, etc) or not, the events will be dispatched on a single thread in the order they were added.

Below there is a brief description of each event type with its representative data structures expressed in json.

Start Event

This event is triggered when the evolution process starts. It provides an opportunity to initialize any resources or perform setup tasks before the evolution begins.

{
    'event_type': 'start_event'
}

Stop Event

This event is triggered when the evolution process stops, either due to reaching a stopping condition or being manually stopped. It provides access to:

• The final metrics of the evolution
• The best individual found
• The final score, or fitness, of the best individual

{
    'event_type': 'stop_event',
    'index': 0, // Current generation number
    // This will be a MetricSet of metrics collected, see Engine's metrics docs for more info
    'metrics': ..., 
    // This will be the decoded best individual found so far. So, if you are 
    // evolving a vector of FloatGenes, this will be a list of floats
    'best': [3.9699993,  1.5489225, -1.7164116,  1.0756674, -1.932127 , -2.3247557], 
    'score': 0.3327971398830414
}

Epoch Start Event

This event is triggered at the start of each generation (epoch) and provides the current generation number. It allows you to perform actions before the evolution step begins, such as resetting counters or logging initial state.

{
    'event_type': 'epoch_start_event',
    'index': 0  // Current generation number
}

Epoch Complete Event

This event is triggered at the end of each generation (epoch) and provides information about:

• The current generation number
• The current metrics from the GeneticEngine
• The best individual found from the GeneticEngine so far
• The best score, or fitness, from the best individual

{
    'event_type': 'epoch_complete_event',
    'index': 0, // Current generation number
    // This will be the current metrics collected, see Engine's metrics docs for more info
    'metrics': ..., 
    // This will be the decoded best individual found so far. So, if you are 
    // evolving a vector of FloatGenes, this will be a list of floats
    'best': [3.9699993,  1.5489225, -1.7164116,  1.0756674, -1.932127 , -2.3247557], 
    'score': 0.3327971398830414,
    'objective': ['min']  // The optimization objective(s) used in this run
}

Engine Improvement Event

This event is triggered when the engine finds a new best individual during the evolution process. It provides:

• The index of the generation where the improvement occurred
• The best individual found at that point
• The score, or fitness, of the best individual

{
    'event_type': 'engine_improvement_event',
    'index': 0, // Current generation number
    // This will be the decoded best individual found so far. So, if you are 
    // evolving a vector of FloatGenes, this will be a list of floats
    'best': [3.9699993,  1.5489225, -1.7164116,  1.0756674, -1.932127 , -2.3247557], 
    'score': 0.3327971398830414
}

Checkpoint Saved Event

This event is triggered when a checkpoint is saved during the evolution process. It provides:

• The index of the generation at which the checkpoint was saved

{
    "event_type": "checkpoint_saved",
    "index": 42,
    "path": "/path/to/checkpoint"
}

Limit Triggered Event

This event is triggered when a limit set on the engine is reached during the evolution process. It provides:

• The index of the generation at which the limit was triggered

{
    "event_type": "limit_triggered",
    "index": 100,
    "limit": "..." // this will be the actual limit that was triggered
}

---

Subscribing to Events

You can subscribe to events in two ways:

Callback Function

The simplest way to subscribe to events is by providing a callback function:

Python Rust

import radiate as rd

engine = (
    rd.Engine.int(10, init_range=(0, 100))
    .fitness(your_fitness_func)
    .subscribe(
        lambda event: print(event)
    )  # Subscribe to all events using a lambda function
    # ... other parameters ...
)

let engine = GeneticEngine::builder()
    .codec(FloatCodec::vector(6, -5.0..5.0))
    .fitness_fn(your_fitness_fn)
    // ... other parameters ...
    .build();

engine.subscribe::<EpochComplete<Vec<f32>>>(|event: &EpochComplete<Vec<f32>>| {
    println!(
        "Printing from event handler! [ {:?} ]: {:?}",
        event.index, event.score
    );
});

// Run the engine
let result = engine.run(|generation| generation.index() >= 100);

Event Handler Class

For more complex event handling, you can create a custom event handler class:

Python Rust

import radiate as rd


# Inherit from EventHandler, tell the super class which event you'd like to subscribe to,
# then override the on_event method
class MySubscriber(rd.EventHandler):
    """
    If no `rd.EventType` is passed to the super constructor, this handler will subscribe to all events.
    Otherwise, it will only subscribe to the specified event type.
    """

    def __init__(self):
        super().__init__(rd.EventType.EPOCH_COMPLETE)

    def on_event(self, event: rd.EngineEvent) -> None:
        print(f"Event: {event}")


# Create an instance of your event handler
handler = MySubscriber()

engine = (
    rd.Engine.int(10, init_range=(0, 100))
    .fitness(your_fitness_func)
    .subscribe(handler)  # Add your handler here
    # ... other parameters ...
)

It's also completely possible to create more advanced forms of visualization or logging through this method. For example, below we will collect the scores from each epoch then use polars to create a DataFrame and finally plot it with plotly.

class ScorePlotterHandler(rd.EventHandler):
    """
    An event handler that collects best scores over epochs and plots them at the end.
    1. On EPOCH_COMPLETE, it appends the best score to a list.
    2. On STOP, it creates a DataFrame and plots the scores over generations.
    """

    def __init__(self):
        super().__init__()  # Not specifying an event type to listen to all events
        self.scores = []

    def on_event(self, event: rd.EngineEvent) -> None:
        if event.event_type == rd.EventType.EPOCH_COMPLETE:
            best_score = event.score()
            self.scores.append(best_score)
        elif event.event_type == rd.EventType.STOP:
            df = pl.DataFrame(
                {"Generation": list(range(len(self.scores))), "Score": self.scores}
            )
            fig = go.Figure(go.Scatter(x=df["Generation"], y=df["Score"], mode="lines"))
            fig.update_layout(
                xaxis_title="Generation",
                yaxis_title="Best Score",
                title="Best Score over Generations",
            )
            fig.show()


# Create an instance of your event handler
handler = ScorePlotterHandler()

engine = (
    rd.Engine.int(10, init_range=(0, 100))
    .fitness(your_fitness_func)
    .subscribe(handler)  # Add your handler here
    # ... other parameters ...
)

struct MyHandler;

impl Handler<EpochComplete<Vec<f32>>> for MyHandler {
    fn handle(&mut self, event: &EpochComplete<Vec<f32>>, _ctx: &EventContext<'_, Self>) {
        println!(
            "Printing from event handler! [ {:?} ]: {:?}",
            event.index, event.score
        );
    }
}

// Create and configure the engine
let engine = GeneticEngine::builder()
    .codec(FloatCodec::vector(6, -5.0..5.0))
    .fitness_fn(your_fitness_fn)
    .subscribe(MyHandler)
    // ... other parameters ...
    .build();

// Run the engine
let result = engine.run(|generation| generation.index() >= 100);

Decorator Shortcuts

For single-purpose handlers, four decorators skip the subclass-and-override boilerplate by pinning a plain function to one EventType. Each one just wraps your function in a CallableEventHandler, so the result is still a normal handler you pass to .subscribe().

Decorator	Fires on
on_start	EventType.START
on_epoch	EventType.EPOCH_COMPLETE
on_improvement	EventType.ENGINE_IMPROVEMENT
on_stop	EventType.STOP
on_limit_triggered	EventType.LIMIT_TRIGGERED
on_checkpoint_saved	EventType.CHECKPOINT_SAVED
on_event	All event types

No shortcut for EPOCH_START

There isn't an on_epoch_start decorator — of the five event types, only these four have a decorator. Use a lambda or an EventHandler subclass if you need to react to EPOCH_START specifically. Also note on_epoch maps to EPOCH_COMPLETE, not EPOCH_START.

Python Rust

import radiate as rd


# Each decorator pins the handler to a single EventType, so you skip the
# subclass-and-override boilerplate for single-purpose handlers.
@rd.on_start  # EventType.START
def log_start(event: rd.EngineEvent) -> None:
    print("Evolution has started!")


@rd.on_epoch  # EventType.EPOCH_COMPLETE
def log_epoch(event: rd.EngineEvent) -> None:
    print(f"Epoch {event.index}: best score = {event.score()}")


@rd.on_stop  # EventType.STOP
def log_stop(event: rd.EngineEvent) -> None:
    print(event.metrics().dashboard())


@rd.on_improvement  # EventType.IMPROVEMENT
def log_improvement(event: rd.EngineEvent) -> None:
    print(f"New best found at epoch {event.index}: {event.score()}")


@rd.on_limit_triggered  # EventType.LIMIT_TRIGGERED
def log_limit_triggered(event: rd.EngineEvent) -> None:
    print(f"Limit triggered at epoch {event.index}")


@rd.on_log  # EventType.LOG
def log_log(event: rd.EngineEvent) -> None:
    print(f"Log event: {event}")


@rd.on_checkpoint_saved  # EventType.CHECKPOINT_SAVED
def log_checkpoint_saved(event: rd.EngineEvent) -> None:
    print(f"Checkpoint saved at epoch {event.index}")


@rd.on_event  # subscribe to all events
def log_event(event: rd.EngineEvent) -> None:
    print(f"Event received: {event}")


# Each decorated function is already a full handler, so subscribe them directly
engine = (
    rd.Engine.int(10, init_range=(0, 100))
    .fitness(your_fitness_func)
    .subscribe(
        log_start,
        log_epoch,
        log_improvement,
        log_stop,
        log_limit_triggered,
        log_log,
        log_checkpoint_saved,
        log_event,
    )
    # ... other parameters ...
)

Decorators are a Python-only convenience; use the callback or EventHandler forms above in Rust.

Built in Handlers

Python Rust

As of 4/25/2026, the python implementation includes one built in event handler called the MetricCollector. This handler collects the metric set at the end of each epoch and stores it in a list for later use. Note to use this handler to its fullest capacity, you should install radiate with the polars (or pandas) and plot extras, as shown below:

uv add "radiate[polars,pandas,plot]"

You can use this handler as follows (this is great when using radiate inside a .ipynb notebook):

import radiate as rd

# Create an instance of the MetricCollector
collector = rd.MetricCollector()

engine = (
    rd.Engine.float(2, init_range=(0.0, 1.0))  # configure your engine as normal
    .fitness(your_fitness_func)
    .subscribe(collector)  # Subscribe the MetricCollector to the engine
    .limit(rd.Limit.generations(100))  # Set a limit for the run
    # ... other parameters ...
)

# Run the engine for 100 generations
engine.run()

# After the run, you can access the collected metrics
# Convert collected metric sets to a df where each row is a single metric (includes all collected metrics).
df = collector.to_polars(lazy=False)  # optional lazy arg - defaults to False

# Same as above but with pandas instead of polars
df = collector.to_pandas()

# Plot specific metrics to a plotly line plot
collector.plot("scores.best", "pct.diversity")

No built-in handlers in rust yet.

---

Best Practices

1. Keep Event Handlers Light:

   • Event handlers are called frequently during evolution
   • Avoid heavy computations in event handlers

2. Use Multiple Subscribers:

   • You can subscribe multiple handlers to the same engine
   • Separate concerns into different handlers
     • Example: one for logging, one for metrics, one for visualization

3. Handle Errors Gracefully:

   • Event handlers should not crash the evolution process
   • Log errors instead of raising exceptions - do not expect the GeneticEngine to throw exceptions

4. Monitor Performance:

   • Be aware that event handling adds some overhead depending on your implementation
   • Use built in metrics to track certain metrics or performance characteristics if possible
   • Be cautious of your implementation - consider disabling event handling in production if not essential