Hugging Face's logo

Spaces:
yashb18
/
deepreads
Running

App Files Community
deepreads / src /config.py
yash-b18
feat: fix Two-Tower collapse, redesign frontend, filter untitled items
2647d97
Raw History Blame Contribute Delete
9.49 kB
"""
Central configuration for the recommendation system project.
All paths, hyperparameters, and dataset settings are defined here.
Scripts should import from this module rather than hardcode values.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional
# ---------------------------------------------------------------------------
# Root paths
# ---------------------------------------------------------------------------
ROOT_DIR = Path(__file__).resolve().parent.parent
DATA_DIR = ROOT_DIR / "data"
RAW_DIR = DATA_DIR / "raw"
PROCESSED_DIR = DATA_DIR / "processed"
OUTPUTS_DIR = DATA_DIR / "outputs"
FIGURES_DIR = OUTPUTS_DIR / "figures"
MODELS_DIR = ROOT_DIR / "models"
NOTEBOOKS_DIR = ROOT_DIR / "notebooks"
# ---------------------------------------------------------------------------
# Dataset configuration
# ---------------------------------------------------------------------------
@dataclass
class DataConfig:
"""Dataset download and processing settings."""
# HuggingFace dataset identifier
hf_dataset_name: str = "McAuley-Lab/Amazon-Reviews-2023"
# Category to use. Smaller categories are faster for development.
# Options: "Video_Games", "Books", "Electronics", "Clothing_Shoes_and_Jewelry"
category: str = "Books"
# Minimum number of reviews a user must have to be included
min_user_reviews: int = 5
# Minimum number of reviews an item must have to be included
min_item_reviews: int = 5
# Threshold above which a rating is considered a positive interaction
positive_rating_threshold: float = 4.0
# Number of negative samples per positive interaction
negative_samples_per_positive: int = 4
# Fraction of data for validation and test splits
val_fraction: float = 0.1
test_fraction: float = 0.1
# Random seed for reproducibility
random_seed: int = 42
# Debug mode: use a tiny subset for fast iteration
debug: bool = False
debug_n_users: int = 500
# Paths (resolved at runtime)
raw_reviews_path: Path = field(default_factory=lambda: RAW_DIR / "reviews.jsonl")
raw_metadata_path: Path = field(default_factory=lambda: RAW_DIR / "metadata.jsonl")
interactions_path: Path = field(default_factory=lambda: PROCESSED_DIR / "interactions.parquet")
metadata_path: Path = field(default_factory=lambda: PROCESSED_DIR / "metadata.parquet")
train_path: Path = field(default_factory=lambda: PROCESSED_DIR / "train.parquet")
val_path: Path = field(default_factory=lambda: PROCESSED_DIR / "val.parquet")
test_path: Path = field(default_factory=lambda: PROCESSED_DIR / "test.parquet")
user_features_path: Path = field(default_factory=lambda: PROCESSED_DIR / "user_features.parquet")
item_features_path: Path = field(default_factory=lambda: PROCESSED_DIR / "item_features.parquet")
train_features_path: Path = field(default_factory=lambda: PROCESSED_DIR / "train_features.parquet")
val_features_path: Path = field(default_factory=lambda: PROCESSED_DIR / "val_features.parquet")
test_features_path: Path = field(default_factory=lambda: PROCESSED_DIR / "test_features.parquet")
user_encoder_path: Path = field(default_factory=lambda: PROCESSED_DIR / "user_encoder.pkl")
item_encoder_path: Path = field(default_factory=lambda: PROCESSED_DIR / "item_encoder.pkl")
# ---------------------------------------------------------------------------
# Model configurations
# ---------------------------------------------------------------------------
@dataclass
class NaiveConfig:
"""Configuration for the naive baseline models."""
top_k: int = 10
model_path: Path = field(default_factory=lambda: MODELS_DIR / "naive_baseline.pkl")
@dataclass
class ClassicalConfig:
"""Configuration for the LightGBM classical model."""
# LightGBM hyperparameters
n_estimators: int = 500
learning_rate: float = 0.05
num_leaves: int = 63
max_depth: int = -1
min_child_samples: int = 20
subsample: float = 0.8
colsample_bytree: float = 0.8
reg_alpha: float = 0.1
reg_lambda: float = 0.1
n_jobs: int = -1
random_state: int = 42
# Training
early_stopping_rounds: int = 50
verbose_eval: int = 100
# Top-K candidates to score per user (from naive retrieval)
candidate_pool_size: int = 500
# Model artifact
model_path: Path = field(default_factory=lambda: MODELS_DIR / "lgbm_model.pkl")
feature_names_path: Path = field(default_factory=lambda: MODELS_DIR / "lgbm_feature_names.pkl")
shap_values_path: Path = field(default_factory=lambda: OUTPUTS_DIR / "shap_values.npy")
@dataclass
class DeepConfig:
"""Configuration for the Two-Tower neural recommender."""
# Architecture
# hidden_dims=[] → single linear projection (no hidden layers), avoids MLP collapse
# on sparse book data. Deep MLPs converge to a constant function with sparse gradients.
embedding_dim: int = 64
hidden_dims: List[int] = field(default_factory=lambda: [])
dropout: float = 0.1
use_metadata: bool = True
use_text: bool = True
# Text encoding
text_embedding_dim: int = 64
max_title_tokens: int = 32
# Training
batch_size: int = 1024
epochs: int = 30
learning_rate: float = 1e-3
weight_decay: float = 1e-5
num_workers: int = 4
patience: int = 8
random_seed: int = 42
# Inference
top_k: int = 10
candidate_pool_size: int = 200
# Model artifacts
model_path: Path = field(default_factory=lambda: MODELS_DIR / "two_tower.pt")
item_embeddings_path: Path = field(default_factory=lambda: MODELS_DIR / "item_embeddings.npy")
item_index_path: Path = field(default_factory=lambda: MODELS_DIR / "item_index.pkl")
# ---------------------------------------------------------------------------
# Evaluation configuration
# ---------------------------------------------------------------------------
@dataclass
class EvalConfig:
"""Evaluation settings."""
k_values: List[int] = field(default_factory=lambda: [5, 10, 20])
primary_k: int = 10
metrics_output_path: Path = field(default_factory=lambda: OUTPUTS_DIR / "metrics.csv")
metrics_table_path: Path = field(default_factory=lambda: OUTPUTS_DIR / "metrics_table.md")
# ---------------------------------------------------------------------------
# Experiment configuration
# ---------------------------------------------------------------------------
@dataclass
class ExperimentConfig:
"""Feature ablation experiment settings."""
experiment_name: str = "feature_ablation"
# Feature sets to compare in ablation
ablation_sets: List[str] = field(default_factory=lambda: [
"id_only",
"id_metadata",
"id_metadata_text",
"id_metadata_text_history",
])
output_dir: Path = field(default_factory=lambda: OUTPUTS_DIR / "experiment")
# ---------------------------------------------------------------------------
# API configuration
# ---------------------------------------------------------------------------
@dataclass
class APIConfig:
"""FastAPI backend configuration."""
host: str = "0.0.0.0"
port: int = 8000
top_k: int = 10
default_model: str = "deep" # "naive" | "classical" | "deep"
cors_origins: List[str] = field(default_factory=lambda: [
"http://localhost:3000",
"http://localhost:3001",
"http://localhost:3002",
"http://localhost:3003",
"https://*.vercel.app",
])
# ---------------------------------------------------------------------------
# Master config
# ---------------------------------------------------------------------------
@dataclass
class Config:
"""Aggregates all sub-configs. Import this in scripts."""
data: DataConfig = field(default_factory=DataConfig)
naive: NaiveConfig = field(default_factory=NaiveConfig)
classical: ClassicalConfig = field(default_factory=ClassicalConfig)
deep: DeepConfig = field(default_factory=DeepConfig)
eval: EvalConfig = field(default_factory=EvalConfig)
experiment: ExperimentConfig = field(default_factory=ExperimentConfig)
api: APIConfig = field(default_factory=APIConfig)
def get_config(debug: bool = False, category: Optional[str] = None) -> Config:
"""
Return a Config instance, optionally overriding debug mode or category.
Args:
debug: If True, use a small subset of data for fast iteration.
category: Override the Amazon product category.
Returns:
Fully constructed Config.
"""
cfg = Config()
if debug:
cfg.data.debug = True
cfg.data.debug_n_users = 100_000 # high cap — download size already controlled
cfg.deep.epochs = 10
cfg.deep.batch_size = 1024
cfg.classical.n_estimators = 200
if category:
cfg.data.category = category
return cfg
# ---------------------------------------------------------------------------
# Ensure output directories exist when this module is imported
# ---------------------------------------------------------------------------
def ensure_dirs(cfg: Optional[Config] = None) -> None:
"""Create all necessary directories if they do not exist."""
dirs = [RAW_DIR, PROCESSED_DIR, OUTPUTS_DIR, FIGURES_DIR, MODELS_DIR]
if cfg:
dirs.append(cfg.experiment.output_dir)
for d in dirs:
d.mkdir(parents=True, exist_ok=True)