Я оптимизирую планирование пути AGV для повышения энергоэффективности с помощью RL. Я не могу понять, почему сеть не обу ⇐ Python

[Anonymous]

Итак, я написал этот код в меру своих возможностей. Я отладил его, и после некоторых усилий он начал работать. Но сеть все еще не учится. Перепробовав тысячи комбинаций вознаграждений, реализаций алгоритмов, гиперпараметров и различных исправлений ошибок, нейронная сеть по-прежнему не тренируется правильно. Лучшими результатами, которых мне удалось достичь, были 500 успешных достижений целей из 5000 эпизодов, а я еще даже не начал рассматривать энергоэффективный путь. С текущими гиперпараметрами и вознаграждениями я могу справиться только с 80 успешными достижениями целей из 5000 эпизодов. Затем либо сеть начинает работать хуже, либо я получаю сообщение об ошибке, указывающее, что в выходных данных сети есть значения NaN (бесконечные или другие недопустимые значения).
Я знаю, что мои навыки Python определенно могут помочь будь лучше. Я также уверен, что у меня есть несколько действительно глупых ошибок (как логических, так и синтаксических), которые я, вероятно, пропустил в коде, тем более что никто больше не проверял проект. Буду очень признателен за любые отзывы по всем возможным аспектам.
У меня могут быть не только простые ошибки в коде, но и алгоритм PPO может быть реализован неправильно, или выбранная нейронная сеть и ее структура может быть неправильной.
Вот структура проекта:

• Папка PPO: содержит два файла:
  
  actor_critic.py: здесь реализуется нейронная сеть.
• ppo.py: реализуется алгоритм оптимизации проксимальной политики (PPO).< /li>

[*]Папка Train: на данный момент содержит только один файл:

• hyperparameter_optimization .py: Изначально этот файл использовался для случайного поиска лучших гиперпараметров. В настоящий момент он используется для попытки обучения основной сети.

[*]environment.py: Этот файл содержит среду. с которым взаимодействует агент.
[*]Main.py: просто служит точкой входа для запуска всего проекта.
[*]path Папка: используется для хранения журналы и сети по мере необходимости.
[*]config.yaml: содержит несколько переменных, двумерную сетку с препятствиями (0 для свободного места, 1 для препятствия) и путь для сохранения. различные предметы.

В окружающей среде для расчета энергопотребления используются упрощенные формулы с заранее заданными коэффициентами.
< strong>environment.py:

Код: Выделить всё

import numpy as np
import logging
logging.basicConfig(filename='path where to save logs', level=logging.DEBUG, format='%(asctime)s %(message)s')

def log_data(message):
logging.debug(message)
class AGVEnvironment:
def __init__(self, map_size, start, goal, obstacles, mass_agv=100, mass_cargo=100, max_speed=5, max_acceleration=2, k_att=1.0, k_rep=100.0, rep_range=5.0):
self.map_size = map_size
self.start = start
self.goal = goal
self.obstacles = obstacles
self.mass_agv = mass_agv
self.mass_cargo = mass_cargo
self.k_linear = 0.1
self.k_turn = 0.05
self.k_mass = 0.02
self.k_friction = 0.03
self.k_speed = 0.01
self.k_acceleration = 0.02
self.k_deceleration = 0.02
self.eta_engine = 0.9
self.max_speed = max_speed
self.max_acceleration = max_acceleration
self.k_att = k_att
self.k_rep = k_rep
self.rep_range = rep_range
self.energy_current_step = 0
self.reset()

def reset(self):
self.position = self.start
self.previous_position = self.start
self.angle = 0
self.speed = 0
self.acceleration = 0
self.total_energy = 0
self.steps = 0
self.total_time = 0
self.done = False
return self._get_state()

def _get_state(self):
distance_to_goal = np.linalg.norm(np.array(self.position) - np.array(self.goal))
potential_field = self._calculate_potential_field()

# Converting to arrays
position = np.array(self.position, dtype=float).flatten()
additional_info = np.array([float(self.angle), float(self.speed), float(self.acceleration), float(distance_to_goal), float(self.total_time)]).flatten()
potential_field = np.array(potential_field, dtype=float).flatten()

state = np.hstack((position, additional_info, potential_field))

return state

def _calculate_potential_field(self):
# Attracting force
att_force = self.k_att * (np.array(self.goal) - np.array(self.position))

# Repulsive force
rep_force = np.zeros(2, dtype=np.float64)
radius = 10  # Radius of the area around the agent's position,to reduce the search area
x, y = int(self.position[0]), int(self.position[1])

# Iterate only over cells within the radius
for i in range(max(0, x - radius), min(self.map_size[0], x + radius + 1)):
for j in range(max(0, y - radius), min(self.map_size[1], y + radius + 1)):
if self.obstacles[j][i] == 1:
obstacle_pos = np.array([i, j])
distance = np.linalg.norm(obstacle_pos - np.array(self.position))
if distance == 0:
rep_force = np.array([-100.0, -100.0], dtype=np.float64)
continue  # Skip the calculation if the agent has already crashed

if distance < self.rep_range:
rep_force += np.float64(self.k_rep) * (1.0 / distance - 1.0 / np.float64(self.rep_range)) * (1.0 / (distance**2)) * (np.array(self.position, dtype=np.float64) - np.array(obstacle_pos, dtype=np.float64)) / distance

total_force = att_force + rep_force
return total_force

