Reinforcement Learning & Robotica

// Multi-Agent PPO Implementation met Centralized Training, Decentralized Execution
import torch
import torch.nn as nn
import torch.optim as optim
from typing import Dict, List, Tuple

class MultiAgentPPO:
    def __init__(self, num_agents: int, state_dim: int, action_dim: int):
        self.num_agents = num_agents
        self.agents = []

        # Initialiseer individuele agent policies
        for i in range(num_agents):
            agent = {
                'actor': ActorNetwork(state_dim, action_dim),
                'critic': CriticNetwork(state_dim * num_agents),  # Centralized critic
                'optimizer_actor': optim.Adam(actor.parameters(), lr=3e-4),
                'optimizer_critic': optim.Adam(critic.parameters(), lr=1e-3)
            }
            self.agents.append(agent)

    def select_actions(self, states: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
        actions = []
        log_probs = []

        with torch.no_grad():
            for i, agent in enumerate(self.agents):
                action_dist = agent['actor'](states[i])
                action = action_dist.sample()
                log_prob = action_dist.log_prob(action)

                actions.append(action)
                log_probs.append(log_prob)

        return torch.stack(actions), torch.stack(log_probs)

    def update(self, batch_data: Dict):
        for epoch in range(self.ppo_epochs):
            for agent_idx, agent in enumerate(self.agents):
                # Actor update met clipped surrogate objective
                ratio = torch.exp(new_log_probs - old_log_probs[agent_idx])
                clipped_ratio = torch.clamp(ratio, 1-self.clip_epsilon, 1+self.clip_epsilon)
                actor_loss = -torch.min(ratio * advantages, clipped_ratio * advantages).mean()

                # Critic update met global state information
                global_state = torch.cat(batch_data['states'], dim=-1)
                value_pred = agent['critic'](global_state)
                critic_loss = nn.MSELoss()(value_pred, batch_data['returns'])

                # Backpropagation
                agent['optimizer_actor'].zero_grad()
                actor_loss.backward()
                agent['optimizer_actor'].step()

                agent['optimizer_critic'].zero_grad()
                critic_loss.backward()
                agent['optimizer_critic'].step()

# Curriculum Learning Framework voor complexe taken
class CurriculumLearningScheduler:
    def __init__(self, initial_difficulty: float = 0.1):
        self.current_difficulty = initial_difficulty
        self.success_threshold = 0.8
        self.failure_threshold = 0.3
        self.difficulty_increment = 0.1

    def update_curriculum(self, success_rate: float) -> float:
        if success_rate > self.success_threshold:
            # Verhoog moeilijkheidsgraad
            self.current_difficulty = min(1.0, self.current_difficulty + self.difficulty_increment)
        elif success_rate < self.failure_threshold:
            # Verlaag moeilijkheidsgraad
            self.current_difficulty = max(0.1, self.current_difficulty - self.difficulty_increment)

        return self.current_difficulty

    def generate_task_parameters(self) -> Dict:
        return {
            'obstacle_density': self.current_difficulty,
            'target_speed': 0.5 + (self.current_difficulty * 1.5),
            'noise_level': self.current_difficulty * 0.2,
            'time_limit': 100 - (self.current_difficulty * 30)
        }