Skip to content

Configuration

RecDistillery experiments are assembled from YAML configuration files. The config tree separates reusable defaults from complete experiment definitions.

Config Layout

config/
  dataset/       dataset definitions
  teacher/       teacher model defaults
  student/       student model defaults
  distillation/  distiller defaults
  optimization/  optimizer and scheduler defaults
  runtime/       runtime options
  evaluation/    metric and top-k defaults
  composites/    reusable experiment templates
  experiments/   complete runnable experiments

Complete Experiments

config/experiments/teacher/
config/experiments/student/
config/experiments/recdistill/

Config Integration

RecDistill configuration integration with the new centralized config system.

load_recdistill_config_from_file(config_path: Union[str, Path]) -> RecDistillConfig

Load and validate RecDistill configuration from YAML file.

Parameters:

Name Type Description Default
config_path Union[str, Path]

Path to configuration YAML file

required

Returns:

Type Description
RecDistillConfig

Validated RecDistillConfig object

Raises:

Type Description
FileNotFoundError

If config file doesn't exist

ValidationError

If config doesn't match schema

Source code in recdistill/config_integration.py
def load_recdistill_config_from_file(config_path: Union[str, Path]) -> RecDistillConfig:
    """
    Load and validate RecDistill configuration from YAML file.

    Args:
        config_path: Path to configuration YAML file

    Returns:
        Validated RecDistillConfig object

    Raises:
        FileNotFoundError: If config file doesn't exist
        ValidationError: If config doesn't match schema
    """
    path = Path(config_path)
    if not path.exists():
        raise FileNotFoundError(f"Configuration file not found: {path}")

    with path.open("r", encoding="utf-8") as fp:
        raw_config = yaml.safe_load(fp) or {}

    normalized = normalize_recdistill_config(raw_config)
    return RecDistillConfig(**normalized)

load_recdistill_experiment(dataset_name: str, teacher_model: str, distiller_strategy: str, student_backbone: Optional[str] = None, teacher_framework: str = 'recbole', student_framework: str = 'recbole', overrides: Optional[Dict[str, Any]] = None) -> RecDistillConfig

Programmatically load RecDistill experiment with centralized configurations.

Parameters:

Name Type Description Default
dataset_name str

Dataset name (amazon_cd, bookcrossing, citeulike)

required
teacher_model str

Teacher model (bprmf, nmf, lgcn)

required
distiller_strategy str

Distillation strategy (de, htd, ftd, unkd)

required
student_backbone Optional[str]

Student backbone (default: same as teacher)

None
overrides Optional[Dict[str, Any]]

Dict of configuration overrides

None

Returns:

Type Description
RecDistillConfig

Validated RecDistillConfig object

Example

config = load_recdistill_experiment( ... dataset_name='citeulike', ... teacher_model='nmf', ... distiller_strategy='de', ... overrides={ ... 'distill_student.optimization.epochs': 50, ... 'distill_student.runtime.seed': 123 ... } ... )

Source code in recdistill/config_integration.py
def load_recdistill_experiment(
    dataset_name: str,
    teacher_model: str,
    distiller_strategy: str,
    student_backbone: Optional[str] = None,
    teacher_framework: str = "recbole",
    student_framework: str = "recbole",
    overrides: Optional[Dict[str, Any]] = None
) -> RecDistillConfig:
    """
    Programmatically load RecDistill experiment with centralized configurations.

    Args:
        dataset_name: Dataset name (amazon_cd, bookcrossing, citeulike)
        teacher_model: Teacher model (bprmf, nmf, lgcn)
        distiller_strategy: Distillation strategy (de, htd, ftd, unkd)
        student_backbone: Student backbone (default: same as teacher)
        overrides: Dict of configuration overrides

    Returns:
        Validated RecDistillConfig object

    Example:
        >>> config = load_recdistill_experiment(
        ...     dataset_name='citeulike',
        ...     teacher_model='nmf',
        ...     distiller_strategy='de',
        ...     overrides={
        ...         'distill_student.optimization.epochs': 50,
        ...         'distill_student.runtime.seed': 123
        ...     }
        ... )
    """
    if student_backbone is None:
        student_backbone = teacher_model

    loader = get_config_loader()
    config_dict = loader.compose_recdistill_experiment(
        dataset_name=dataset_name,
        teacher_model=teacher_model,
        distiller_strategy=distiller_strategy,
        student_backbone=student_backbone,
        teacher_framework=teacher_framework,
        student_framework=student_framework,
    )

    # Apply overrides if provided
    if overrides:
        config_dict = _apply_overrides(config_dict, overrides)

    return RecDistillConfig(**normalize_recdistill_config(config_dict))

normalize_recdistill_config(config: Dict[str, Any]) -> Dict[str, Any]

Normalize RecDistill config dictionaries into RecDistillConfig shape.

Source code in recdistill/config_integration.py
def normalize_recdistill_config(config: Dict[str, Any]) -> Dict[str, Any]:
    """
    Normalize RecDistill config dictionaries into RecDistillConfig shape.
    """
    if isinstance(config, dict) and "preset" in config and "config" in config:
        config = config["config"]

    if isinstance(config, dict):
        config = get_config_loader().resolve_config_modules(config)

    normalized = copy.deepcopy(config or {})
    if "distill_student" not in normalized:
        raise ValueError("RecDistill configs must define a top-level 'distill_student' block.")
    train_conf = normalized.get("distill_student") or {}
    normalized["distill_student"] = train_conf

    student_conf = train_conf.setdefault("student", {})
    teacher_conf = train_conf.setdefault("teacher", {})
    distill_conf = train_conf.setdefault("distillation", {})

    strategy = (
        distill_conf.get("strategy")
        or _infer_strategy_from_distillation_block(distill_conf)
        or "DE"
    )
    strategy = str(strategy).replace("-", "_").upper()
    distill_conf["strategy"] = strategy
    student_conf.pop("model", None)
    active_methods = {part for part in strategy.replace("+", "_").split("_") if part}
    if "DE" not in active_methods:
        distill_conf["lambda_de"] = 0.0

    if "model" in teacher_conf and teacher_conf["model"] is not None:
        teacher_conf["model"] = str(teacher_conf["model"]).upper()
    if "backbone" in student_conf and student_conf["backbone"] is not None:
        student_conf["backbone"] = str(student_conf["backbone"]).upper()

    for key in (
        "temperature",
        "lambda_de",
        "lambda_kl",
        "lambda_rrd",
        "lambda_unkd",
        "num_experts",
    ):
        if key in student_conf and key not in distill_conf:
            distill_conf[key] = student_conf[key]

    optimization_conf = train_conf.setdefault("optimization", {})
    if "early_stopping" in train_conf:
        raise ValueError("Use distill_student.optimization.early_stopping, not distill_student.early_stopping.")
    optimization_conf.setdefault("early_stopping", {})
    train_conf.setdefault("runtime", {})
    train_conf.setdefault("evaluation", {})
    return normalized

recdistill_config_to_dict(config: Union[RecDistillConfig, Dict[str, Any]]) -> Dict[str, Any]

Return a plain normalized dictionary for runner code.

Source code in recdistill/config_integration.py
def recdistill_config_to_dict(config: Union[RecDistillConfig, Dict[str, Any]]) -> Dict[str, Any]:
    """Return a plain normalized dictionary for runner code."""
    if isinstance(config, RecDistillConfig):
        return config.model_dump()
    return normalize_recdistill_config(config)

get_default_dataset_path(dataset_name: str, file_type: str = 'train') -> str

Get default data path for a dataset.

Parameters:

Name Type Description Default
dataset_name str

Dataset name

required
file_type str

'train', 'val', or 'test'

'train'

Returns:

Type Description
str

Path to dataset file

Example

path = get_default_dataset_path('citeulike', 'train') print(path) data/citeulike/train.tsv

Source code in recdistill/config_integration.py
def get_default_dataset_path(dataset_name: str, file_type: str = 'train') -> str:
    """
    Get default data path for a dataset.

    Args:
        dataset_name: Dataset name
        file_type: 'train', 'val', or 'test'

    Returns:
        Path to dataset file

    Example:
        >>> path = get_default_dataset_path('citeulike', 'train')
        >>> print(path)
        data/citeulike/train.tsv
    """
    loader = get_config_loader()
    dataset_cfg = loader.load_dataset_config(dataset_name)

    if file_type == 'train':
        return dataset_cfg.train_path
    elif file_type == 'val':
        return dataset_cfg.validation_path or dataset_cfg.train_path
    elif file_type == 'test':
        return dataset_cfg.test_path
    else:
        raise ValueError(f"Unknown file_type: {file_type}")

