Quickstart TensorFlow¶

Dans cet exemple, nous allons apprendre à entraîner un réseau neuronal convolutionnel sur CIFAR-10 en utilisant le framework Flower et TensorFlow. Tout d’abord, il est recommandé de créer un environnement virtuel et d’exécuter tout dans un virtualenv.

Utilisons flwr new pour créer un projet complet Flower+TensorFlow. Il générera tous les fichiers nécessaires pour exécuter une fédération de 10 nœuds en utilisant la stratégie FedAvg. Par défaut, l’application générée utilise un profil de simulation local qui soumet à un SuperLink géré local, qui exécute ensuite le run avec le runtime d’exécution Simulation Flower. Le jeu de données sera partitionné en utilisant les partitions de Flower Dataset’s IidPartitioner.

Maintenant que nous avons une idée approximative de ce que cet exemple est sur, allons-y. Tout d’abord, installez Flower dans votre nouvel environnement :

# In a new Python environment
$ pip install flwr[simulation]

Ensuite, exécutez la commande suivante :

$ flwr new @flwrlabs/quickstart-tensorflow

Après avoir exécuté cela, vous remarquerez que un nouveau répertoire nommé quickstart-tensorflow a été créé. Il devrait avoir la structure suivante :

quickstart-tensorflow
├── tfexample
│   ├── __init__.py
│   ├── client_app.py   # Defines your ClientApp
│   ├── server_app.py   # Defines your ServerApp
│   └── task.py         # Defines your model, training and data loading
├── pyproject.toml      # Project metadata like dependencies and configs
└── README.md

Si vous n’avez pas encore installé le projet et ses dépendances, vous pouvez le faire avec :

# From the directory where your pyproject.toml is
$ pip install -e .

Pour lancer le projet, faites:

# Run with default arguments and stream logs
$ flwr run . --stream

Le processus flwr run . soumet l’exécution, imprime l’ID d’exécution et retourne sans diffuser les journaux. Pour le workflow local complet, voir Exécuter Flower Localement avec un SuperLink Géré.

Avec les arguments par défaut, vous verrez une sortie en flux continu comme celle-ci :

Starting local SuperLink on 127.0.0.1:39093...
Successfully started run 1859953118041441032
INFO :      Starting FedAvg strategy:
INFO :          ├── Number of rounds: 3
INFO :      [ROUND 1/3]
INFO :      configure_train: Sampled 5 nodes (out of 10)
INFO :      aggregate_train: Received 5 results and 0 failures
INFO :          └──> Aggregated MetricRecord: {'train_loss': 2.0013, 'train_acc': 0.2624}
INFO :      configure_evaluate: Sampled 10 nodes (out of 10)
INFO :      aggregate_evaluate: Received 10 results and 0 failures
INFO :          └──> Aggregated MetricRecord: {'eval_acc': 0.1216, 'eval_loss': 2.2686}
INFO :      [ROUND 2/3]
INFO :      ...
INFO :      [ROUND 3/3]
INFO :      ...
INFO :      Strategy execution finished in 16.60s
INFO :      Final results:
INFO :          ServerApp-side Evaluate Metrics:
INFO :          {}

Vous pouvez également surcharger les paramètres définis dans la section [tool.flwr.app.config] de pyproject.toml comme suit:

# Override some arguments
$ flwr run . --run-config "num-server-rounds=5 batch-size=16"

Les données¶

Ce tutoriel utilise Flower Datasets pour télécharger et partitionner facilement le jeu de données CIFAR-10. Dans cet exemple, vous utiliserez la méthode IidPartitioner pour générer des partitions num_partitions. Vous pouvez choisir parmi les other partitioners disponibles dans Flower Datasets. Chaque partie ClientApp appellera cette fonction pour créer les tableaux NumPy correspondant à leur partition de données.

partitioner = IidPartitioner(num_partitions=num_partitions)
fds = FederatedDataset(
    dataset="uoft-cs/cifar10",
    partitioners={"train": partitioner},
)
partition = fds.load_partition(partition_id, "train")
partition.set_format("numpy")

