fedlang-py 0.0.1

A Federated Learning Framework for Erlang-Python Integration

fedlang-py

Python Library for Federated Learning Experiments under the fedlang Framework

Overview

fedlang-py defines a set of Python classes required for implementing Federated Learning experiments under the fedlang framework [1]. Built on an actor-oriented middleware architecture, it provides a declarative approach to defining federated learning workflows through JSON-based configuration files, enabling researchers and practitioners to focus on algorithm design rather than distributed system implementation.

The library supports seamless integration with Erlang-based distributed systems and provides both local testing capabilities and distributed deployment options.

Key Features

• Declarative Orchestration: Define complete federated learning workflows using JSON configuration files
• Actor-Based Architecture: Clean separation between server and client entities with standardized communication patterns
• Flexible Execution Models: Support for both iterative and one-shot execution stages
• Erlang Interoperability: Built-in support for integration with Erlang-based distributed messaging systems
• Local Testing: Run and debug federated learning experiments locally before deployment
• Extensible Design: Easy integration of custom federated learning algorithms
• Lightweight: No external dependencies for core functionality

Installation

Using pip

pip install fedlang-py

From Source

git clone https://github.com/jlcorcuera/fedlang-py.git
cd fedlang-py
pip install -e .

Requirements

• Python 3.9 or higher
• No external dependencies for core functionality

Quick Start

1. Define Your Federation Plan

Create a JSON file that describes your federated learning workflow:

{
  "name": "My Federated Algorithm",
  "server_class": "my_package.server.MyServer",
  "client_class": "my_package.client.MyClient",
  "stages": [
    {
      "name": "Initialization",
      "type": "run_once",
      "pipeline": [
        { "entity": "#server", "method": "initialize" },
        { "entity": "#available_clients", "method": "setup" }
      ]
    },
    {
      "name": "Training Loop",
      "type": "run_many",
      "stop_condition": "is_converged",
      "pipeline": [
        { "entity": "#server", "method": "broadcast_model" },
        { "entity": "#available_clients", "method": "train_local" },
        { "entity": "#server", "method": "aggregate_updates" }
      ]
    }
  ]
}

2. Implement Server and Client Classes

from fedlangpy.core.entities import FedlangEntity, pickle_io

class MyServer(FedlangEntity):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.global_model = None
        self.round_count = 0

    def initialize(self, _):
        """Initialize the global model"""
        self.global_model = self._create_initial_model()
        return self.global_model

    @pickle_io
    def broadcast_model(self, _):
        """Send current model to clients"""
        return self.global_model

    @pickle_io
    def aggregate_updates(self, client_updates):
        """Aggregate updates from clients"""
        self.global_model = self._average_models(client_updates)
        self.round_count += 1
        return self.global_model

    def is_converged(self):
        """Check if training should stop"""
        return self.round_count >= self.parameters.get("max_rounds", 100)


class MyClient(FedlangEntity):
    def __init__(self, data=None, **kwargs):
        super().__init__(**kwargs)
        self.local_data = data
        self.local_model = None

    def setup(self, _):
        """Initialize client state"""
        self.local_model = None

    @pickle_io
    def train_local(self, global_model):
        """Train model on local data"""
        self.local_model = self._train(global_model, self.local_data)
        return self.local_model

3. Run the Experiment

from fedlangpy.core.utils import load_plan, run_experiment
from my_package.server import MyServer
from my_package.client import MyClient

# Load the federation plan
plan = load_plan('./my_plan.json')

# Create server and clients
server = MyServer(type='server')
clients = [
    MyClient(type='client', id=i, data=datasets[i])
    for i in range(num_clients)
]

# Define execution parameters
parameters = {
    "learning_rate": 0.01,
    "batch_size": 32,
    "max_rounds": 50
}

# Execute the federated learning experiment
run_experiment(plan, server, clients, parameters)

Architecture

Core Components

The framework consists of three main components:

1. Federation Plans (models.py)

Federation plans are structured configurations that define:

• FLPlan: Top-level configuration specifying server/client classes and execution stages
• FLStage: Individual stages that can be run_once or run_many (iterative)
• FLPipeline: Specific method invocations within each stage

2. Federated Entities (entities.py)

• FedlangEntity: Base class for server and client implementations

  • Provides call() method for dynamic method invocation
  • Handles parameter management via set_parameters()
  • Supports Erlang interoperability (byte array/list conversion)
  • Tracks current round number

• @pickle_io Decorator: Automatically serializes/deserializes complex data structures passed between entities

3. Execution Engine (utils.py)

• load_plan(): Loads and validates federation plans from JSON

  • Verifies class existence and importability
  • Validates method signatures
  • Checks stop conditions for iterative stages

• run_experiment(): Orchestrates the federated learning execution

  • Manages execution context (server, clients, parameters)
  • Routes method calls to appropriate entities
  • Handles iterative execution with stop conditions
  • Supports client selection filtering