def step(self, action):
if self.done:
raise Exception("Episode has finished.  Please reset the environment.")
previous_state = self._get_state()
discrete_action, continuous_action = action
self._apply_discrete_action(discrete_action)
if self.position == self.previous_position: # Check for going out of bounds and return values
reward = np.array([-100.00])
return previous_state, reward, self.done
self._apply_continuous_action(continuous_action)
self.steps += 1
time_step = self._calculate_time()
self.total_time += time_step
self._calculate_energy_consumption(time_step)

reward = self._calculate_reward()
if self.done: # Check if true has already exited from apply_discrete action
return self._get_state(), reward, self.done
self.done = self._check_done()

return self._get_state(), reward, self.done

def _apply_discrete_action(self, action):
x, y = self.position
self.previous_position = self.position

if action == 0:  # Up
self.position = (x, y - 1)
elif action == 1:  # Down
self.position = (x, y + 1)
elif action == 2:  # Left
self.position = (x - 1, y)
elif action == 3:  # Right
self.position = (x + 1, y)
elif action == 4:  # Up-left
self.position = (x - 1, y - 1)
elif action == 5:  # Up-right
self.position = (x + 1, y - 1)
elif action == 6:  # Down-left
self.position = (x - 1, y + 1)
elif action == 7:  # Down-right
self.position = (x + 1, y + 1)

# Handling map out-of-bounds and collisions with obstacles

if 0 = 3000:
#         log_data(f'Model saved Hyperparameters {hyperparams} are effective. {reached_goal_count} goals reached after 100000 episodes.')
#         print(f'Hyperparameters {hyperparams} effective. {reached_goal_count} goals reached after 100000 episodes.')
#         ppo.save(os.path.join(config['checkpoint_dir'], f'PPO_{episode_reward}.pth'))
if (episode + 1) % 20000 == 0:
if reached_goal_count  0:
avg_time = total_time_sum / reached_goal_count
avg_steps = total_steps_sum / reached_goal_count
avg_energy = total_energy_sum / reached_goal_count
else:
avg_time = avg_steps = avg_energy = 0

print(f'Episode {episode + 1} summary:')
print(f'Total reached goals: {reached_goal_count}')
print(f'Average time to reach goal: {avg_time}')
print(f'Average steps to reach goal: {avg_steps}')
print(f'Average energy consumption to reach goal: {avg_energy}')
log_data(f'Episode {episode + 1} summary:')
log_data(f'Total reached goals: {reached_goal_count}')
log_data(f'Average time to reach goal: {avg_time}')
log_data(f'Average steps to reach goal: {avg_steps}')
log_data(f'Average energy consumption to reach goal: {avg_energy}')
# Reset statistics for the next block of 5000 episodes
reached_goal_count = 0
total_time_sum = 0
total_steps_sum = 0
total_energy_sum = 0

if len(memory.states) >= 10 * hyperparams['batch_size']:
ppo.update(memory, hyperparams['batch_size'])
memory.clear_memory()

# ppo.save(os.path.join(config['checkpoint_dir'], f'PPO_{episode}.pth'))
avg_reward = total_reward / config['num_eval_episodes']
#plt.close(fig)
return avg_reward

def optimize_hyperparameters(num_trials=1000, early_stopping_rounds=50):
config = load_config()
best_hyperparams = None
best_reward = -float('inf')
no_improvement_rounds = 0

for trial in range(num_trials):
hyperparams = random_hyperparameters()

avg_reward = evaluate_hyperparameters(hyperparams, config)

log_data(f'Trial {trial+1}/{num_trials}: Hyperparameters: {hyperparams} \t Average Reward: {avg_reward}')
print(f'Trial {trial+1}/{num_trials}: Hyperparameters: {hyperparams} \t Average Reward: {avg_reward}')

if avg_reward > best_reward:
best_reward = avg_reward
best_hyperparams = hyperparams
no_improvement_rounds = 0

# Save best hyperparameters
with open(os.path.join(config['checkpoint_dir'], 'best_hyperparams.yaml'), 'w') as file:
yaml.dump(best_hyperparams, file)
else:
no_improvement_rounds += 1

if no_improvement_rounds >= early_stopping_rounds:
print(f'Early stopping triggered after {trial+1} trials.')
break

print(f'Best Hyperparameters: {best_hyperparams} \t Best Average Reward: {best_reward}')
return best_hyperparams

if __name__ == "__main__":
optimize_hyperparameters()

Main.py:

Код: Выделить всё

from Train.hyperparameter_optimization import optimize_hyperparameters
if __name__ == "__main__":
num_trials = 10000
early_stopping_rounds = 10000
optimize_hyperparameters(num_trials=num_trials, early_stopping_rounds=early_stopping_rounds)

config.yaml:

Код: Выделить всё

map_size: [5, 5]
start: [0, 0]
goal: [4, 4]
obstacles:  [[0, 0, 0, 0, 0],
[0, 0, 0, 0, 0],
[0, 0, 1, 0, 0],
[0, 0, 0, 0, 0],
[0, 0, 0, 0, 0]]
mass_agv: 100
mass_cargo: 100
max_speed: 5
max_acceleration: 2

checkpoint_dir: "path to folder path"

max_episodes: 1000
max_timesteps: 300
num_eval_episodes: 100000
log_interval: 10
save_interval: 50
num_trials: 10000
early_stopping_rounds: 10000

Использованная карта очень проста: сетка 5x5 с одним препятствием, расположенным по диагонали (даже на этой базовой карте я все еще не могу добиться приличных результатов).


Подробнее здесь: https://stackoverflow.com/questions/790 ... ant-figure