TensorFlow Federated 0.76.0中build_federated_averaging_process属性缺失求助

问题重现 #

以下是基于鸢尾花数据集的联邦学习代码：

import pandas as pd
import numpy as np
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

import tensorflow as tf
import tensorflow_federated as tff

iris = load_iris()
df = pd.DataFrame(iris.data,columns=iris.feature_names)
df['Species']=iris.target

# Splitting the dataframe into input features and target variables
x = df.drop('Species',axis=1)
y = df['Species']

# Function to create client datasets (assuming data is pre-partitioned)
def create_tf_dataset(client_data):
  """Creates a tf.data.Dataset from the provided client data (features, labels)."""
  features, labels = client_data
  return tf.data.Dataset.from_tensor_slices((features, labels))

# Split data into client datasets (simulating data partitioning)
client_datasets = []
num_clients = 5
for i in range(num_clients):
  start_index = int(i * (len(x) / num_clients))
  end_index = int((i + 1) * (len(x) / num_clients))
  client_features = x[start_index:end_index]
  client_labels = y[start_index:end_index]
  client_datasets.append(create_tf_dataset((client_features, client_labels)))

# Define the model architecture (replace with your desired model complexity)
def model_fn(inputs):
   features, _ = inputs  # We only use features for classification
   dense1 = tf.keras.layers.Dense(10, activation='relu')(features)
   dense2 = tf.keras.layers.Dense(3, activation='softmax')(dense1)  # 3 units for 3 Iris classes
   return tf.keras.Model(inputs=features, outputs=dense2)

# Define the client optimizer
client_optimizer = tf.keras.optimizers.SGD(learning_rate=0.1)

# Define the server optimizer (for server-sided aggregation)
server_optimizer = tf.keras.optimizers.SGD(learning_rate=0.01)
fed_learning_model = tff.learning.build_federated_averaging_process(
     model_fn,
     client_optimizer_fn=client_optimizer,
     server_optimizer_fn=server_optimizer)

运行时触发错误：

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-13-e5966e29fc79> in <cell line: 1>()
----> 1 fed_learning_model = tff.learning.build_federated_averaging_process(
      2      model_fn,
      3      client_optimizer_fn=client_optimizer,
      4      server_optimizer_fn=server_optimizer)

AttributeError: module 'tensorflow_federated.python.learning' has no attribute 'build_federated_averaging_process'

环境配置 #

• Python 3.10.12
• TensorFlow 2.14.1
• TensorFlow Federated 0.76.0

解决方案 #

TensorFlow Federated从0.21.0版本开始弃用了tff.learning.build_federated_averaging_process，替代方案是使用tff.learning.algorithms.build_federated_averaging_process，同时需要调整模型定义方式以适配新API要求：

修改要点 #

1. 替换API路径：使用tff.learning.algorithms.build_federated_averaging_process替代旧API
2. 调整模型函数：新API要求返回tff.learning.Model，可通过tff.learning.from_keras_model将Keras模型转换为TFF兼容格式
3. 优化器传入方式：客户端和服务器优化器需通过函数返回实例，避免序列化问题
4. 数据集预处理：给客户端数据集添加batch操作，提升训练效率

修改后的完整代码 #

import pandas as pd
import numpy as np
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler

import tensorflow as tf
import tensorflow_federated as tff

# 加载并预处理鸢尾花数据集
iris = load_iris()
df = pd.DataFrame(iris.data, columns=iris.feature_names)
df['Species'] = iris.target

# 特征和标签分离
x = df.drop('Species', axis=1)
y = df['Species']

# 标准化特征
scaler = StandardScaler()
x_scaled = scaler.fit_transform(x)

# 创建客户端数据集的函数，添加batch处理
def create_tf_dataset(client_data):
    features, labels = client_data
    # 转换为one-hot标签，适配分类任务
    labels = tf.one_hot(labels, depth=3)
    return tf.data.Dataset.from_tensor_slices((features, labels)).batch(8)

# 模拟数据分区，分配给5个客户端
num_clients = 5
client_datasets = []
for i in range(num_clients):
    start_idx = int(i * len(x_scaled) / num_clients)
    end_idx = int((i + 1) * len(x_scaled) / num_clients)
    client_x = x_scaled[start_idx:end_idx]
    client_y = y[start_idx:end_idx]
    client_datasets.append(create_tf_dataset((client_x, client_y)))

# 定义Keras模型
def create_keras_model():
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(10, activation='relu', input_shape=(4,)),
        tf.keras.layers.Dense(3, activation='softmax')
    ])
    return model

# 定义TFF兼容的模型函数
def model_fn():
    keras_model = create_keras_model()
    return tff.learning.from_keras_model(
        keras_model,
        input_spec=client_datasets[0].element_spec,
        loss=tf.keras.losses.CategoricalCrossentropy(),
        metrics=[tf.keras.metrics.CategoricalAccuracy()]
    )

# 定义优化器生成函数
def client_optimizer_fn():
    return tf.keras.optimizers.SGD(learning_rate=0.1)

def server_optimizer_fn():
    return tf.keras.optimizers.SGD(learning_rate=0.01)

# 构建联邦平均训练流程
fed_avg_process = tff.learning.algorithms.build_federated_averaging_process(
    model_fn=model_fn,
    client_optimizer_fn=client_optimizer_fn,
    server_optimizer_fn=server_optimizer_fn
)

# 初始化服务器状态
server_state = fed_avg_process.initialize()

# 运行联邦训练
num_rounds = 10
for round_num in range(num_rounds):
    server_state, metrics = fed_avg_process.next(server_state, client_datasets)
    print(f"Round {round_num+1}: {metrics}")

关键说明 #

• tff.learning.from_keras_model负责将Keras模型包装为TFF可识别的模型，需指定输入规格、损失函数和评估指标
• 客户端数据集添加了batch操作，并且将标签转换为one-hot编码，适配分类任务的损失计算
• 优化器通过函数返回实例，确保TFF能够正确序列化和分发到客户端

内容的提问来源于stack exchange，提问作者Eldhose Kurian