AttributeError: 'SymbolicTensor' object has no attribute 'numpy'を解決したい

Question

実現したいこと

AttributeError: 'SymbolicTensor' object has no attribute 'numpy'を解決したい

発生している問題・分からないこと

Pythonでkerasを使ったMLPモデルで予測をしています。
評価関数はF1マクロ平均でカスタム関数を作っています。
使用環境はGoogle Colabです。
目的変数は保険料の価格帯で、[0,1,2]から構成される多クラス分類です。

エラーメッセージ

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
<ipython-input-9-61406dba8cad> in <cell line: 70>()
     68 
     69 # モデルのトレーニング
---> 70 history = model.fit(X_train_split, y_train_split, validation_data=(X_val, y_val), epochs=50, batch_size=32, verbose=1)
     71 
     72 # バリデーションデータで予測

2 frames
/tmp/__autograph_generated_filevx5nsktk.py in tf__update_state(self, y_true, y_pred, sample_weight)
     10                 y_true_labels = ag__.converted_call(ag__.ld(tf).argmax, (ag__.ld(y_true),), dict(axis=1), fscope)
     11                 y_pred_labels = ag__.converted_call(ag__.ld(tf).argmax, (ag__.ld(y_pred),), dict(axis=1), fscope)
---> 12                 y_true_np = ag__.converted_call(ag__.ld(y_true).numpy, (), None, fscope)
     13                 y_pred_np = ag__.converted_call(ag__.ld(y_pred).numpy, (), None, fscope)
     14                 f1 = ag__.converted_call(ag__.ld(tf).py_function, (), dict(func=ag__.ld(custom_f1_score), inp=[ag__.ld(y_true_np), ag__.ld(y_pred_np)], Tout=ag__.ld(tf).float32, name='f1_score'), fscope)

AttributeError: 'SymbolicTensor' object has no attribute 'numpy'

該当のソースコード

# 必要なライブラリのインストール
!pip install tensorflow

import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix, f1_score

# カスタムF1スコアメトリックの定義
class F1ScoreMacro(tf.keras.metrics.Metric):
    def __init__(self, name='f1_score_macro', **kwargs):
        super(F1ScoreMacro, self).__init__(name=name, **kwargs)
        self.f1 = self.add_weight(name='f1', initializer='zeros', dtype=tf.float32)

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = tf.argmax(y_true, axis=1, output_type=tf.int32)
        y_pred = tf.argmax(y_pred, axis=1, output_type=tf.int32)

        # Extract class labels from one-hot encoded tensors
        y_true_labels = tf.argmax(y_true, axis=1)
        y_pred_labels = tf.argmax(y_pred, axis=1)

        y_true_np = y_true.numpy()
        y_pred_np = y_pred.numpy()

        f1 = tf.py_function(func=custom_f1_score, inp=[y_true_np, y_pred_np], Tout=tf.float32, name='f1_score')
        self.f1.assign_add(f1)

    def result(self):
        return self.f1

    def reset_states(self):
        self.f1.assign(0.0)

#
,,,
(dataの読込、分割)
,,,

