Labelling Functions

Labelling functions bridge the gap between low-level environment observations and high-level symbolic events that RM/CRMs understand.

Overview

In the RM/CRM framework, labelling functions play a critical role in abstracting raw environment observations into meaningful symbolic events. These events form the vocabulary that the RM/CRM uses to define rewards and transitions.

Core Concept

A labelling function maps an environment transition (observation, action, next observation) to a set of symbolic events (a truth assignment for a set of propositions). This abstraction layer provides several benefits:

Modularity: Separate environment dynamics from task specification
Reusability: Same environment can support different tasks by changing the labelling
Expressiveness: Define complex tasks in terms of abstract events rather than raw states

Implementation

Base Class

The RM/CRM framework provides a base LabellingFunction class that handles the core functionality:

from abc import ABC
from enum import Enum
from typing import Any, Generic, TypeVar

ObsType = TypeVar("ObsType")  # Type of observations
ActType = TypeVar("ActType")  # Type of actions

class LabellingFunction(ABC, Generic[ObsType, ActType]):
    """Base class for labelling functions."""
    
    @staticmethod
    def event(func):
        """Decorator to register an event detection method."""
        setattr(func, "_is_event_method", True)
        return func
        
    def __call__(self, obs: ObsType, action: ActType, next_obs: ObsType) -> set[Enum]:
        """Detect all events in a transition."""
        # Implementation details...

The base class provides:

Type parameters for observation and action types
An event decorator to identify event detection methods
A __call__ method that runs all event detectors on a transition

Creating a Labelling Function

To create a custom labelling function, follow these steps:

Define your events as an Enum:

class Symbol(Enum):
    A = auto()
    B = auto()
    C = auto()

Create a subclass of LabellingFunction:

class MyLabellingFunction(LabellingFunction[MyObsType, MyActType]):
    # Implementation...

Implement event detection methods with the @event decorator:

@LabellingFunction.event
def detect_event_x(self, obs, action, next_obs) -> Symbol | None:
    # Logic to detect event X
    if condition_met:
        return Symbol.X
    return None

Each event method should:

Accept the parameters (self, obs, action, next_obs)
Return an Enum value when the event is detected
Return None when the event is not detected

How It Works

When you call a labelling function with lf(obs, action, next_obs), it:

Looks for all methods decorated with @LabellingFunction.event
Calls each method with the provided transition
Collects all non-None results into a set
Returns this set of detected events

Example: Letter World

Let’s examine a complete example from the Letter World environment:

from enum import Enum, auto
import numpy as np
from pycrm.label import LabellingFunction

class Symbol(Enum):
    """Symbols in the Letter World environment."""
    A = auto()
    B = auto()
    C = auto()

class LetterWorldLabellingFunction(LabellingFunction[np.ndarray, int]):
    """Labelling function for the Letter World environment."""

    A_B_POSITION = np.array([1, 1])
    C_POSITION = np.array([1, 5])

    @LabellingFunction.event
    def test_symbol_a(
        self, obs: np.ndarray, action: np.ndarray, next_obs: np.ndarray
    ) -> Symbol | None:
        """Return if the agent observes symbol A."""
        if next_obs[0] == 0 and np.array_equal(next_obs[1:], self.A_B_POSITION):
            return Symbol.A
        return None

    @LabellingFunction.event
    def test_symbol_b(
        self, obs: np.ndarray, action: np.ndarray, next_obs: np.ndarray
    ) -> Symbol | None:
        """Return if the agent observes symbol B."""
        if next_obs[0] == 1 and np.array_equal(next_obs[1:], self.A_B_POSITION):
            return Symbol.B
        return None

    @LabellingFunction.event
    def test_symbol_c(
        self, obs: np.ndarray, action: np.ndarray, next_obs: np.ndarray
    ) -> Symbol | None:
        """Return if the agent observes symbol C."""
        if next_obs[0] == 1 and np.array_equal(next_obs[1:], self.C_POSITION):
            return Symbol.C
        return None

This example:

Defines three symbols (A, B, C) as events
Creates a labelling function that works with numpy array observations and integer actions
Implements three event detection methods, each checking specific conditions
Returns the appropriate symbol when conditions are met

Using Labelling Functions

Once defined, you use a labelling function as part of a cross-product environment:

# Create components
ground_env = LetterWorld()
lf = LetterWorldLabellingFunction()
crm = LetterWorldCountingRewardMachine()

# Create cross-product environment
env = LetterWorldCrossProduct(
    ground_env=ground_env,
    crm=crm,
    lf=lf,
    max_steps=500,
)

# When env.step() is called, the labelling function
# automatically converts observations to events
next_obs, reward, terminated, truncated, info = env.step(action)
detected_events = info.get('events', [])  # Events detected in this step

Best Practices

When creating labelling functions:

Keep events meaningful: Define events that have semantic meaning for your task
Be consistent: Ensure event detection is consistent and deterministic
Handle edge cases: Consider all possible transitions and what they should mean
Document conditions: Clearly document when each event is triggered
Use clear naming: Name your event detection methods descriptively

Summary

Labelling functions are a powerful abstraction mechanism in the RM/CRM framework. They:

Convert low-level observations to high-level events
Allow for modular task specification
Enable the same environment to support different tasks
Provide a clean separation between environment dynamics and reward logic

By properly designing your labelling function, you can create complex tasks that would be difficult to specify using traditional reward functions.

Get Started

Example

Code Concepts

Labelling Functions

Labelling Functions

Overview

Core Concept

Implementation

Base Class

Creating a Labelling Function

How It Works

Example: Letter World

Using Labelling Functions

Best Practices

Summary

​Labelling Functions

​Overview

​Core Concept

​Implementation

​Base Class

​Creating a Labelling Function

​How It Works

​Example: Letter World

​Using Labelling Functions

​Best Practices

​Summary

Labelling Functions

Overview

Core Concept

Implementation

Base Class

Creating a Labelling Function

How It Works

Example: Letter World

Using Labelling Functions

Best Practices

Summary