print_config_summary(config: RecDistillConfig) -> None

Print a summary of RecDistill configuration.

Parameters:

Name Type Description Default
config RecDistillConfig

RecDistillConfig object

required
Source code in recdistill/config_integration.py
def print_config_summary(config: RecDistillConfig) -> None:
    """
    Print a summary of RecDistill configuration.

    Args:
        config: RecDistillConfig object
    """
    train_cfg = config.distill_student

    print("\n" + "=" * 70)
    print("RECDISTILL CONFIGURATION SUMMARY")
    print("=" * 70)

    print("\n📊 DATA")
    print(f"  Dataset: {train_cfg.dataset}")

    print("\n👨‍🏫 TEACHER")
    print(f"  Model: {train_cfg.teacher.model}")
    print(f"  Embedding Dim: {train_cfg.teacher.embedding_dim}")
    print(f"  Path: {train_cfg.teacher.path or 'Not specified (will be auto-located)'}")

    print("\n👨‍🎓 STUDENT")
    print(f"  Backbone: {train_cfg.student.backbone}")
    print(f"  Distiller: {train_cfg.distillation.strategy}")
    print(f"  Embedding Dim: {train_cfg.student.embedding_dim}")

    print("\n🔬 DISTILLATION")
    print(f"  Strategy: {train_cfg.distillation.strategy}")
    print(f"  Temperature: {train_cfg.distillation.temperature}")
    if hasattr(train_cfg.distillation, 'lambda_kl'):
        print(f"  Lambda KL: {train_cfg.distillation.lambda_kl}")
    if hasattr(train_cfg.distillation, 'lambda_de'):
        print(f"  Lambda DE: {train_cfg.distillation.lambda_de}")

    print("\n⚙️  OPTIMIZATION")
    print(f"  Epochs: {train_cfg.optimization.epochs}")
    print(f"  Batch Size: {train_cfg.optimization.batch_size}")
    print(f"  Learning Rate: {train_cfg.optimization.learning_rate}")
    print(f"  L2 Reg: {train_cfg.optimization.l2_reg}")

    print("\n🏃 RUNTIME")
    print(f"  Seed: {train_cfg.runtime.seed}")
    print(f"  Device: {train_cfg.runtime.device or 'Auto (cuda if available)'}")
    print(f"  Num Workers: {train_cfg.runtime.num_workers}")
    print(f"  Output Path: {train_cfg.runtime.output_path}")
    if train_cfg.runtime.wandb.get('enabled'):
        print(f"  W&B: Enabled")

    print("\n📈 EVALUATION")
    print(f"  K: {train_cfg.evaluation.k}")
    print(f"  Every N epochs: {train_cfg.evaluation.every}")
    print(f"  Validation Only: {train_cfg.evaluation.val_only}")
    print(f"  Selection Metric: {train_cfg.evaluation.selection_metric}")

    print("\n⏸️  EARLY STOPPING")
    early_stopping = train_cfg.optimization.early_stopping
    if early_stopping:
        print(f"  Enabled: {early_stopping.enabled}")
        print(f"  Metric: {early_stopping.metric}")
        print(f"  Patience: {early_stopping.patience}")
        print(f"  Min Delta: {early_stopping.min_delta}")
    else:
        print(f"  Disabled")

    print("\n" + "=" * 70 + "\n")

validate_config(config: RecDistillConfig) -> bool

Validate RecDistill configuration for common issues.

Parameters:

Name Type Description Default
config RecDistillConfig

RecDistillConfig object

required

Returns:

Type Description
bool

True if configuration is valid, False otherwise

Source code in recdistill/config_integration.py
def validate_config(config: RecDistillConfig) -> bool:
    """
    Validate RecDistill configuration for common issues.

    Args:
        config: RecDistillConfig object

    Returns:
        True if configuration is valid, False otherwise
    """
    train_cfg = config.distill_student
    issues = []

    # Check embedding dimensions
    if train_cfg.student.embedding_dim >= train_cfg.teacher.embedding_dim:
        issues.append(
            f"⚠️  Student embedding dim ({train_cfg.student.embedding_dim}) "
            f"should be smaller than teacher ({train_cfg.teacher.embedding_dim})"
        )

    # Check early stopping if enabled
    early_stopping = train_cfg.optimization.early_stopping
    if early_stopping and early_stopping.enabled:
        if early_stopping.patience > train_cfg.optimization.epochs:
            issues.append(
                f"Early stopping patience ({early_stopping.patience}) "
                f"should be smaller than total epochs ({train_cfg.optimization.epochs})"
            )

    if issues:
        print("\n⚠️  Configuration warnings:")
        for issue in issues:
            print(f"  {issue}")
        return False

    print("Configuration validation passed.")
    return True

list_example_experiments() -> None

Print available pre-configured experiments.

Source code in recdistill/config_integration.py
def list_example_experiments() -> None:
    """Print available pre-configured experiments."""
    loader = get_config_loader()

    print("\n" + "=" * 70)
    print("AVAILABLE EXPERIMENTS")
    print("=" * 70)

    datasets = loader.list_datasets()
    models = loader.list_models()
    distillers = loader.list_distillers()

    print("\nPredefine combinations:")
    for dataset in datasets:
        for distiller in distillers:
            for teacher in models['teacher']:
                print(f"  python train_distiller.py \\")
                print(f"    --dataset {dataset} \\")
                print(f"    --teacher {teacher} \\")
                print(f"    --distiller {distiller}")

    print("\n" + "=" * 70 + "\n")

Config Loader

Configuration loader with validation. Centralizes loading and validation of all configuration files.

ConfigLoader

Unified configuration loader for RecDistill.

Source code in config/config_loader.py
 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
