A3C - Asynchronous Advantage Actor-Critic

여러 에이전트를 비동기로 돌리는 Actor-Critic

A3C(Asynchronous Advantage Actor-Critic)는 여러 개의 에이전트(worker)가 각자 별도의 환경에서 동시에 경험을 수집하고, 그 결과를 하나의 전역 신경망(global network) 에 비동기적으로 반영하며 학습하는 강화학습 알고리즘이다.
Actor는 어떤 행동을 할지 확률적으로 결정하는 정책을 학습하고, Critic은 현재 상태가 얼마나 좋은지 나타내는 가치 함수를 학습한다. 그리고 Advantage는 “실제로 얻은 결과가 현재 가치 예측보다 얼마나 더 좋거나 나빴는가”를 나타내어, 정책을 더 안정적으로 업데이트하도록 돕는 역할을 한다.

A3C의 핵심은 다음 세 가지 아이디어가 결합된 구조라는 점이다.

• Actor-Critic 구조
• Advantage 기반 정책 업데이트
• 비동기 병렬 학습

여러 worker가 병렬로 다양한 경험을 빠르게 모으기 때문에 데이터 효율과 탐험 성능이 좋아지고, 서로 다른 경로의 경험이 섞이면서 학습이 더 안정적으로 이루어진다. 이후에는 이 구조를 동기식으로 단순화한 A2C가 등장해, PPO 등 최신 알고리즘의 기반이 되었다.
([논문])

---

1. Actor-Critic와 Advantage 다시 정리

1-1. Actor-Critic 구조

• Actor: 상태 s를 보고 어떤 행동 a를 할지 확률적으로 결정하는 정책 π(a|s)를 학습한다.
• Critic: 상태 s가 얼마나 좋은지 나타내는 가치 함수 V(s) 또는 Q(s, a)를 학습한다.

즉, Actor는 “무엇을 할까?”를 담당하고, Critic은 “방금 행동이 얼마나 괜찮았는가?”를 평가해주는 구조다.
정책 기반(Policy-Based)의 유연함과 가치 기반(Value-Based)의 평가 능력을 동시에 가져가는 셈이다.

1-2. Advantage란 무엇인가?

Advantage는 “현재 행동이 예상보다 얼마나 더 좋았는가” 를 나타내는 값이다.

