多智能體強化學習（MARL）介紹和代碼示例詳情 - 強化學習,tensorflow,代碼示例,JavaScript,前端開發,yyds乾貨盤點 WangLanguager 博客

多智能體強化學習（MARL）介紹

多智能體強化學習（Multi-Agent Reinforcement Learning, MARL）是強化學習的一個分支，涉及多個智能體在同一環境中學習和決策。MARL 主要關注以下幾個方面：

環境交互：

多個智能體同時與環境交互，它們的行為不僅影響自身的獎勵，也會影響其他智能體的獎勵。

合作與競爭：

智能體可以是合作的（例如共同完成任務）或競爭的（例如在博弈中爭奪資源），這使得學習變得更加複雜。

信息共享：

智能體之間可以共享信息，或者各自獨立工作，信息共享的程度會影響學習的效率和效果。

非平穩性：

由於環境中智能體的行為是動態的，單個智能體的學習過程變得非平穩，增加了學習的挑戰。

MARL 的基本算法

以下是一些常見的 MARL 算法：

獨立 Q 學習（Independent Q-Learning）：

每個智能體獨立地使用 Q 學習算法，視其他智能體為環境的一部分。

集中訓練，分散執行（CTDE）：

在訓練時使用全局信息，但在執行時每個智能體只使用局部信息。

多智能體 PPO（MAPPO）：

基於 Proximal Policy Optimization（PPO）的方法，適用於多智能體場景。

MARL 代碼示例

以下是一個使用 Python 和 TensorFlow 實現的簡單 MARL 示例，其中兩個智能體在一個簡單的環境中競爭獲取獎勵。

import numpy as np
import tensorflow as tf
from tensorflow.keras import layers, models
import random

# 環境參數
GRID_SIZE = 5
NUM_AGENTS = 2
ACTIONS = [0, 1]  # 0: 左, 1: 右
REWARD_POSITION = GRID_SIZE - 1  # 獎勵位置

# 創建 Q 網絡
def create_q_network():
    model = models.Sequential()
    model.add(layers.Dense(24, activation='relu', input_shape=(1,)))
    model.add(layers.Dense(24, activation='relu'))
    model.add(layers.Dense(len(ACTIONS)))  # 動作空間
    return model

# 經驗回放緩衝區
class ReplayBuffer:
    def __init__(self, max_size):
        self.buffer = []

    def add(self, experience):
        self.buffer.append(experience)

    def sample(self, batch_size):
        return random.sample(self.buffer, batch_size)

    def size(self):
        return len(self.buffer)

# 訓練 MARL
def train_marl():
    q_models = [create_q_network() for _ in range(NUM_AGENTS)]
    replay_buffer = ReplayBuffer(10000)
    gamma = 0.99
    episodes = 1000
    batch_size = 32

    for episode in range(episodes):
        states = [0] * NUM_AGENTS  # 初始化狀態
        total_rewards = [0] * NUM_AGENTS

        for step in range(100):
            actions = []
            for i in range(NUM_AGENTS):
                state_input = np.reshape(states[i], [1, 1])
                action_probs = q_models[i].predict(state_input)
                action = np.argmax(action_probs)  # 選擇最大 Q 值的動作
                actions.append(action)

            # 執行動作並觀察下一個狀態和獎勵
            next_states = []
            rewards = []
            done = False
            for i in range(NUM_AGENTS):
                next_state = states[i] + (1 if actions[i] == 1 else -1)
                next_state = np.clip(next_state, 0, GRID_SIZE - 1)
                reward = 1 if next_state == REWARD_POSITION else 0
                next_states.append(next_state)
                rewards.append(reward)
                total_rewards[i] += reward

            # 存儲經驗到緩衝區
            for i in range(NUM_AGENTS):
                replay_buffer.add((states[i], actions[i], rewards[i], next_states[i], done))

            states = next_states

            # 訓練 Q 網絡
            if replay_buffer.size() >= batch_size:
                batch = replay_buffer.sample(batch_size)
                for state, action, reward, next_state, done in batch:
                    state_input = np.reshape(state, [1, 1])
                    next_state_input = np.reshape(next_state, [1, 1])

                    # 更新 Q 網絡
                    target = reward + (gamma * np.max(q_models[0].predict(next_state_input)) * (not done))
                    with tf.GradientTape() as tape:
                        q_values = q_models[0](state_input)
                        loss = tf.keras.losses.MSE(target, q_values[0][action])

                    grads = tape.gradient(loss, q_models[0].trainable_variables)
                    q_models[0].optimizer.apply_gradients(zip(grads, q_models[0].trainable_variables))

        print(f"Episode: {episode}, Total Rewards: {total_rewards}")

if __name__ == "__main__":
    train_marl()

Find More