# モデルの構築
model = Sequential()
model.add(Dense(64, input_dim=X_train_split.shape[1], activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(32, activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(3, activation='softmax'))  # 出力層、クラス数は3

# モデルのコンパイル
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=[F1ScoreMacro()])

# モデルのトレーニング
history = model.fit(X_train_split, y_train_split, validation_data=(X_val, y_val), epochs=50, batch_size=32, verbose=1)

試したこと・調べたこと

• teratailやGoogle等で検索した
• ソースコードを自分なりに変更した
• 知人に聞いた
• その他

上記の詳細・結果

カスタム関数である、F1ScoreMacro()の内容を、historyへ渡すときにエラーが起きてる認識です。
具体的には、カスタム関数内のTensor型のデータをNumpy型へ変換できずに、矛盾が起きていると考えています。
kerasはgraph executionで動くので、eager excutionに直すための"run_eagerly=True"を入れたり、
F1ScoreMacro()にNumpyに変換するためのコードを入れたりしましたが、改善されないので、質問しました。
(以下、そのコードです)

class F1ScoreMacro(tf.keras.metrics.Metric):
    def __init__(self, name='f1_score_macro', **kwargs):
        super(F1ScoreMacro, self).__init__(name=name, **kwargs)
        self.f1 = self.add_weight(name='f1', initializer='zeros', dtype=tf.float32)

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = tf.argmax(y_true, axis=1)
        y_pred = tf.argmax(y_pred, axis=1)
        
        def f1_score_np(y_true, y_pred):
            return f1_score(y_true, y_pred, average='macro')
        
        f1 = tf.numpy_function(f1_score_np, [y_true, y_pred], tf.float32)
        self.f1.assign_add(f1)

    def result(self):
        return self.f1

    def reset_states(self):
        self.f1.assign(0.0)

補足

かなり雑多な質問になってしまいましたが、どなたか改善方法などわかる方いましたら、ぜひご教授いただきたいです。
よろしくお願いいたします。

Comments

[meg_]

> 2 frames
上記は展開できませんか？

訓練データ、検証データの型は何でしょうか？

[bsdfan]

tf.py_function の関数の中(custom_f1_score の中)で、.numpy() してやらないといけないのではないでしょうか。

Answer 1

皆さん、ご回答いただきありがとうございました。
以下の修正をすることで、エラーの解消になりましたので、ご報告いたします。

①Eager Executionを明示的に有効にする
tf.config.experimental_run_functions_eagerly(True)
②tf.numpy_functionのdtypeをtf.float64に設定する
tf.numpy_function(f1_score_np, [y_true, y_pred], tf.float64)
③self.f1.assign_addの引数をtf.castでfloat32にキャストする
self.f1.assign_add(tf.cast(f1, tf.float32))

以上の結果、MLPモデルを実装することができました。
コード全文も記載します。
皆さん回答いただきまして、ありがとうございました。

# Google Driveをマウント
from google.colab import drive
drive.mount('/content/drive')

# 必要なライブラリのインストール
!pip install tensorflow

import pandas as pd
import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix, f1_score

# Eager Executionを有効にする（TensorFlow 2.xではデフォルトで有効）
tf.config.experimental_run_functions_eagerly(True)

# tf.data APIでもEager Executionを有効にする
tf.data.experimental.enable_debug_mode()

# カスタムF1スコアメトリックの定義
class F1ScoreMacro(tf.keras.metrics.Metric):
    def __init__(self, name='f1_score_macro', **kwargs):
        super(F1ScoreMacro, self).__init__(name=name, **kwargs)
        self.f1 = self.add_weight(name='f1', initializer='zeros', dtype=tf.float32)

    def update_state(self, y_true, y_pred, sample_weight=None):
        y_true = tf.argmax(y_true, axis=1)
        y_pred = tf.argmax(y_pred, axis=1)

        def f1_score_np(y_true, y_pred):
            return f1_score(y_true, y_pred, average='macro')

        f1 = tf.numpy_function(f1_score_np, [y_true, y_pred], tf.float64)
        self.f1.assign_add(tf.cast(f1, tf.float32))

    def result(self):
        return self.f1

    def reset_state(self):  # メソッド名を変更
        self.f1.assign(0.0)

# データの読み込み
data = pd.read_csv('MYPATH')

# 本来のテストデータを分離
train_data = data[data['charges'] != -1]
test_data = data[data['charges'] == -1]

# トレーニングデータの特徴量と目的変数に分割
X_train = train_data.drop('charges', axis=1)
y_train = pd.get_dummies(train_data['charges'])  # One-hotエンコーディング

# テストデータの特徴量に分割（目的変数は使用しない）
X_test = test_data.drop('charges', axis=1)

# 学習用データをさらにトレーニングセットとバリデーションセットに分割
X_train_split, X_val, y_train_split, y_val = train_test_split(X_train, y_train, test_size=0.2, random_state=42)

# クラス重みの定義
class_weights = {0: 1.0, 1: 5.0, 2: 5.0}

# モデルの構築
model = Sequential()
model.add(Dense(64, input_dim=X_train_split.shape[1], activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(32, activation='relu'))
model.add(Dropout(0.5))
model.add(Dense(3, activation='softmax'))  # 出力層、クラス数は3

# モデルのコンパイル
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=[F1ScoreMacro()])

# モデルのトレーニング
history = model.fit(X_train_split, y_train_split, validation_data=(X_val, y_val), epochs=50, batch_size=32, verbose=1, class_weight=class_weights)

# バリデーションデータで予測
y_pred_prob = model.predict(X_val)
y_pred = np.argmax(y_pred_prob, axis=1)
y_val_labels = np.argmax(y_val.values, axis=1)

# モデルの評価
conf_matrix = confusion_matrix(y_val_labels, y_pred)
class_report = classification_report(y_val_labels, y_pred, zero_division=0)
f1 = f1_score(y_val_labels, y_pred, average='macro')

print("Confusion Matrix:\n", conf_matrix)
print("\nClassification Report:\n", class_report)
print("\nMacro Average F1 Score:", f1)