• Advantage: A(s, a) = Target − V(s)
• Target(1-step TD 타깃 예시): Target = r + γ · V(s')

즉,

• r + γ · V(s') : 실제로 한 번 움직여 보고 얻은 “새로운 평가”
• V(s) : Critic이 원래 생각하던 “현재 상태의 가치”

따라서,

• A(s, a) > 0 이면 → 예상보다 좋았으니, 그 행동의 확률을 늘려야 한다.
• A(s, a) < 0 이면 → 예상보다 나빴으니, 그 행동의 확률을 줄여야 한다.

Actor는 이 Advantage를 이용해 좋은 행동의 확률은 증가, 나쁜 행동의 확률은 감소하도록 정책을 업데이트한다.

---

2. A3C가 등장한 이유

2-1. 문제 1 – 데이터 상관성

초기 딥 RL(DQN 등)은 보통 환경 하나 + 에이전트 하나 구조를 썼다. 이 경우 에이전트가 한 환경에서만 학습하기 때문에, 시간에 따라 얻는 경험들이 서로 강하게 상관 되어 있다.
DQN은 이를 완화하기 위해 Experience Replay로 샘플을 섞어 쓰지만, 어차피 하나의 환경 궤적에서 나온 데이터라는 한계가 있다.

2-2. 문제 2 – 경험 수집 속도

환경이 하나뿐이면 경험 수집 자체가 느리다.
특히 시뮬레이션이 무거운 환경이나 CPU 기반 환경에서는, “경험을 모으는 속도”가 병목 이 되기 쉽다.

---

3. A3C의 핵심 아이디어

3-1. 여러 worker를 동시에 돌리기

아이디어 자체는 단순하다.

환경을 여러 개 띄우고, 에이전트도 여러 개 띄운 다음,
각자 돌아가며 경험을 모아 공유된 하나의 글로벌 모델을 업데이트하자.

구조를 텍스트로 그리면 다음과 같다.

Worker 1 -> 환경 1
Worker 2 -> 환경 2
Worker 3 -> 환경 3
Worker 4 -> 환경 4
...

• 각 worker는 독립적인 환경에서 탐험한다.
• 동시에 여러 개의 trajectory를 수집하므로 경험 수집 속도가 빨라지고, 데이터 다양성도 올라간다.
• 서로 다른 경로에서 온 경험이 섞이면서, 학습이 더 안정적으로 진행되는 효과도 있다.

3-2. 비동기 업데이트

A3C의 포인트는 여러 worker가 서로를 기다리지 않고(asynchronous) 업데이트를 한다는 점이다.

• 각 worker는 글로벌 네트워크를 복사해 local model을 만든다.
• 환경에서 일정 step(또는 n-step) 동안 데이터를 모은다.
• 이 데이터를 기반으로 local model에서 그라디언트를 계산한다.
• 계산된 그라디언트를 전역 신경망(global network) 에 바로 반영해 파라미터를 업데이트한다.
• 업데이트된 글로벌 파라미터를 다시 local model에 복사해서 동기화한다.

이 과정을 각 worker가 동시에, 비동기로 수행하기 때문에 A3C라는 이름이 붙었다.

---

4. A3C 전체 구조 한눈에 보기

Global Network (Actor + Critic)
                ↑
                │
 ┌──────────────┼──────────────┐
 │              │              │
Worker 1      Worker 2      Worker 3
환경 탐험       환경 탐험       환경 탐험
데이터 수집     데이터 수집     데이터 수집
gradient 계산  gradient 계산  gradient 계산
 │              │              │
 └──────────────┴──────────────┘
        글로벌 모델 업데이트

각 worker는 다음 순서를 반복한다.

1. 글로벌 모델 복사 → local model 생성
2. 환경에서 일정 step 동안 roll-out
   • 상태, 행동, 보상, 다음 상태를 모은다.
   • 아래 값을 계산한다.
   • target = r + γ · V(s')
   • advantage = target − V(s)
3. local model 기준으로 gradient 계산
   • Actor Loss: log π(a|s) × Advantage
   • Critic Loss: (V(s) − target)²
4. 글로벌 모델 업데이트
   • local gradient를 글로벌 파라미터에 복사해 업데이트
5. 업데이트된 글로벌 모델 → 다시 local model에 복사
6. 이 과정을 에피소드가 끝날 때까지 반복

---

5. A3C CartPole 구현 코드 분석

아래 코드는 CartPole-v1 환경에서 A3C 구조를 멀티프로세싱 으로 구현한 예제이다.
핵심은:

• 전역 모델(global_model) 을 share_memory()로 모든 프로세스가 공유
• 각 worker 프로세스는 local 모델을 만든 뒤, global 모델을 업데이트
• 별도의 테스트 프로세스는 같은 글로벌 모델을 읽어서 성능을 모니터링

import gymnasium as gym
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.distributions import Categorical
import torch.multiprocessing as mp
import time

# Hyperparameters
n_train_processes = 3       # 학습용 프로세스를 3개 동작
learning_rate = 0.0002
update_interval = 5         # 한 번 업데이트 전에 최대 몇 step까지 데이터를 모을지
gamma = 0.98
max_train_ep = 300
max_test_ep = 100


class ActorCritic(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(4, 256)
        self.fc_pi = nn.Linear(256, 2)
        self.fc_v = nn.Linear(256, 1)

    def pi(self, x, softmax_dim=0):
        x = F.relu(self.fc1(x))
        x = self.fc_pi(x)
        return F.softmax(x, dim=softmax_dim)

    def v(self, x):
        x = F.relu(self.fc1(x))
        return self.fc_v(x)

• CartPole-v1의 상태 차원이 4이므로 nn.Linear(4, 256)으로 시작한다.
• fc_pi는 행동 2개(왼쪽, 오른쪽)에 대한 확률을 출력하고,
• fc_v는 해당 상태의 가치 (V(s)) 하나를 출력한다.

def train(global_model, rank):
    local_model = ActorCritic()
    # global_model의 현재 가중치를 local_model에 복사
    local_model.load_state_dict(global_model.state_dict())

    # 실제 업데이트 대상은 global_model
    optimizer = optim.Adam(global_model.parameters(), lr=learning_rate)
    env = gym.make("CartPole-v1")

    for n_epi in range(max_train_ep):
        done = False
        s, _ = env.reset()

        while not done:
            s_lst, a_lst, r_lst = [], [], []

            for _ in range(update_interval):
                prob = local_model.pi(torch.from_numpy(s).float())
                m = Categorical(prob)
                a = m.sample().item()

                s_prime, r, terminated, truncated, info = env.step(a)
                done = terminated or truncated

                s_lst.append(s)
                a_lst.append([a])
                r_lst.append(r / 100.0)

                s = s_prime
                if done:
                    break

• update_interval 동안 rollout을 수행하며 (s, a, r)를 저장한다.
• 보상은 r / 100.0 으로 스케일을 줄여 학습을 좀 더 안정적으로 만든다.

            s_final = torch.tensor(s_prime, dtype=torch.float)
            # 에피소드가 끝났으면 미래 가치가 없음
            R = 0.0 if done else local_model.v(s_final).item()

            td_target_lst = []
            # 보상 리스트를 뒤에서부터 순회하며 n-step Return 계산
            for reward in r_lst[::-1]:
                R = gamma * R + reward
                td_target_lst.append([R])
            td_target_lst.reverse()

            s_batch = torch.tensor(s_lst, dtype=torch.float)
            a_batch = torch.tensor(a_lst)
            td_target = torch.tensor(td_target_lst, dtype=torch.float)
            # Advantage = TD target - V(s)
            advantage = td_target - local_model.v(s_batch)

            pi = local_model.pi(s_batch, softmax_dim=1)  # 예: (5, 2)
            pi_a = pi.gather(1, a_batch)

            value_loss = F.smooth_l1_loss(local_model.v(s_batch), td_target.detach())
            policy_loss = -torch.log(pi_a) * advantage.detach()
            loss = policy_loss + value_loss

            optimizer.zero_grad()
            loss.mean().backward()

            for global_param, local_param in zip(global_model.parameters(), local_model.parameters()):
                global_param._grad = local_param.grad

            # 글로벌 모델 업데이트
            optimizer.step()
            # 최신 global_model 가중치를 다시 local_model에 복사
            local_model.load_state_dict(global_model.state_dict())

        if n_epi % 10 == 0:
            print(f"[TRAIN {rank}] episode={n_epi}", flush=True)

    env.close()
    print(f"Training process {rank} finished.", flush=True)

• Critic Loss: value_loss = smooth_l1_loss(V(s), target)
• Actor Loss: policy_loss = -log π(a|s) * advantage
• local 모델에서 backward를 돌리지만, optimizer는 global_model 을 대상으로 한다.
• global_param._grad = local_param.grad 로 그라디언트를 전역 파라미터에 복사한 뒤 optimizer.step()으로 업데이트한다.
• 이후 전역 파라미터를 다시 local 모델에 복사해 최신 상태로 맞춘다.

---

6. 테스트 프로세스와 메인 루프

def test(global_model):
    env = gym.make("CartPole-v1")
    score = 0.0
    print_interval = 5

    for n_epi in range(max_test_ep):
        done = False
        s, _ = env.reset()

        while not done:
            prob = global_model.pi(torch.from_numpy(s).float())
            a = Categorical(prob).sample().item()

            s_prime, r, terminated, truncated, info = env.step(a)
            done = terminated or truncated

            s = s_prime
            score += r

        if n_epi % print_interval == 0 and n_epi != 0:
            print(f"[TEST] episode={n_epi}, avg score={score / print_interval:.1f}", flush=True)
            score = 0.0
            time.sleep(0.5)

    env.close()

• 학습과는 달리, 테스트는 하나의 프로세스에서 공유된 global 모델을 읽기만 한다.
• 별도 업데이트 없이 현재 성능을 주기적으로 출력한다.

if __name__ == "__main__":
    mp.set_start_method("spawn", force=True)

    global_model = ActorCritic()
    # 전역 모델을 프로세스 간 공유 메모리에 올림
    global_model.share_memory()

    processes = []
    # 학습 프로세스 3개, 테스트 프로세스 1개
    for rank in range(n_train_processes + 1):
        if rank == 0:
            p = mp.Process(target=test, args=(global_model,))
        else:
            p = mp.Process(target=train, args=(global_model, rank))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()
        print("process exitcode:", p.exitcode, flush=True)

• share_memory() 덕분에 모든 프로세스가 같은 global_model 파라미터를 직접 공유한다.
• rank == 0 은 테스트, 그 외는 학습 worker로 동작한다.

---

7. A3C 정리

• 핵심 아이디어
  • Actor-Critic 구조 + Advantage + 멀티프로세싱 비동기 학습
• 장점
  • 여러 환경에서 병렬로 경험을 모아 데이터 효율과 탐험 성능을 끌어올린다.
  • 서로 다른 trajectory가 섞이며 학습이 더 안정적이 된다.
  • CPU 여러 코어를 적극 활용할 수 있다.
• 단점 / 한계
  • 비동기 업데이트 특성상 gradient 충돌(gradient interference) 문제가 생길 수 있다.
  • 구현 난이도가 동기식 A2C에 비해 높고, GPU batch 학습과 궁합도 좋지 않다.

이 한계를 보완한 것이 바로 A2C(Advantage Actor-Critic) 이고, PPO 등 실전에서 자주 쓰이는 알고리즘은 대부분 이 A2C 스타일의 동기식 Advantage Actor-Critic 을 기반으로 발전해 나갔다.