class ConfigLoader:
    """Unified configuration loader for RecDistill."""

    def __init__(self, config_root: Optional[Path] = None):
        """
        Initialize config loader.

        Args:
            config_root: Root directory for config files. Defaults to ./config
        """
        if config_root is None:
            config_root = Path(__file__).parent

        self.root = Path(config_root)
        self._cache: Dict[str, Any] = {}

    def _load_yaml(self, filepath: Path) -> Dict[str, Any]:
        """Load YAML file with caching."""
        filepath = Path(filepath)
        if not filepath.is_absolute():
            filepath = self.root / filepath

        if str(filepath) in self._cache:
            return self._cache[str(filepath)]

        if not filepath.exists():
            raise FileNotFoundError(f"Configuration file not found: {filepath}")

        with open(filepath, 'r', encoding='utf-8-sig') as f:
            data = yaml.safe_load(f)

        self._cache[str(filepath)] = data
        return data

    def _resolve_config_path(self, path: Union[str, Path]) -> Path:
        config_path = Path(path)
        return config_path if config_path.is_absolute() else self.root / config_path

    def load_module_config(self, module_path: Union[str, Path], overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Load a module config and apply optional overrides."""
        data = copy.deepcopy(self._load_yaml(self._resolve_config_path(module_path)) or {})
        if overrides:
            data = _deep_merge(data, overrides)
        return data

    def resolve_config_modules(self, config: Any) -> Any:
        """Resolve `default: path/to/module.yaml` sections recursively.

        Paths may be absolute or relative to the config root. Sibling keys in
        the same mapping override the loaded defaults.
        """
        if isinstance(config, list):
            return [self.resolve_config_modules(item) for item in config]
        if not isinstance(config, dict):
            return config

        if "default" in config:
            default_path = config.get("default")
            if default_path is None:
                base: Dict[str, Any] = {}
            elif isinstance(default_path, (str, Path)):
                base = self.resolve_config_modules(self.load_module_config(default_path))
            else:
                raise TypeError("Config module `default` must be a path string.")
            overrides = {
                key: self.resolve_config_modules(value)
                for key, value in config.items()
                if key != "default"
            }
            return _deep_merge(base, overrides)

        return {key: self.resolve_config_modules(value) for key, value in config.items()}

    def load_recdistill_config(self, config_path: Union[str, Path]) -> RecDistillConfig:
        """
        Load and validate RecDistill configuration.

        Args:
            config_path: Path to RecDistill config YAML file

        Returns:
            Validated RecDistillConfig object

        Raises:
            FileNotFoundError: If config file doesn't exist
            ValidationError: If config doesn't match schema
        """
        data = self._load_yaml(config_path)
        if isinstance(data, dict) and "preset" in data and "config" in data:
            data = data["config"]
        data = self.resolve_config_modules(data)
        from recdistill.config_integration import normalize_recdistill_config

        data = normalize_recdistill_config(data)
        try:
            return RecDistillConfig(**data)
        except ValidationError as e:
            raise ValueError(f"Invalid RecDistill config in {config_path}:\n{e}")

    def load_preset(self, preset_path: Union[str, Path]) -> ConfigPreset:
        """
        Load a wrapped experiment preset.

        Presets preserve provenance metadata under `preset` and keep the actual
        training payload under `config`.
        """
        data = self._load_yaml(preset_path)
        try:
            return ConfigPreset(**data)
        except ValidationError as e:
            raise ValueError(f"Invalid config preset in {preset_path}:\n{e}")

    def load_recdistill_preset(self, preset_path: Union[str, Path]) -> RecDistillConfig:
        """Load a RecDistill preset and validate its inner config."""
        from recdistill.config_integration import normalize_recdistill_config

        preset = self.load_preset(preset_path)
        if preset.preset.kind.lower() != "recdistill":
            raise ValueError(f"Preset is not a RecDistill preset: {preset_path}")
        try:
            return RecDistillConfig(**normalize_recdistill_config(preset.config))
        except ValidationError as e:
            raise ValueError(f"Invalid RecDistill config inside preset {preset_path}:\n{e}")

    def load_dataset_config(self, dataset_name: str) -> DataConfig:
        """
        Load dataset configuration.

        Args:
            dataset_name: Name of dataset (amazon_cd, bookcrossing, citeulike)

        Returns:
            Validated DataConfig object
        """
        config_path = self.root / "dataset" / f"{dataset_name}.yaml"
        data = self._load_yaml(config_path)
        try:
            return DataConfig(**data)
        except ValidationError as e:
            raise ValueError(f"Invalid dataset config for {dataset_name}:\n{e}")

    def load_model_config(self, model_type: str, model_name: str, framework: str | None = None) -> ModelConfig:
        """
        Load model configuration.

        Args:
            model_type: Type of model ('teacher' or 'student')
            model_name: Name of model (bprmf, nmf, lgcn, etc.)

        Returns:
            Validated ModelConfig object
        """
        model_slug = str(model_name).strip().lower()
        framework_slug = str(framework).strip().lower() if framework else None
        if framework_slug:
            config_path = self.root / model_type / framework_slug / f"{model_slug}.yaml"
        else:
            config_path = self.root / model_type / f"{model_slug}.yaml"
            if not config_path.exists():
                default_path = self.root / model_type / "recbole" / f"{model_slug}.yaml"
                if default_path.exists():
                    config_path = default_path
                else:
                    matches = sorted((self.root / model_type).glob(f"*/{model_slug}.yaml"))
                    if len(matches) == 1:
                        config_path = matches[0]
                    elif len(matches) > 1:
                        choices = ", ".join(path.parent.name for path in matches)
                        raise ValueError(
                            f"Ambiguous model config for {model_type}/{model_name}. "
                            f"Specify framework. Available frameworks: {choices}."
                        )
        data = self._load_yaml(config_path)
        try:
            return ModelConfig(**data)
        except ValidationError as e:
            raise ValueError(f"Invalid model config for {model_type}/{model_name}:\n{e}")

    def compose_teacher_training(
        self,
        dataset_name: str,
        model_name: str,
        framework: str = "recbole",
        model_type: str = "teacher",
    ) -> Dict[str, Any]:
        """
        Compose a generic teacher training configuration.

        Args:
            dataset_name: Dataset name
            model_name: Model name
            framework: Framework implementation to use
            model_type: Model family to load ('teacher' or 'student')

        Returns:
            Complete teacher training configuration dictionary
        """
        self.load_dataset_config(dataset_name)
        model = self.load_model_config(model_type, model_name, framework=framework)

        config_path = self.root / "composites" / "teacher_template.yaml"
        template = copy.deepcopy(self._load_yaml(config_path))
        train = template.setdefault("train_teacher", {})

        train["dataset"] = dataset_name
        train["teacher"] = {"default": _module_path("teacher", model.framework, model.model)}
        return template

    def compose_student_training(
        self,
        dataset_name: str,
        model_name: str,
        framework: str = "recbole",
        model_type: str = "student",
    ) -> Dict[str, Any]:
        """
        Compose a generic plain-student training configuration.
        """
        self.load_dataset_config(dataset_name)
        model = self.load_model_config(model_type, model_name, framework=framework)

        config_path = self.root / "composites" / "student_template.yaml"
        template = copy.deepcopy(self._load_yaml(config_path))
        train = template.setdefault("train_student", {})

        train["dataset"] = dataset_name
        train["student"] = {"default": _module_path("student", model.framework, model.backbone)}
        return template

    def save_generated_experiment(
        self,
        *,
        kind: str,
        name: str,
        config: Dict[str, Any],
        path_parts: list[str],
        experiment_id: str | None = None,
    ) -> Path:
        """Persist an on-the-fly composed config as an experiment file."""
        del path_parts
        experiment = config.get("experiment") if isinstance(config.get("experiment"), dict) else {}
        experiment_id = normalize_experiment_id(experiment_id or experiment.get("id") or new_experiment_id())
        kind_slug = _slug(kind)
        framework, model, dataset = _experiment_identity(self.resolve_config_modules(config), kind_slug)
        run_dir = experiment_run_dir(
            kind_slug,
            experiment_id,
            framework=framework,
            model=model,
            dataset=dataset,
        )
        experiment["id"] = str(experiment_id)
        experiment.setdefault("name", _slug(name))
        experiment.setdefault("kind", kind_slug)
        root_key = _training_root_key(kind_slug)
        if root_key in config and isinstance(config[root_key], dict):
            runtime = config[root_key].setdefault("runtime", {})
            if isinstance(runtime, dict) and not runtime.get("output_path"):
                runtime["output_path"] = str(
                    run_dir
                    / "artifacts"
                    / experiment_artifact_filename(
                        kind=kind_slug,
                        experiment_id=experiment_id,
                        framework=framework,
                        model=model,
                        dataset=dataset,
                    )
                )
        config = {
            "experiment": experiment,
            **{key: value for key, value in config.items() if key != "experiment"},
        }
        experiment_path = run_dir / "config" / f"{_slug(name)}_{experiment_id}.yaml"
        experiment_path.parent.mkdir(parents=True, exist_ok=True)
        with experiment_path.open("w", encoding="utf-8") as fp:
            yaml.safe_dump(config, fp, sort_keys=False, allow_unicode=False)
        return experiment_path

    def save_generated_preset(
        self,
        *,
        kind: str,
        family: str,
        name: str,
        config: Dict[str, Any],
        path_parts: list[str],
    ) -> Path:
        """Backward-compatible alias for saving generated experiment configs."""
        del family
        return self.save_generated_experiment(
            kind=kind,
            name=name,
            config=config,
            path_parts=path_parts,
        )

    def compose_recdistill_experiment(
        self,
        dataset_name: str,
        teacher_model: str,
        distiller_strategy: str,
        student_backbone: str = None,
        teacher_framework: str = "recbole",
        student_framework: str = "recbole",
    ) -> Dict[str, Any]:
        """
        Compose a complete RecDistill experiment configuration.

        Args:
            dataset_name: Dataset name
            teacher_model: Teacher model (bprmf, nmf, lgcn)
            distiller_strategy: Distillation strategy (de, htd, ftd, unkd)
            student_backbone: Student backbone (default: same as teacher)
            teacher_framework: Framework used to resolve the teacher model config
            student_framework: Framework used to resolve the student backbone config

        Returns:
            Complete distillation configuration dictionary
        """
        if student_backbone is None:
            student_backbone = teacher_model

        dataset = self.load_dataset_config(dataset_name)
        teacher_cfg = self.load_model_config("teacher", teacher_model, framework=teacher_framework)
        student_cfg = self.load_model_config("student", student_backbone, framework=student_framework)
        from recdistill.model_validation import validate_distillation_request

        validate_distillation_request(
            teacher_framework=teacher_cfg.framework,
            teacher_model=teacher_cfg.model,
            student_framework=student_cfg.framework,
            student_backbone=student_cfg.backbone,
            distiller=distiller_strategy,
        )

        config_path = self.root / "composites" / "recdistill_template.yaml"
        template = copy.deepcopy(self._load_yaml(config_path))

        # Merge configs
        train = template["distill_student"]
        train["dataset"] = dataset_name
        train["teacher"] = {"default": _module_path("teacher", teacher_cfg.framework, teacher_cfg.model)}
        train["student"] = {"default": _module_path("student", student_cfg.framework, student_cfg.backbone)}
        train["distillation"] = {
            "default": f"distillation/{_slug(distiller_strategy)}.yaml",
            "strategy": str(distiller_strategy).replace("-", "_").upper(),
        }

        return template

    @staticmethod
    def _copy_model_specific_fields(model: ModelConfig, target: Dict[str, Any]) -> None:
        for key in ("num_layers", "lightgcn_layers", "mlp_hidden_size", "mlp_dims"):
            value = getattr(model, key, None)
            if value is not None:
                target[key] = value

    @staticmethod
    def _remove_none_values(target: Dict[str, Any]) -> None:
        for key in list(target):
            if target[key] is None:
                del target[key]

    @staticmethod
    def _elliot_model_key(backbone: str) -> str:
        mapping = {
            "BPRMF": "torch.BPRMF",
            "BPR": "torch.BPRMF",
            "LGCN": "torch.LightGCN",
            "LIGHTGCN": "torch.LightGCN",
            "NGCF": "torch.NGCF",
            "DGCF": "torch.DGCF",
            "SGL": "torch.SGL",
            "ULTRAGCN": "torch.UltraGCN",
            "ULTRA_GCN": "torch.UltraGCN",
            "NMF": "NeuMFTorch",
            "NFM": "NeuMFTorch",
            "NEUMF": "NeuMFTorch",
        }
        normalized = backbone.upper()
        if normalized not in mapping:
            raise ValueError(f"Unsupported Elliot model backbone: {backbone}")
        return mapping[normalized]

    @staticmethod
    def _elliot_model_label(backbone: str) -> str:
        normalized = backbone.upper()
        if normalized in {"LIGHTGCN"}:
            return "LGCN"
        if normalized in {"ULTRA_GCN"}:
            return "ULTRAGCN"
        if normalized in {"NFM", "NEUMF"}:
            return "NMF"
        return normalized

    def clear_cache(self):
        """Clear configuration cache."""
        self._cache.clear()

    def list_datasets(self) -> list[str]:
        """List available datasets."""
        datasets_dir = self.root / "dataset"
        return [f.stem for f in datasets_dir.glob("*.yaml")]

    def list_models(self, model_type: str = None) -> Dict[str, list[str]]:
        """List available models by type."""
        if model_type:
            type_dir = self.root / model_type
            return {
                model_type: [
                    path.relative_to(type_dir).with_suffix("").as_posix()
                    for path in sorted(type_dir.rglob("*.yaml"))
                ]
            }
        else:
            result = {}
            for model_type_name in ("teacher", "student"):
                type_dir = self.root / model_type_name
                if type_dir.exists():
                    result[model_type_name] = [
                        path.relative_to(type_dir).with_suffix("").as_posix()
                        for path in sorted(type_dir.rglob("*.yaml"))
                    ]
            return result

    def list_distillers(self) -> list[str]:
        """List available distiller strategies."""
        distillers_dir = self.root / "distillation"
        return [f.stem for f in distillers_dir.glob("*.yaml")]

    def list_presets(self, kind: Optional[str] = None) -> list[str]:
        """List experiment files relative to config/experiments."""
        return self.list_experiments(kind)

    def list_experiments(self, kind: Optional[str] = None) -> list[str]:
        """List experiment files relative to config/experiments."""
        experiments_dir = self.root / "experiments"
        if not experiments_dir.exists():
            return []
        files = sorted(path for path in experiments_dir.rglob("*.yaml") if path.is_file())
        if kind is not None:
            prefix = kind.lower()
            files = [
                path
                for path in files
                if path.relative_to(experiments_dir).parts[0].lower() == prefix
            ]
        return [path.relative_to(experiments_dir).as_posix() for path in files]

root = Path(config_root) instance-attribute

__init__(config_root: Optional[Path] = None)

Initialize config loader.

Parameters:

Name Type Description Default
config_root Optional[Path]

Root directory for config files. Defaults to ./config

None
Source code in config/config_loader.py
def __init__(self, config_root: Optional[Path] = None):
    """
    Initialize config loader.

    Args:
        config_root: Root directory for config files. Defaults to ./config
    """
    if config_root is None:
        config_root = Path(__file__).parent

    self.root = Path(config_root)
    self._cache: Dict[str, Any] = {}

load_module_config(module_path: Union[str, Path], overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Load a module config and apply optional overrides.

Source code in config/config_loader.py
def load_module_config(self, module_path: Union[str, Path], overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Load a module config and apply optional overrides."""
    data = copy.deepcopy(self._load_yaml(self._resolve_config_path(module_path)) or {})
    if overrides:
        data = _deep_merge(data, overrides)
    return data

resolve_config_modules(config: Any) -> Any

Resolve default: path/to/module.yaml sections recursively.

Paths may be absolute or relative to the config root. Sibling keys in the same mapping override the loaded defaults.

Source code in config/config_loader.py
def resolve_config_modules(self, config: Any) -> Any:
    """Resolve `default: path/to/module.yaml` sections recursively.

    Paths may be absolute or relative to the config root. Sibling keys in
    the same mapping override the loaded defaults.
    """
    if isinstance(config, list):
        return [self.resolve_config_modules(item) for item in config]
    if not isinstance(config, dict):
        return config

    if "default" in config:
        default_path = config.get("default")
        if default_path is None:
            base: Dict[str, Any] = {}
        elif isinstance(default_path, (str, Path)):
            base = self.resolve_config_modules(self.load_module_config(default_path))
        else:
            raise TypeError("Config module `default` must be a path string.")
        overrides = {
            key: self.resolve_config_modules(value)
            for key, value in config.items()
            if key != "default"
        }
        return _deep_merge(base, overrides)

    return {key: self.resolve_config_modules(value) for key, value in config.items()}

load_recdistill_config(config_path: Union[str, Path]) -> RecDistillConfig

Load and validate RecDistill configuration.

Parameters:

Name Type Description Default
config_path Union[str, Path]

Path to RecDistill config YAML file

required

Returns:

Type Description
RecDistillConfig

Validated RecDistillConfig object

Raises:

Type Description
FileNotFoundError

If config file doesn't exist

ValidationError

If config doesn't match schema

Source code in config/config_loader.py
def load_recdistill_config(self, config_path: Union[str, Path]) -> RecDistillConfig:
    """
    Load and validate RecDistill configuration.

    Args:
        config_path: Path to RecDistill config YAML file

    Returns:
        Validated RecDistillConfig object

    Raises:
        FileNotFoundError: If config file doesn't exist
        ValidationError: If config doesn't match schema
    """
    data = self._load_yaml(config_path)
    if isinstance(data, dict) and "preset" in data and "config" in data:
        data = data["config"]
    data = self.resolve_config_modules(data)
    from recdistill.config_integration import normalize_recdistill_config

    data = normalize_recdistill_config(data)
    try:
        return RecDistillConfig(**data)
    except ValidationError as e:
        raise ValueError(f"Invalid RecDistill config in {config_path}:\n{e}")

load_preset(preset_path: Union[str, Path]) -> ConfigPreset

Load a wrapped experiment preset.

Presets preserve provenance metadata under preset and keep the actual training payload under config.

Source code in config/config_loader.py
def load_preset(self, preset_path: Union[str, Path]) -> ConfigPreset:
    """
    Load a wrapped experiment preset.

    Presets preserve provenance metadata under `preset` and keep the actual
    training payload under `config`.
    """
    data = self._load_yaml(preset_path)
    try:
        return ConfigPreset(**data)
    except ValidationError as e:
        raise ValueError(f"Invalid config preset in {preset_path}:\n{e}")

load_recdistill_preset(preset_path: Union[str, Path]) -> RecDistillConfig

Load a RecDistill preset and validate its inner config.

Source code in config/config_loader.py
def load_recdistill_preset(self, preset_path: Union[str, Path]) -> RecDistillConfig:
    """Load a RecDistill preset and validate its inner config."""
    from recdistill.config_integration import normalize_recdistill_config

    preset = self.load_preset(preset_path)
    if preset.preset.kind.lower() != "recdistill":
        raise ValueError(f"Preset is not a RecDistill preset: {preset_path}")
    try:
        return RecDistillConfig(**normalize_recdistill_config(preset.config))
    except ValidationError as e:
        raise ValueError(f"Invalid RecDistill config inside preset {preset_path}:\n{e}")

load_dataset_config(dataset_name: str) -> DataConfig

Load dataset configuration.

Parameters:

Name Type Description Default
dataset_name str

Name of dataset (amazon_cd, bookcrossing, citeulike)

required

Returns:

Type Description
DataConfig

Validated DataConfig object

Source code in config/config_loader.py
def load_dataset_config(self, dataset_name: str) -> DataConfig:
    """
    Load dataset configuration.

    Args:
        dataset_name: Name of dataset (amazon_cd, bookcrossing, citeulike)

    Returns:
        Validated DataConfig object
    """
    config_path = self.root / "dataset" / f"{dataset_name}.yaml"
    data = self._load_yaml(config_path)
    try:
        return DataConfig(**data)
    except ValidationError as e:
        raise ValueError(f"Invalid dataset config for {dataset_name}:\n{e}")

load_model_config(model_type: str, model_name: str, framework: str | None = None) -> ModelConfig

Load model configuration.

Parameters:

Name Type Description Default
model_type str

Type of model ('teacher' or 'student')

required
model_name str

Name of model (bprmf, nmf, lgcn, etc.)

required

Returns:

Type Description
ModelConfig

Validated ModelConfig object

Source code in config/config_loader.py
def load_model_config(self, model_type: str, model_name: str, framework: str | None = None) -> ModelConfig:
    """
    Load model configuration.

    Args:
        model_type: Type of model ('teacher' or 'student')
        model_name: Name of model (bprmf, nmf, lgcn, etc.)

    Returns:
        Validated ModelConfig object
    """
    model_slug = str(model_name).strip().lower()
    framework_slug = str(framework).strip().lower() if framework else None
    if framework_slug:
        config_path = self.root / model_type / framework_slug / f"{model_slug}.yaml"
    else:
        config_path = self.root / model_type / f"{model_slug}.yaml"
        if not config_path.exists():
            default_path = self.root / model_type / "recbole" / f"{model_slug}.yaml"
            if default_path.exists():
                config_path = default_path
            else:
                matches = sorted((self.root / model_type).glob(f"*/{model_slug}.yaml"))
                if len(matches) == 1:
                    config_path = matches[0]
                elif len(matches) > 1:
                    choices = ", ".join(path.parent.name for path in matches)
                    raise ValueError(
                        f"Ambiguous model config for {model_type}/{model_name}. "
                        f"Specify framework. Available frameworks: {choices}."
                    )
    data = self._load_yaml(config_path)
    try:
        return ModelConfig(**data)
    except ValidationError as e:
        raise ValueError(f"Invalid model config for {model_type}/{model_name}:\n{e}")

compose_teacher_training(dataset_name: str, model_name: str, framework: str = 'recbole', model_type: str = 'teacher') -> Dict[str, Any]

Compose a generic teacher training configuration.

Parameters:

Name Type Description Default
dataset_name str

Dataset name

required
model_name str

Model name

required
framework str

Framework implementation to use

'recbole'
model_type str

Model family to load ('teacher' or 'student')

'teacher'

Returns:

Type Description
Dict[str, Any]

Complete teacher training configuration dictionary

Source code in config/config_loader.py
def compose_teacher_training(
    self,
    dataset_name: str,
    model_name: str,
    framework: str = "recbole",
    model_type: str = "teacher",
) -> Dict[str, Any]:
    """
    Compose a generic teacher training configuration.

    Args:
        dataset_name: Dataset name
        model_name: Model name
        framework: Framework implementation to use
        model_type: Model family to load ('teacher' or 'student')

    Returns:
        Complete teacher training configuration dictionary
    """
    self.load_dataset_config(dataset_name)
    model = self.load_model_config(model_type, model_name, framework=framework)

    config_path = self.root / "composites" / "teacher_template.yaml"
    template = copy.deepcopy(self._load_yaml(config_path))
    train = template.setdefault("train_teacher", {})

    train["dataset"] = dataset_name
    train["teacher"] = {"default": _module_path("teacher", model.framework, model.model)}
    return template

compose_student_training(dataset_name: str, model_name: str, framework: str = 'recbole', model_type: str = 'student') -> Dict[str, Any]

Compose a generic plain-student training configuration.

Source code in config/config_loader.py
def compose_student_training(
    self,
    dataset_name: str,
    model_name: str,
    framework: str = "recbole",
    model_type: str = "student",
) -> Dict[str, Any]:
    """
    Compose a generic plain-student training configuration.
    """
    self.load_dataset_config(dataset_name)
    model = self.load_model_config(model_type, model_name, framework=framework)

    config_path = self.root / "composites" / "student_template.yaml"
    template = copy.deepcopy(self._load_yaml(config_path))
    train = template.setdefault("train_student", {})

    train["dataset"] = dataset_name
    train["student"] = {"default": _module_path("student", model.framework, model.backbone)}
    return template

save_generated_experiment(*, kind: str, name: str, config: Dict[str, Any], path_parts: list[str], experiment_id: str | None = None) -> Path

Persist an on-the-fly composed config as an experiment file.

Source code in config/config_loader.py
def save_generated_experiment(
    self,
    *,
    kind: str,
    name: str,
    config: Dict[str, Any],
    path_parts: list[str],
    experiment_id: str | None = None,
) -> Path:
    """Persist an on-the-fly composed config as an experiment file."""
    del path_parts
    experiment = config.get("experiment") if isinstance(config.get("experiment"), dict) else {}
    experiment_id = normalize_experiment_id(experiment_id or experiment.get("id") or new_experiment_id())
    kind_slug = _slug(kind)
    framework, model, dataset = _experiment_identity(self.resolve_config_modules(config), kind_slug)
    run_dir = experiment_run_dir(
        kind_slug,
        experiment_id,
        framework=framework,
        model=model,
        dataset=dataset,
    )
    experiment["id"] = str(experiment_id)
    experiment.setdefault("name", _slug(name))
    experiment.setdefault("kind", kind_slug)
    root_key = _training_root_key(kind_slug)
    if root_key in config and isinstance(config[root_key], dict):
        runtime = config[root_key].setdefault("runtime", {})
        if isinstance(runtime, dict) and not runtime.get("output_path"):
            runtime["output_path"] = str(
                run_dir
                / "artifacts"
                / experiment_artifact_filename(
                    kind=kind_slug,
                    experiment_id=experiment_id,
                    framework=framework,
                    model=model,
                    dataset=dataset,
                )
            )
    config = {
        "experiment": experiment,
        **{key: value for key, value in config.items() if key != "experiment"},
    }
    experiment_path = run_dir / "config" / f"{_slug(name)}_{experiment_id}.yaml"
    experiment_path.parent.mkdir(parents=True, exist_ok=True)
    with experiment_path.open("w", encoding="utf-8") as fp:
        yaml.safe_dump(config, fp, sort_keys=False, allow_unicode=False)
    return experiment_path

save_generated_preset(*, kind: str, family: str, name: str, config: Dict[str, Any], path_parts: list[str]) -> Path

Backward-compatible alias for saving generated experiment configs.

Source code in config/config_loader.py
def save_generated_preset(
    self,
    *,
    kind: str,
    family: str,
    name: str,
    config: Dict[str, Any],
    path_parts: list[str],
) -> Path:
    """Backward-compatible alias for saving generated experiment configs."""
    del family
    return self.save_generated_experiment(
        kind=kind,
        name=name,
        config=config,
        path_parts=path_parts,
    )

compose_recdistill_experiment(dataset_name: str, teacher_model: str, distiller_strategy: str, student_backbone: str = None, teacher_framework: str = 'recbole', student_framework: str = 'recbole') -> Dict[str, Any]

Compose a complete RecDistill experiment configuration.

Parameters:

Name Type Description Default
dataset_name str

Dataset name

required
teacher_model str

Teacher model (bprmf, nmf, lgcn)

required
distiller_strategy str

Distillation strategy (de, htd, ftd, unkd)

required
student_backbone str

Student backbone (default: same as teacher)

None
teacher_framework str

Framework used to resolve the teacher model config

'recbole'
student_framework str

Framework used to resolve the student backbone config

'recbole'

Returns:

Type Description
Dict[str, Any]

Complete distillation configuration dictionary

Source code in config/config_loader.py
def compose_recdistill_experiment(
    self,
    dataset_name: str,
    teacher_model: str,
    distiller_strategy: str,
    student_backbone: str = None,
    teacher_framework: str = "recbole",
    student_framework: str = "recbole",
) -> Dict[str, Any]:
    """
    Compose a complete RecDistill experiment configuration.

    Args:
        dataset_name: Dataset name
        teacher_model: Teacher model (bprmf, nmf, lgcn)
        distiller_strategy: Distillation strategy (de, htd, ftd, unkd)
        student_backbone: Student backbone (default: same as teacher)
        teacher_framework: Framework used to resolve the teacher model config
        student_framework: Framework used to resolve the student backbone config

    Returns:
        Complete distillation configuration dictionary
    """
    if student_backbone is None:
        student_backbone = teacher_model

    dataset = self.load_dataset_config(dataset_name)
    teacher_cfg = self.load_model_config("teacher", teacher_model, framework=teacher_framework)
    student_cfg = self.load_model_config("student", student_backbone, framework=student_framework)
    from recdistill.model_validation import validate_distillation_request

    validate_distillation_request(
        teacher_framework=teacher_cfg.framework,
        teacher_model=teacher_cfg.model,
        student_framework=student_cfg.framework,
        student_backbone=student_cfg.backbone,
        distiller=distiller_strategy,
    )

    config_path = self.root / "composites" / "recdistill_template.yaml"
    template = copy.deepcopy(self._load_yaml(config_path))

    # Merge configs
    train = template["distill_student"]
    train["dataset"] = dataset_name
    train["teacher"] = {"default": _module_path("teacher", teacher_cfg.framework, teacher_cfg.model)}
    train["student"] = {"default": _module_path("student", student_cfg.framework, student_cfg.backbone)}
    train["distillation"] = {
        "default": f"distillation/{_slug(distiller_strategy)}.yaml",
        "strategy": str(distiller_strategy).replace("-", "_").upper(),
    }

    return template

clear_cache()

Clear configuration cache.

Source code in config/config_loader.py
def clear_cache(self):
    """Clear configuration cache."""
    self._cache.clear()

list_datasets() -> list[str]

List available datasets.

Source code in config/config_loader.py
def list_datasets(self) -> list[str]:
    """List available datasets."""
    datasets_dir = self.root / "dataset"
    return [f.stem for f in datasets_dir.glob("*.yaml")]

list_models(model_type: str = None) -> Dict[str, list[str]]

List available models by type.

Source code in config/config_loader.py
def list_models(self, model_type: str = None) -> Dict[str, list[str]]:
    """List available models by type."""
    if model_type:
        type_dir = self.root / model_type
        return {
            model_type: [
                path.relative_to(type_dir).with_suffix("").as_posix()
                for path in sorted(type_dir.rglob("*.yaml"))
            ]
        }
    else:
        result = {}
        for model_type_name in ("teacher", "student"):
            type_dir = self.root / model_type_name
            if type_dir.exists():
                result[model_type_name] = [
                    path.relative_to(type_dir).with_suffix("").as_posix()
                    for path in sorted(type_dir.rglob("*.yaml"))
                ]
        return result

list_distillers() -> list[str]

List available distiller strategies.

Source code in config/config_loader.py
def list_distillers(self) -> list[str]:
    """List available distiller strategies."""
    distillers_dir = self.root / "distillation"
    return [f.stem for f in distillers_dir.glob("*.yaml")]

list_presets(kind: Optional[str] = None) -> list[str]

List experiment files relative to config/experiments.

Source code in config/config_loader.py
def list_presets(self, kind: Optional[str] = None) -> list[str]:
    """List experiment files relative to config/experiments."""
    return self.list_experiments(kind)

list_experiments(kind: Optional[str] = None) -> list[str]

List experiment files relative to config/experiments.

Source code in config/config_loader.py
def list_experiments(self, kind: Optional[str] = None) -> list[str]:
    """List experiment files relative to config/experiments."""
    experiments_dir = self.root / "experiments"
    if not experiments_dir.exists():
        return []
    files = sorted(path for path in experiments_dir.rglob("*.yaml") if path.is_file())
    if kind is not None:
        prefix = kind.lower()
        files = [
            path
            for path in files
            if path.relative_to(experiments_dir).parts[0].lower() == prefix
        ]
    return [path.relative_to(experiments_dir).as_posix() for path in files]

get_config_loader(config_root: Optional[Path] = None) -> ConfigLoader

Get or create global config loader instance.

Source code in config/config_loader.py
def get_config_loader(config_root: Optional[Path] = None) -> ConfigLoader:
    """Get or create global config loader instance."""
    global _global_loader
    if _global_loader is None:
        _global_loader = ConfigLoader(config_root)
    return _global_loader

reset_config_loader()

Reset global config loader instance.

Source code in config/config_loader.py
def reset_config_loader():
    """Reset global config loader instance."""
    global _global_loader
    _global_loader = None

Schemas

Pydantic schemas for configuration validation. Centralizes all configuration structure definitions for Elliot and RecDistill.

DataConfig

Bases: BaseModel

Dataset configuration.

Source code in config/schemas.py
class DataConfig(BaseModel):
    """Dataset configuration."""
    model_config = ConfigDict(extra="allow")

    name: str = Field(..., description="Dataset name (amazon_cd, bookcrossing, citeulike)")
    train_path: str = Field(..., description="Path to training data")
    test_path: str = Field(..., description="Path to test data")
    validation_path: Optional[str] = Field(default=None, description="Path to validation data")
    side_information: Optional[Dict[str, Any]] = Field(default=None)

    @field_validator("train_path", "test_path", "validation_path")
    @classmethod
    def validate_existing_dataset_path(cls, value: Optional[str]) -> Optional[str]:
        if value is None:
            return value
        if ".." in Path(value).parts:
            return value
        path = Path(value)
        if not path.is_absolute():
            path = Path.cwd() / path
        if not path.exists():
            raise ValueError(f"Dataset path does not exist: {value}")
        return value

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

name: str = Field(..., description='Dataset name (amazon_cd, bookcrossing, citeulike)') class-attribute instance-attribute

train_path: str = Field(..., description='Path to training data') class-attribute instance-attribute

test_path: str = Field(..., description='Path to test data') class-attribute instance-attribute

validation_path: Optional[str] = Field(default=None, description='Path to validation data') class-attribute instance-attribute

side_information: Optional[Dict[str, Any]] = Field(default=None) class-attribute instance-attribute

validate_existing_dataset_path(value: Optional[str]) -> Optional[str] classmethod

Source code in config/schemas.py
@field_validator("train_path", "test_path", "validation_path")
@classmethod
def validate_existing_dataset_path(cls, value: Optional[str]) -> Optional[str]:
    if value is None:
        return value
    if ".." in Path(value).parts:
        return value
    path = Path(value)
    if not path.is_absolute():
        path = Path.cwd() / path
    if not path.exists():
        raise ValueError(f"Dataset path does not exist: {value}")
    return value

ModelConfig

Bases: BaseModel

Model configuration (common for Teacher/Student).

Source code in config/schemas.py
class ModelConfig(BaseModel):
    """Model configuration (common for Teacher/Student)."""
    model_config = ConfigDict(extra="allow")

    framework: str = Field(default="recbole", description="Framework implementation (recbole, elliot, lenskit)")
    backbone: Optional[str] = Field(default=None, description="Student backbone (BPRMF, NMF, NGCF, LGCN, etc.)")
    model: Optional[str] = Field(default=None, description="Teacher model (BPRMF, NMF, NGCF, LGCN, etc.)")
    embedding_dim: int = Field(..., description="Embedding dimension")
    learning_rate: float = Field(default=0.001)
    l2_reg: float = Field(default=0.0001)
    dropout: float = Field(default=0.0)
    factors: Optional[int] = Field(default=None)  # For legacy compatibility

    @model_validator(mode="after")
    def normalize_model_name(self) -> "ModelConfig":
        name = self.backbone or self.model
        if name is None:
            raise ValueError("Model config must define either backbone or model.")
        normalized = canonical_model_name(name)
        self.backbone = normalized
        self.model = normalized
        self.framework = str(self.framework).strip().lower()
        return self

    @field_validator("backbone", "model")
    @classmethod
    def normalize_optional_model_name(cls, value: Optional[str]) -> Optional[str]:
        if value is None:
            return None
        return canonical_model_name(value)

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

framework: str = Field(default='recbole', description='Framework implementation (recbole, elliot, lenskit)') class-attribute instance-attribute

backbone: Optional[str] = Field(default=None, description='Student backbone (BPRMF, NMF, NGCF, LGCN, etc.)') class-attribute instance-attribute

model: Optional[str] = Field(default=None, description='Teacher model (BPRMF, NMF, NGCF, LGCN, etc.)') class-attribute instance-attribute

embedding_dim: int = Field(..., description='Embedding dimension') class-attribute instance-attribute

learning_rate: float = Field(default=0.001) class-attribute instance-attribute

l2_reg: float = Field(default=0.0001) class-attribute instance-attribute

dropout: float = Field(default=0.0) class-attribute instance-attribute

factors: Optional[int] = Field(default=None) class-attribute instance-attribute

normalize_model_name() -> ModelConfig

Source code in config/schemas.py
@model_validator(mode="after")
def normalize_model_name(self) -> "ModelConfig":
    name = self.backbone or self.model
    if name is None:
        raise ValueError("Model config must define either backbone or model.")
    normalized = canonical_model_name(name)
    self.backbone = normalized
    self.model = normalized
    self.framework = str(self.framework).strip().lower()
    return self

normalize_optional_model_name(value: Optional[str]) -> Optional[str] classmethod

Source code in config/schemas.py
@field_validator("backbone", "model")
@classmethod
def normalize_optional_model_name(cls, value: Optional[str]) -> Optional[str]:
    if value is None:
        return None
    return canonical_model_name(value)

OptimizationConfig

Bases: BaseModel

Training optimization parameters.

Source code in config/schemas.py
class OptimizationConfig(BaseModel):
    """Training optimization parameters."""
    epochs: int = Field(default=100)
    batch_size: int = Field(default=512)
    learning_rate: float = Field(default=0.001)
    l2_reg: float = Field(default=0.0001)
    early_stopping: "EarlyStoppingConfig" = Field(default_factory=lambda: EarlyStoppingConfig())
    bayesian: Dict[str, Any] = Field(default_factory=lambda: {"enabled": False})

epochs: int = Field(default=100) class-attribute instance-attribute

batch_size: int = Field(default=512) class-attribute instance-attribute

learning_rate: float = Field(default=0.001) class-attribute instance-attribute

l2_reg: float = Field(default=0.0001) class-attribute instance-attribute

early_stopping: EarlyStoppingConfig = Field(default_factory=(lambda: EarlyStoppingConfig())) class-attribute instance-attribute

bayesian: Dict[str, Any] = Field(default_factory=(lambda: {'enabled': False})) class-attribute instance-attribute

EvaluationConfig

Bases: BaseModel

Evaluation configuration.

Source code in config/schemas.py
class EvaluationConfig(BaseModel):
    """Evaluation configuration."""
    cutoffs: List[int] = Field(default=[10, 20, 50])
    k: int = Field(default=20)
    every: int = Field(default=5)
    batch_size: int = Field(default=1024)
    simple_metrics: List[str] = Field(default=["nDCGRendle2020", "Recall", "Precision", "HR"])
    val_only: bool = Field(default=True)
    selection_split: str = Field(default="val")
    selection_metric: str = Field(default="ndcg")
    enabled: bool = Field(default=True)
    assert_no_train_leak: bool = Field(default=True)
    relevance_threshold: int = Field(default=0)

    @field_validator("selection_split")
    @classmethod
    def validate_selection_split(cls, value: str) -> str:
        normalized = str(value).lower()
        if normalized not in {"val", "test"}:
            raise ValueError("selection_split must be either 'val' or 'test'.")
        return normalized

    @field_validator("selection_metric")
    @classmethod
    def validate_selection_metric(cls, value: str) -> str:
        normalized = str(value).lower()
        if normalized not in {"precision", "recall", "ndcg", "hr"}:
            raise ValueError("selection_metric must be one of precision, recall, ndcg, hr.")
        return normalized

    @model_validator(mode="after")
    def validate_selection_policy(self) -> "EvaluationConfig":
        if self.val_only and self.selection_split == "test":
            raise ValueError("evaluation.val_only=true is incompatible with selection_split='test'.")
        return self

cutoffs: List[int] = Field(default=[10, 20, 50]) class-attribute instance-attribute

k: int = Field(default=20) class-attribute instance-attribute

every: int = Field(default=5) class-attribute instance-attribute

batch_size: int = Field(default=1024) class-attribute instance-attribute

simple_metrics: List[str] = Field(default=['nDCGRendle2020', 'Recall', 'Precision', 'HR']) class-attribute instance-attribute

val_only: bool = Field(default=True) class-attribute instance-attribute

selection_split: str = Field(default='val') class-attribute instance-attribute

selection_metric: str = Field(default='ndcg') class-attribute instance-attribute

enabled: bool = Field(default=True) class-attribute instance-attribute

assert_no_train_leak: bool = Field(default=True) class-attribute instance-attribute

relevance_threshold: int = Field(default=0) class-attribute instance-attribute

validate_selection_split(value: str) -> str classmethod

Source code in config/schemas.py
@field_validator("selection_split")
@classmethod
def validate_selection_split(cls, value: str) -> str:
    normalized = str(value).lower()
    if normalized not in {"val", "test"}:
        raise ValueError("selection_split must be either 'val' or 'test'.")
    return normalized

validate_selection_metric(value: str) -> str classmethod

Source code in config/schemas.py
@field_validator("selection_metric")
@classmethod
def validate_selection_metric(cls, value: str) -> str:
    normalized = str(value).lower()
    if normalized not in {"precision", "recall", "ndcg", "hr"}:
        raise ValueError("selection_metric must be one of precision, recall, ndcg, hr.")
    return normalized

validate_selection_policy() -> EvaluationConfig

Source code in config/schemas.py
@model_validator(mode="after")
def validate_selection_policy(self) -> "EvaluationConfig":
    if self.val_only and self.selection_split == "test":
        raise ValueError("evaluation.val_only=true is incompatible with selection_split='test'.")
    return self

EarlyStoppingConfig

Bases: BaseModel

Early stopping configuration.

Source code in config/schemas.py
class EarlyStoppingConfig(BaseModel):
    """Early stopping configuration."""
    enabled: bool = Field(default=True)
    mode: str = Field(default="val_metric")  # val_metric or check
    metric: str = Field(default="ndcg")
    patience: int = Field(default=10)
    min_delta: float = Field(default=0.0)
    warmup: int = Field(default=0)
    restore_best: bool = Field(default=False)

enabled: bool = Field(default=True) class-attribute instance-attribute

mode: str = Field(default='val_metric') class-attribute instance-attribute

metric: str = Field(default='ndcg') class-attribute instance-attribute

patience: int = Field(default=10) class-attribute instance-attribute

min_delta: float = Field(default=0.0) class-attribute instance-attribute

warmup: int = Field(default=0) class-attribute instance-attribute

restore_best: bool = Field(default=False) class-attribute instance-attribute

RuntimeConfig

Bases: BaseModel

Runtime configuration.

Source code in config/schemas.py
class RuntimeConfig(BaseModel):
    """Runtime configuration."""
    seed: int = Field(default=42)
    device: Optional[str] = Field(default=None)
    num_workers: int = Field(default=4)
    output_path: Optional[str] = Field(default=None)
    output_strategy: str = Field(default="fixed")
    save_every: int = Field(default=0)
    wandb: Dict[str, Any] = Field(default_factory=lambda: {"enabled": False})
    extra_args: List[str] = Field(default_factory=list)

seed: int = Field(default=42) class-attribute instance-attribute

device: Optional[str] = Field(default=None) class-attribute instance-attribute

num_workers: int = Field(default=4) class-attribute instance-attribute

output_path: Optional[str] = Field(default=None) class-attribute instance-attribute

output_strategy: str = Field(default='fixed') class-attribute instance-attribute

save_every: int = Field(default=0) class-attribute instance-attribute

wandb: Dict[str, Any] = Field(default_factory=(lambda: {'enabled': False})) class-attribute instance-attribute

extra_args: List[str] = Field(default_factory=list) class-attribute instance-attribute

DistillerConfig

Bases: BaseModel

Knowledge distillation configuration.

Source code in config/schemas.py
class DistillerConfig(BaseModel):
    """Knowledge distillation configuration."""
    model_config = ConfigDict(extra="allow")

    strategy: str = Field(..., description="Distillation strategy (DE, HTD, FTD, UnKD, etc.)")
    temperature: float = Field(default=3.0)
    lambda_kl: float = Field(default=0.5)
    lambda_de: float = Field(default=0.1)
    # Distillation Experts specific
    num_experts: int = Field(default=20)
    # UnKD specific
    unkd: Optional[Dict[str, Any]] = Field(default=None)
    # Add more distiller-specific configs as needed

    @field_validator("strategy")
    @classmethod
    def normalize_strategy(cls, value: str) -> str:
        methods = parse_distiller_methods(value)
        if not methods:
            raise ValueError("distillation.strategy must contain at least one distiller.")
        if "HTD" in methods and "FTD" in methods:
            raise ValueError("HTD and FTD cannot be active at the same time.")
        return "_".join(methods)

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

strategy: str = Field(..., description='Distillation strategy (DE, HTD, FTD, UnKD, etc.)') class-attribute instance-attribute

temperature: float = Field(default=3.0) class-attribute instance-attribute

lambda_kl: float = Field(default=0.5) class-attribute instance-attribute

lambda_de: float = Field(default=0.1) class-attribute instance-attribute

num_experts: int = Field(default=20) class-attribute instance-attribute

unkd: Optional[Dict[str, Any]] = Field(default=None) class-attribute instance-attribute

normalize_strategy(value: str) -> str classmethod

Source code in config/schemas.py
@field_validator("strategy")
@classmethod
def normalize_strategy(cls, value: str) -> str:
    methods = parse_distiller_methods(value)
    if not methods:
        raise ValueError("distillation.strategy must contain at least one distiller.")
    if "HTD" in methods and "FTD" in methods:
        raise ValueError("HTD and FTD cannot be active at the same time.")
    return "_".join(methods)

TeacherConfig

Bases: BaseModel

Teacher model configuration.

Source code in config/schemas.py
class TeacherConfig(BaseModel):
    """Teacher model configuration."""
    model_config = ConfigDict(extra="allow")

    model: str = Field(...)
    embedding_dim: Optional[int] = Field(default=None)
    path: Optional[str] = Field(default=None)
    framework: str = Field(default="auto")
    format: str = Field(default="auto")

    @field_validator("model")
    @classmethod
    def normalize_model(cls, value: str) -> str:
        try:
            return canonical_model_name(value)
        except ValueError:
            return str(value)

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

model: str = Field(...) class-attribute instance-attribute

embedding_dim: Optional[int] = Field(default=None) class-attribute instance-attribute

path: Optional[str] = Field(default=None) class-attribute instance-attribute

framework: str = Field(default='auto') class-attribute instance-attribute

format: str = Field(default='auto') class-attribute instance-attribute

normalize_model(value: str) -> str classmethod

Source code in config/schemas.py
@field_validator("model")
@classmethod
def normalize_model(cls, value: str) -> str:
    try:
        return canonical_model_name(value)
    except ValueError:
        return str(value)

StudentConfig

Bases: BaseModel

Student model configuration.

Source code in config/schemas.py
class StudentConfig(BaseModel):
    """Student model configuration."""
    model_config = ConfigDict(extra="allow")

    framework: str = Field(default="recbole")
    backbone: str = Field(...)
    embedding_dim: int = Field(...)
    lambda_de: float = Field(default=0.1)
    num_experts: int = Field(default=20)
    temperature: float = Field(default=0.1)

    @field_validator("backbone")
    @classmethod
    def normalize_backbone(cls, value: str) -> str:
        return canonical_model_name(value)

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

framework: str = Field(default='recbole') class-attribute instance-attribute

backbone: str = Field(...) class-attribute instance-attribute

embedding_dim: int = Field(...) class-attribute instance-attribute

lambda_de: float = Field(default=0.1) class-attribute instance-attribute

num_experts: int = Field(default=20) class-attribute instance-attribute

temperature: float = Field(default=0.1) class-attribute instance-attribute

normalize_backbone(value: str) -> str classmethod

Source code in config/schemas.py
@field_validator("backbone")
@classmethod
def normalize_backbone(cls, value: str) -> str:
    return canonical_model_name(value)

DistillStudentTrainingConfig

Bases: BaseModel

RecDistill student training configuration.

Source code in config/schemas.py
class DistillStudentTrainingConfig(BaseModel):
    """RecDistill student training configuration."""
    model_config = ConfigDict(extra="allow")

    dataset: str = Field(...)
    teacher: TeacherConfig = Field(...)
    student: StudentConfig = Field(...)
    optimization: OptimizationConfig = Field(...)
    distillation: DistillerConfig = Field(...)
    runtime: RuntimeConfig = Field(...)
    evaluation: EvaluationConfig = Field(...)

    @model_validator(mode="after")
    def validate_cross_config_contracts(self) -> "DistillStudentTrainingConfig":
        if self.teacher.embedding_dim is not None and self.teacher.embedding_dim <= self.student.embedding_dim:
            raise ValueError("teacher.embedding_dim must be greater than student.embedding_dim.")

        strategy_methods = set(parse_distiller_methods(self.distillation.strategy))

        active_methods = strategy_methods
        lambda_by_method = {
            "DE": float(getattr(self.distillation, "lambda_de", 0.0)),
            "RRD": float(getattr(self.distillation, "lambda_rrd", 0.0)),
            "UNKD": float(getattr(self.distillation, "lambda_unkd", 0.0)),
        }
        topology = getattr(self.distillation, "topology", {}) or {}
        lambda_td = float(topology.get("lambda_td", getattr(self.distillation, "lambda_td", 0.0)))
        if lambda_td > 0.0 and {"HTD", "FTD"}.isdisjoint(active_methods):
            raise ValueError("lambda_td > 0 requires HTD or FTD in distillation.strategy.")
        for method, value in lambda_by_method.items():
            if value > 0.0 and method not in active_methods:
                raise ValueError(f"lambda for {method} is > 0 but {method} is not active.")
        return self

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

dataset: str = Field(...) class-attribute instance-attribute

teacher: TeacherConfig = Field(...) class-attribute instance-attribute

student: StudentConfig = Field(...) class-attribute instance-attribute

optimization: OptimizationConfig = Field(...) class-attribute instance-attribute

distillation: DistillerConfig = Field(...) class-attribute instance-attribute

runtime: RuntimeConfig = Field(...) class-attribute instance-attribute

evaluation: EvaluationConfig = Field(...) class-attribute instance-attribute

validate_cross_config_contracts() -> DistillStudentTrainingConfig

Source code in config/schemas.py
@model_validator(mode="after")
def validate_cross_config_contracts(self) -> "DistillStudentTrainingConfig":
    if self.teacher.embedding_dim is not None and self.teacher.embedding_dim <= self.student.embedding_dim:
        raise ValueError("teacher.embedding_dim must be greater than student.embedding_dim.")

    strategy_methods = set(parse_distiller_methods(self.distillation.strategy))

    active_methods = strategy_methods
    lambda_by_method = {
        "DE": float(getattr(self.distillation, "lambda_de", 0.0)),
        "RRD": float(getattr(self.distillation, "lambda_rrd", 0.0)),
        "UNKD": float(getattr(self.distillation, "lambda_unkd", 0.0)),
    }
    topology = getattr(self.distillation, "topology", {}) or {}
    lambda_td = float(topology.get("lambda_td", getattr(self.distillation, "lambda_td", 0.0)))
    if lambda_td > 0.0 and {"HTD", "FTD"}.isdisjoint(active_methods):
        raise ValueError("lambda_td > 0 requires HTD or FTD in distillation.strategy.")
    for method, value in lambda_by_method.items():
        if value > 0.0 and method not in active_methods:
            raise ValueError(f"lambda for {method} is > 0 but {method} is not active.")
    return self

RecDistillConfig

Bases: BaseModel

Root RecDistill configuration.

Source code in config/schemas.py
class RecDistillConfig(BaseModel):
    """Root RecDistill configuration."""
    distill_student: DistillStudentTrainingConfig = Field(...)

distill_student: DistillStudentTrainingConfig = Field(...) class-attribute instance-attribute

PresetMetadata

Bases: BaseModel

Metadata for experiment presets.

Source code in config/schemas.py
class PresetMetadata(BaseModel):
    """Metadata for experiment presets."""
    model_config = ConfigDict(extra="allow")

    schema_version: int = Field(default=1)
    kind: str = Field(..., description="Preset kind, for example recdistill, teacher, student, or raw")
    family: str = Field(..., description="Logical preset family")

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

schema_version: int = Field(default=1) class-attribute instance-attribute

kind: str = Field(..., description='Preset kind, for example recdistill, teacher, student, or raw') class-attribute instance-attribute

family: str = Field(..., description='Logical preset family') class-attribute instance-attribute

ConfigPreset

Bases: BaseModel

Wrapped preset preserving config plus metadata.

Source code in config/schemas.py
class ConfigPreset(BaseModel):
    """Wrapped preset preserving config plus metadata."""
    model_config = ConfigDict(extra="allow")

    preset: PresetMetadata = Field(...)
    config: Dict[str, Any] = Field(...)

model_config = ConfigDict(extra='allow') class-attribute instance-attribute

preset: PresetMetadata = Field(...) class-attribute instance-attribute

config: Dict[str, Any] = Field(...) class-attribute instance-attribute