Как преобразовать SymbolicTensor в np.ndarray при использовании @tf.function в Keras для RL агента?

Я хочу написать RL агента на keras, но у меня возникает проблема. Класс я написал, но при попытке вызвать функцию DQN.train_step вылетает ошибка AttributeError: 'SymbolicTensor' object has no attribute 'numpy'. Из-за декоратора @tf.function нейросеть возвращает объект класса tensorflow.python.framework.ops.SymbolicTensor, а не как обычно np.ndarray. Я не хочу убирать декоратор, потому что он ускоряет работу программы примерно в 50 раз. Проблема в этом участке кода:

with tf.GradientTape() as tape:
    current_q_values = self.predict(state)
    next_q_values = self.predict(next_state)
    max_next_q = tf.reduce_max(next_q_values)
    target_q_values = current_q_values.numpy()  # здесь возникает ошибка
    target_q_values[0, action] = reward + self.discount_factor * max_next_q * (1 - done)
    loss = self.loss_fn(current_q_values, target_q_values)

В строке target_q_values = current_q_values.numpy() надо заменить current_q_values.numpy() на какую-то функцию или что-то ещё, что будет изменять тип данных c tensorflow.python.framework.ops.SymbolicTensor на np.ndarray, но как это сделать я не знаю.

import numpy as np
import tensorflow as tf
from keras.models import Model

class DQN:
    def __init__(self, model, optimizer, loss_fn,
                 discount_factor: int | float = 0.98, lambda_: int | float = 0):
        """
        :param model: Нейросеть, keras.models.Sequential.
        :param optimizer: Оптимизатор нейросети.
        :param loss_fn: Функция потерь нейросети.
        :param discount_factor: Дисконтирующий фактор / насколько важна награда в следующем состоянии для агента.
        :param lambda_: Насколько важно помнить ранее приобретённые навыки.
        """

        self.model = layers
        self.optimizer = optimizer
        self.loss_fn = loss_fn
        self.discount_factor = discount_factor
        self.lambda_ = lambda_
        self.prior_weights = self.fisher_matrix = None

    @tf.function
    def train_step(self, state: np.ndarray, action: int, reward: int | float,
                       next_state: np.ndarray, done: int):
        """
        :param state: Состояние, на основе которого агент принял решение совершить действие action.
        :param action: Действие, которое совершило переход из состояния state в состояние next_state.
        :param reward: Награда за действие action.
        :param next_state: Состояние, в которое был совершён переход из состояния state с помощью действия action.
        :param done: Терминальное состояние, принимает значение 0 или 1.
        """

        with tf.GradientTape() as tape:
            current_q_values = self.predict(state)
            next_q_values = self.predict(next_state)
            max_next_q = tf.reduce_max(next_q_values)
            target_q_values = current_q_values.numpy()
            target_q_values[0, action] = reward + self.discount_factor * max_next_q * (1 - done)
            loss = self.loss_fn(current_q_values, target_q_values)

            loss += compute_penalty_loss(self.weights, self.prior_weights, self.fisher_matrix, self.lambda_)

        gradients = tape.gradient(loss, self.trainable_variables)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_variables))

    def predict(self, x, *args, **kwargs):
        return self.model(x, *args, **kwargs)

Я не знаю как мне решить эту проблему, помогите мне, пожалуйста.

Версии библиотек, которые я использую:

  • tensorflow 2.16.2
  • keras 3.4.1
  • numpy 1.26.4

Ответы (1 шт):

Автор решения: Bobby_Shmurda31

Надо изменить код в with tf.GradientTape() as tape. Здесь просто адаптирован код под тип данных tensorflow.python.framework.ops.SymbolicTensor, массивы здесь не используются.

with tf.GradientTape() as tape:
    current_q_values = self.predict(state)
    next_q_values = self.predict(next_state)
    max_next_q = tf.reduce_max(next_q_values)

    # Создаем копию current_q_values и обновляем её значения
    target_q_values = tf.identity(current_q_values)
    target_q_values = tf.tensor_scatter_nd_update(
        target_q_values,
        indices=[[0, action]],  # Обновляем значение Q для данного действия
        updates=[reward + self.discount_factor * max_next_q * (1 - done)]
    )

    loss = self.loss_fn(current_q_values, target_q_values)

    loss += compute_penalty_loss(self.weights, self.prior_weights, self.fisher_matrix, self.lambda_)
→ Ссылка