Entity Types

The framework recognizes these entity types in pipeline definitions:

Entity	Description
#server	The single server instance
#available_clients	All registered client instances
#selected_clients	Filtered subset of clients (populated via pipeline outputs)
#parameters	Execution parameters dictionary

Execution Flow

1. Plan Loading: Federation plan is parsed and validated
2. Initialization: Server and clients are instantiated with parameters
3. Stage Execution: For each stage in sequence:
   • run_once stages: Execute pipeline once
   • run_many stages: Execute pipeline iteratively until stop condition is met
4. Pipeline Processing: For each step in the pipeline:
   • Identify target entity (server or client collection)
   • Invoke specified method with input from previous step
   • Store output in context for subsequent steps
5. Completion: Return control after all stages complete

Implementation Guidelines

Class Requirements

Requirement	Description
Inheritance	All server and client classes must extend FedlangEntity
Type Declaration	Instantiate with type='server' or type='client'
Method Signatures	Pipeline methods accept one argument; stop conditions accept none
Serialization	Use @pickle_io for methods exchanging complex objects
Parameters	Access via self.parameters["key"]

Example: Federated Averaging

from fedlangpy.core.entities import FedlangEntity, pickle_io
import numpy as np

class FedAvgServer(FedlangEntity):
    def init(self, _):
        self.model_weights = self._initialize_model()
        return self.model_weights

    @pickle_io
    def aggregate(self, client_weights):
        # Simple averaging
        self.model_weights = np.mean(client_weights, axis=0)
        return self.model_weights

    def should_stop(self):
        current_round = self.get_current_round()
        max_rounds = self.parameters["num_rounds"]
        return current_round >= max_rounds

class FedAvgClient(FedlangEntity):
    def __init__(self, data, labels, **kwargs):
        super().__init__(**kwargs)
        self.X = data
        self.y = labels

    @pickle_io
    def local_train(self, global_weights):
        # Train for specified epochs
        local_weights = self._sgd_train(
            global_weights,
            self.X,
            self.y,
            epochs=self.parameters["local_epochs"]
        )
        return local_weights

Advanced Features

Client Selection

Use pipeline outputs to filter clients dynamically:

{
  "pipeline": [
    {
      "entity": "#server",
      "method": "select_clients",
      "output": "#selected_clients"
    },
    {
      "entity": "#selected_clients",
      "method": "train"
    }
  ]
}

The server's select_clients() method should return a list of client IDs.

Input/Output Chaining

Pass data between pipeline steps using input/output specifications:

{
  "pipeline": [
    {
      "entity": "#server",
      "method": "prepare_task",
      "output": "task_data"
    },
    {
      "entity": "#available_clients",
      "method": "process_task",
      "input": "task_data"
    }
  ]
}

Round Tracking

Access the current iteration number in any entity:

def my_method(self, data):
    current_round = self.get_current_round()
    # Use round number for scheduling, logging, etc.

Erlang Integration

fedlang-py is designed to integrate with Erlang-based distributed systems. The framework handles conversion between Python and Erlang data types:

• Strings: Automatically converts between Python strings and Erlang character lists
• Binary Data: Handles byte arrays for complex object serialization
• Method Invocation: Supports dynamic method dispatch from Erlang nodes

Contributors

• Alessio Bechini - Google Scholar - [e-mail]
• José Luis Corcuera Bárcena - Google Scholar - [e-mail]

Citation

If you use fedlang-py in your research, please cite:

Paper:

A. Bechini, J. L. Corcuera Bárcena, "Devising an actor-based middleware support to federated learning experiments and systems", Future Generation Computer Systems, Volume 166, 2025, 107646. DOI: 10.1016/j.future.2024.107646

BibTeX:

@article{bechini2025devising,
  title={Devising an actor-based middleware support to federated learning experiments and systems},
  author={Bechini, Alessio and Corcuera Barcena, Jose Luis},
  journal={Future Generation Computer Systems},
  volume={166},
  pages={107646},
  year={2025},
  publisher={Elsevier},
  doi={10.1016/j.future.2024.107646}
}

Acknowledgments

This work has been partially funded by:

• PNRR - Tuscany Health Ecosystem (THE) - Spoke 6 - Precision Medicine & Personalized Healthcare (CUP I53C22000780001) under the NextGeneration EU programme
• PNRR - M4C2 - Investimento 1.3, Partenariato Esteso PE00000013 - FAIR - Future Artificial Intelligence Research - Spoke 1 "Human-centered AI"
• PON 2014-2021 - "Research and Innovation", DM MUR 1062/2021, Project: "Progettazione e sperimentazione di algoritmi di federated learning per data stream mining"
• Italian Ministry of University and Research (MUR) - FoReLab and CrossLab projects (Departments of Excellence)

License

This project is licensed under the Apache License 2.0. See the LICENSE file for details.

---

Built at the University of Pisa for the Federated Learning Research Community