Skip to content

pipeline

logger = logging.getLogger(__name__) module-attribute

EvaluatorPipelineBuilder dataclass

Bases: Builder

Builder to facilitate construction of evaluator.

Provides methods to set specific values for the evaluator and enforce checks such that the evaluator can be constructed correctly and to avoid possible errors when the evaluator is executed.

Source code in src/recnexteval/evaluators/builder/pipeline.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass
class EvaluatorPipelineBuilder(Builder):
    """Builder to facilitate construction of evaluator.

    Provides methods to set specific values for the evaluator and enforce checks
    such that the evaluator can be constructed correctly and to avoid possible
    errors when the evaluator is executed.
    """

    ignore_unknown_user: bool = False
    """Ignore unknown user in the evaluation"""
    ignore_unknown_item: bool = False
    """Ignore unknown item in the evaluation"""
    algo_state_mgr: AlgorithmStateManager = field(default_factory=AlgorithmStateManager)
    """Algorithm state manager for pipeline evaluation"""

    def add_algorithm(
        self,
        algorithm: type[Algorithm],
        params: dict[str, int] = {},
        algo_uuid: None | uuid.UUID = None,
    ) -> None:
        """Add algorithm to evaluate.

        Adding algorithm to evaluate on. The algorithm can be added by specifying the class type
        or by specifying the class name as a string.

        Args:
            algorithm: Algorithm to evaluate.
            params: Parameter for the algorithm.

        Raises:
            ValueError: If algorithm is not found in ALGORITHM_REGISTRY.
        """
        if not self._check_setting_exist():
            raise RuntimeError(
                "Setting has not been set. To ensure conformity, of the addition of"
                " other components please set the setting first. Call add_setting() method."
            )

        self.algo_state_mgr.register(algorithm_ptr=algorithm, params=params, algo_uuid=algo_uuid)

    def _check_ready(self) -> None:
        """Check if the builder is ready to construct Evaluator.

        Raises:
            RuntimeError: If there are invalid configurations.
        """
        super()._check_ready()

        if len(self.algo_state_mgr) == 0:
            raise RuntimeError("No algorithms specified, can't construct Evaluator")

        for algo_state in self.algo_state_mgr.values():
            if (
                algo_state.params is not None
                and "K" in algo_state.params
                and algo_state.params["K"] < self.setting.top_K
            ):
                warn(
                    f"Algorithm {algo_state.name} has K={algo_state.params['K']} but setting"
                    f" is configured top_K={self.setting.top_K}. The number of predictions"
                    " returned by the model is less than the K value to evaluate the predictions"
                    " on. This may distort the metric value."
                )

    def build(self) -> EvaluatorPipeline:
        """Build Evaluator object.

        Raises:
            RuntimeError: If no metrics, algorithms or settings are specified.

        Returns:
            EvaluatorPipeline: The built evaluator object.
        """
        self._check_ready()
        return EvaluatorPipeline(
            algo_state_mgr=self.algo_state_mgr,
            metric_entries=list(self.metric_entries.values()),
            setting=self.setting,
            metric_k=self.metric_k,
            ignore_unknown_user=self.ignore_unknown_user,
            ignore_unknown_item=self.ignore_unknown_item,
            seed=self.seed,
        )

ignore_unknown_user = False class-attribute instance-attribute

Ignore unknown user in the evaluation

ignore_unknown_item = False class-attribute instance-attribute

Ignore unknown item in the evaluation

algo_state_mgr = field(default_factory=AlgorithmStateManager) class-attribute instance-attribute

Algorithm state manager for pipeline evaluation

setting = field(init=False) class-attribute instance-attribute

Setting to evaluate the algorithms on

metric_k = 10 class-attribute instance-attribute

K value for metrics

seed = 42 class-attribute instance-attribute

Random seed for reproducibility

metric_entries = field(default_factory=dict) class-attribute instance-attribute

dict of metrics to evaluate algorithm on. Using dict instead of list for fast lookup

add_algorithm(algorithm, params={}, algo_uuid=None)

Add algorithm to evaluate.

Adding algorithm to evaluate on. The algorithm can be added by specifying the class type or by specifying the class name as a string.

Parameters:

Name Type Description Default
algorithm type[Algorithm]

Algorithm to evaluate.

required
params dict[str, int]

Parameter for the algorithm.

{}

Raises:

Type Description
ValueError

If algorithm is not found in ALGORITHM_REGISTRY.