# Divide data on each node: 80% train, 20% test
partition = partition.train_test_split(test_size=0.2)
x_train, y_train = partition["train"]["img"] / 255.0, partition["train"]["label"]
x_test, y_test = partition["test"]["img"] / 255.0, partition["test"]["label"]

Le Modèle¶

Ensuite, nous avons besoin d’un modèle. Nous avons défini un simple réseau neuronal convolutionnel (CNN), mais vous pouvez le remplacer par un modèle plus sophistiqué si vous le souhaitez :

def load_model(learning_rate: float = 0.001):
    # Define a simple CNN for CIFAR-10 and set Adam optimizer
    model = keras.Sequential(
        [
            keras.Input(shape=(32, 32, 3)),
            layers.Conv2D(32, kernel_size=(3, 3), activation="relu"),
            layers.MaxPooling2D(pool_size=(2, 2)),
            layers.Conv2D(64, kernel_size=(3, 3), activation="relu"),
            layers.MaxPooling2D(pool_size=(2, 2)),
            layers.Flatten(),
            layers.Dropout(0.5),
            layers.Dense(10, activation="softmax"),
        ]
    )
    optimizer = keras.optimizers.Adam(learning_rate)
    model.compile(
        optimizer=optimizer,
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"],
    )
    return model

L’Application Client¶

Les principales modifications que nous devons apporter pour utiliser Tensorflow avec Flower ont trait à la conversion du ArrayRecord reçu dans le Message en tableaux numpy pour utilisation avec la fonction intégrée set_weights(). Après entraînement, la fonction get_weights() peut être utilisée pour extraire puis packer les tableaux numpy mis à jour dans un Message du ClientApp. Nous pouvons faire usage des méthodes intégrées dans le ArrayRecord pour effectuer ces conversions:

@app.train()
def train(msg: Message, context: Context):

    # Load the model
    model = load_model(context.run_config["learning-rate"])
    # Extract the ArrayRecord from Message and convert to numpy ndarrays
    model.set_weights(msg.content["arrays"].to_numpy_ndarrays())

    # Train the model
    ...

    # Pack the model weights into an ArrayRecord
    model_record = ArrayRecord(model.get_weights())

Le reste de la fonctionnalité est directement inspiré du cas centralisé. Le ClientApp comporte trois méthodes de base (train, evaluate, et query) que nous pouvons implémenter à des fins différentes. Par exemple : train pour entraîner le modèle reçu en utilisant les données locales ; evaluate pour évaluer la performance du modèle reçu sur un jeu de validation ; et query pour récupérer des informations sur le nœud exécutant le ClientApp. Dans ce tutoriel, nous ne ferons que faire usage de train et evaluate.

Voyons comment la méthode train peut être implémentée. Elle reçoit en arguments d’entrée un Message depuis le ServerApp. Par défaut, elle porte :

un ArrayRecord avec les tableaux du modèle à fédérer. Par défaut, ils peuvent être récupérés avec la clé "arrays" lors de l’accès au contenu du message.
une ConfigRecord avec la configuration transmise depuis le ServerApp. Par défaut, elle peut être récupérée avec la clé "config" lors de l’accès au contenu du message.

La méthode train reçoit également le Context, donnant accès aux configurations pour votre exécution et nœud. Les hyperparamètres de la configuration d’exécution sont définis dans la section pyproject.toml de votre application Flower. La configuration du nœud ne peut être configurée que lors de l’exécution de Flower avec le Deployment Runtime et ce, directement configurable pendant les simulations.

# Flower ClientApp
app = ClientApp()


@app.train()
def train(msg: Message, context: Context):
    """Train the model on local data."""

    # Reset local Tensorflow state
    keras.backend.clear_session()

    # Load the data
    partition_id = context.node_config["partition-id"]
    num_partitions = context.node_config["num-partitions"]
    x_train, y_train, _, _ = load_data(partition_id, num_partitions)

    # Load the model
    model = load_model(context.run_config["learning-rate"])
    model.set_weights(msg.content["arrays"].to_numpy_ndarrays())
    epochs = context.run_config["local-epochs"]
    batch_size = context.run_config["batch-size"]
    verbose = context.run_config.get("verbose")

    # Train the model
    history = model.fit(
        x_train,
        y_train,
        epochs=epochs,
        batch_size=batch_size,
        verbose=verbose,
    )

    # Get training metrics
    train_loss = history.history["loss"][-1] if "loss" in history.history else None
    train_acc = (
        history.history["accuracy"][-1] if "accuracy" in history.history else None
    )

    # Pack and send the model weights and metrics as a message
    model_record = ArrayRecord(model.get_weights())
    metrics = {"num-examples": len(x_train)}
    if train_loss is not None:
        metrics["train_loss"] = train_loss
    if train_acc is not None:
        metrics["train_acc"] = train_acc
    content = RecordDict({"arrays": model_record, "metrics": MetricRecord(metrics)})
    return Message(content=content, reply_to=msg)

La méthode @app.evaluate() serait presque identique avec deux exceptions : (1) le modèle n’est pas entraîné localement, mais il est utilisé pour évaluer sa performance sur le jeu de validation local ; (2) inclure le modèle dans la réponse Message n’est plus nécessaire car il ne subit aucune modification locale.

L’Application Serveur¶

Pour construire un ServerApp, nous définissons sa méthode @app.main(). Cette méthode reçoit en arguments d’entrée :

Un objet Grid qui sera utilisé pour interagir avec les nœuds s’exécutant la ClientApp afin d’impliquer les clients dans une ronde d’entraînement/évaluation/requête ou autre.
Un objet Context qui fournit accès à la configuration de l’exécution.

Dans cet exemple, nous utilisons FedAvg et la configurons avec une valeur spécifique de fraction_train qui est lue à partir de la configuration d’exécution. Vous pouvez trouver la valeur par défaut définie dans la partie pyproject.toml. Ensuite, l’exécution de la stratégie est lancée lorsqu’on invoque sa méthode start. À cela on passe:

l’objet Grid.
un ArrayRecord portant un modèle initialisé aléatoirement qui servira de modèle global pour l’apprentissage fédéré.
Le paramètre num_rounds spécifiant combien de rondes de FedAvg effectuer.

# Create the ServerApp
app = ServerApp()


@app.main()
def main(grid: Grid, context: Context) -> None:
    """Main entry point for the ServerApp."""
    # Load config
    num_rounds = context.run_config["num-server-rounds"]
    fraction_train = context.run_config["fraction-train"]

    # Load initial model
    model = load_model()
    arrays = ArrayRecord(model.get_weights())

    # Define and start FedAvg strategy
    strategy = FedAvg(
        fraction_train=fraction_train,
    )

    result = strategy.start(
        grid=grid,
        initial_arrays=arrays,
        num_rounds=num_rounds,
    )

    if context.run_config["save-model"]:
        # Save the final model
        ndarrays = result.arrays.to_numpy_ndarrays()
        final_model_name = "final_model.keras"
        print(f"Saving final model to disk as {final_model_name}...")
        model.set_weights(ndarrays)
        model.save(final_model_name)

Notez que la méthode start du stratégie retourne un objet de résultat. Cet objet contient toutes les informations pertinentes sur le processus FL, y compris les poids finaux du modèle comme un ArrayRecord, et des métriques d’entraînement et d’évaluation fédérées comme des MetricRecords. Vous pouvez facilement enregistrer les métriques à l’aide de Python’s pprint et sauvegarder les poids du modèle final à l’aide de la fonction Tensorflow’s save().

Félicitations ! Vous avez réussi à créer et exécuter votre premier système d’apprentissage fédéré.

Astuce

Vérifiez la documentation de Run simulations pour en savoir plus sur la façon de configurer et d’exécuter les simulations Flower.

Note

Vérifiez le code source de la version étendue de ce tutoriel dans examples/quickstart-tensorflow dans le dépôt GitHub Flower.