Source code in src/recnexteval/evaluators/builder/pipeline.py
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def add_algorithm(
    self,
    algorithm: type[Algorithm],
    params: dict[str, int] = {},
    algo_uuid: None | uuid.UUID = None,
) -> None:
    """Add algorithm to evaluate.

    Adding algorithm to evaluate on. The algorithm can be added by specifying the class type
    or by specifying the class name as a string.

    Args:
        algorithm: Algorithm to evaluate.
        params: Parameter for the algorithm.

    Raises:
        ValueError: If algorithm is not found in ALGORITHM_REGISTRY.
    """
    if not self._check_setting_exist():
        raise RuntimeError(
            "Setting has not been set. To ensure conformity, of the addition of"
            " other components please set the setting first. Call add_setting() method."
        )

    self.algo_state_mgr.register(algorithm_ptr=algorithm, params=params, algo_uuid=algo_uuid)

build()

Build Evaluator object.

Raises:

Type Description
RuntimeError

If no metrics, algorithms or settings are specified.

Returns:

Name Type Description
EvaluatorPipeline EvaluatorPipeline

The built evaluator object.

Source code in src/recnexteval/evaluators/builder/pipeline.py
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def build(self) -> EvaluatorPipeline:
    """Build Evaluator object.

    Raises:
        RuntimeError: If no metrics, algorithms or settings are specified.

    Returns:
        EvaluatorPipeline: The built evaluator object.
    """
    self._check_ready()
    return EvaluatorPipeline(
        algo_state_mgr=self.algo_state_mgr,
        metric_entries=list(self.metric_entries.values()),
        setting=self.setting,
        metric_k=self.metric_k,
        ignore_unknown_user=self.ignore_unknown_user,
        ignore_unknown_item=self.ignore_unknown_item,
        seed=self.seed,
    )

set_metric_k(K)

Set K value for all metrics.

Parameters:

Name Type Description Default
K int

K value to set for all metrics.

required
Source code in src/recnexteval/evaluators/builder/base.py
49
50
51
52
53
54
55
def set_metric_k(self, K: int) -> None:
    """Set K value for all metrics.

    Args:
        K: K value to set for all metrics.
    """
    self.metric_k = K

add_metric(metric)

Add metric to evaluate algorithm on.

Metric will be added to the metric_entries dict where it will later be converted to a list when the evaluator is constructed.

Note

If K is not yet specified, the setting's top_K value will be used. This requires the setting to be set before adding the metric.

Parameters:

Name Type Description Default
metric str | type

Metric to evaluate algorithm on.

required

Raises:

Type Description
ValueError

If metric is not found in METRIC_REGISTRY.

RuntimeError

If setting is not set.

Source code in src/recnexteval/evaluators/builder/base.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def add_metric(self, metric: str | type) -> None:
    """Add metric to evaluate algorithm on.

    Metric will be added to the metric_entries dict where it will later be
    converted to a list when the evaluator is constructed.

    Note:
        If K is not yet specified, the setting's top_K value will be used. This
        requires the setting to be set before adding the metric.

    Args:
        metric: Metric to evaluate algorithm on.

    Raises:
        ValueError: If metric is not found in METRIC_REGISTRY.
        RuntimeError: If setting is not set.
    """
    if not self._check_setting_exist():
        raise RuntimeError(
            "Setting has not been set. To ensure conformity, of the addition of"
            " other components please set the setting first. Call add_setting() method."
        )

    metric = arg_to_str(metric)

    if metric not in METRIC_REGISTRY:
        raise ValueError(f"Metric {metric} could not be resolved.")

    if not hasattr(self, "metric_k"):
        self.metric_k = self.setting.top_K
        warn(
            "K value not yet specified before setting metric, using setting's top_K value."
            " We recommend specifying K value for metric. If you want to change the K value,"
            " you can clear all metric entry and set the K value before adding metrics."
        )

    metric_name = f"{metric}_{self.metric_k}"
    if metric_name in self.metric_entries:
        logger.warning(f"Metric {metric_name} already exists. Skipping adding metric.")
        return

    self.metric_entries[metric_name] = MetricEntry(metric, self.metric_k)

add_setting(setting)

Add setting to the evaluator builder.

Note

The setting should be set before adding metrics or algorithms to the evaluator.

Parameters:

Name Type Description Default
setting Setting

Setting to evaluate the algorithms on.

required

Raises:

Type Description
ValueError

If setting is not of instance Setting.

Source code in src/recnexteval/evaluators/builder/base.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def add_setting(self, setting: Setting) -> None:
    """Add setting to the evaluator builder.

    Note:
        The setting should be set before adding metrics or algorithms
        to the evaluator.

    Args:
        setting: Setting to evaluate the algorithms on.

    Raises:
        ValueError: If setting is not of instance Setting.
    """
    if not isinstance(setting, Setting):
        raise ValueError(f"setting should be of type Setting, got {type(setting)}")
    if hasattr(self, "setting") and self.setting is not None:
        warn("Setting is already set. Continuing will overwrite the setting.")

    self.setting = setting

clear_metrics()

Clear all metrics from the builder.

Source code in src/recnexteval/evaluators/builder/base.py
120
121
122
def clear_metrics(self) -> None:
    """Clear all metrics from the builder."""
    self.metric_entries.clear()