강화학습¶

"""관측은 상태가 아니다: 근거리 레이더를 단 로봇 청소기."""

import numpy as np
from collections import deque

H, W = 11, 21          # 방의 크기 (세로 11칸, 가로 21칸)
RADAR = 2              # 로봇은 자기 주변 2칸까지만 볼 수 있다
MAX_STEPS = 4000       # 이 걸음 수를 넘으면 실패로 친다
MOVES = {"up": (-1, 0), "down": (1, 0), "left": (0, -1), "right": (0, 1)}
STEP_OF = {v: k for k, v in MOVES.items()}      # 변위 -> 행동 이름 (역방향 표)


# === 문 하나로 이어진 두 개의 방 ===
def make_room():
    """가장자리는 벽, 가운데 칸막이에 문이 하나 뚫린 방을 만든다."""
    wall = np.zeros((H, W), dtype=bool)
    wall[0, :] = wall[-1, :] = wall[:, 0] = wall[:, -1] = True   # 바깥 테두리
    wall[:, 10] = True                       # 가운데 칸막이
    wall[5, 10] = False                      # 문 한 칸
    # 이 구조가 중요하다. 문을 찾아 통과해야만 오른쪽 방을 청소할 수 있다.
    return wall


def visible(pos):
    """현재 위치에서 레이더에 잡히는 칸들의 목록. 이것이 관측 o_t 다."""
    r, c = pos
    return [(i, j) for i in range(r - RADAR, r + RADAR + 1)
                   for j in range(c - RADAR, c + RADAR + 1)
                   if 0 <= i < H and 0 <= j < W]


# === 에이전트 A: 상태 = 지금 이 순간의 레이더 화면. 기억이 없다 ===
def agent_observation(pos, dirty, known, known_dirty, rng):
    """s_t = o_t 로 두는 정책. 보이는 것에만 반응한다."""
    # 레이더 안에 더러운 칸이 있는가?
    seen = [cell for cell in visible(pos) if dirty[cell]]
    if seen:
        # 있으면 그중 가장 가까운 칸(맨해튼 거리) 쪽으로 한 걸음 간다
        tr, tc = min(seen, key=lambda t: abs(t[0] - pos[0]) + abs(t[1] - pos[1]))
        options = []
        if tr < pos[0]: options.append("up")
        if tr > pos[0]: options.append("down")
        if tc < pos[1]: options.append("left")
        if tc > pos[1]: options.append("right")
        if options:
            return options[rng.integers(len(options))]
    # 보이는 범위에 더러운 칸이 없으면 무작위로 헤맨다.
    # 어디를 이미 청소했는지 기억하지 못하므로 이것 말고는 할 수 있는 게 없다.
    return list(MOVES)[rng.integers(4)]


# === 에이전트 B: 상태 = 지금까지 쌓아 올린 지도 ===
def agent_map(pos, dirty, known, known_dirty, rng):
    """더럽다고 아는 칸 또는 아직 안 가 본 칸 중 가장 가까운 곳으로 간다.

    너비우선탐색(BFS)으로 최단 경로를 찾는다.
    핵심은 탐색이 실제 방이 아니라 **로봇이 가진 지도(known)** 위에서 이뤄진다는 것이다.
    즉 이 정책은 자기가 아는 만큼만 계획할 수 있다.
    """
    parent, queue, goal = {pos: None}, deque([pos]), None
    while queue:
        cur = queue.popleft()
        # 목표 조건: 더러운 것으로 기록된 칸이거나(known_dirty),
        #            아직 한 번도 관측하지 못한 칸이다(known == 0).
        # 둘째 조건이 탐험을 만들어 낸다. 미지의 영역이 곧 목표가 되기 때문이다.
        if cur != pos and (known_dirty[cur] or known[cur] == 0):
            goal = cur
            break
        for dr, dc in MOVES.values():
            nxt = (cur[0] + dr, cur[1] + dc)
            if (0 <= nxt[0] < H and 0 <= nxt[1] < W
                    and nxt not in parent and known[nxt] != 2):   # 벽으로 아는 칸은 지나가지 않는다
                parent[nxt] = cur
                queue.append(nxt)
    if goal is None:
        return list(MOVES)[rng.integers(4)]     # 갈 곳이 없으면(있을 수 없지만) 무작위

    # 찾은 목표에서 부모를 거슬러 올라가 "첫 걸음"이 무엇이었는지 알아낸다
    cur = goal
    while parent[cur] != pos:
        cur = parent[cur]
    return STEP_OF[(cur[0] - pos[0], cur[1] - pos[1])]


# === 한 번의 에피소드: 방을 다 치우거나 시간이 다할 때까지 ===
def run(agent, seed):
    rng = np.random.default_rng(seed)
    wall = make_room()
    dirty = ~wall.copy()          # 벽이 아닌 모든 칸이 처음엔 더럽다
    total = dirty.sum()
    pos = (1, 1)                  # 왼쪽 위 구석에서 출발

    # 로봇이 **자기 머릿속에** 들고 있는 지도. 이것이 상태 x_t 다.
    #   known:       0 = 모름, 1 = 빈 칸, 2 = 벽
    #   known_dirty: 더럽다고 기록해 둔 칸
    # 처음에는 전부 0, 즉 아무것도 모르는 채로 시작한다.
    known = np.zeros((H, W), np.int8)
    known_dirty = np.zeros((H, W), bool)

    for step in range(1, MAX_STEPS + 1):
        dirty[pos] = False                                   # 발밑을 청소한다

        # 상태 갱신 x_{t+1} = f(x_t, a_t, o_{t+1}).
        # 새 관측이 들어오면 지도를 덮어쓴다. 이 한 줄이 A와 B를 가르는 전부다.
        for cell in visible(pos):                            # o_t 가 도착
            known[cell] = 2 if wall[cell] else 1
            known_dirty[cell] = dirty[cell]

        if not dirty.any():
            return step, 1.0                                 # 방을 다 치웠다

        # 정책이 행동 a_t 를 고르고, 벽이 아니면 이동한다
        dr, dc = MOVES[agent(pos, dirty, known, known_dirty, rng)]
        nxt = (pos[0] + dr, pos[1] + dc)
        if not wall[nxt]:
            pos = nxt

    # 시간이 다했다. 얼마나 치웠는지(청소율)를 함께 돌려준다.
    return MAX_STEPS, 1 - dirty.sum() / total


for label, agent in [("s = o  (radar only) ", agent_observation),
                     ("x = accumulated map ", agent_map)]:
    results = [run(agent, seed) for seed in range(30)]
    steps = np.array([r[0] for r in results])
    coverage = np.array([r[1] for r in results])
    finished = coverage >= 1.0
    median = np.median(steps[finished]) if finished.any() else float("nan")
    print(f"{label}: finished {finished.mean():4.0%} of runs | "
          f"median steps {median:6.0f} | mean coverage {coverage.mean():6.1%}")

# --- (1) 의 하한을 확인한다 ---
wall = make_room()
n_free = int((~wall).sum())
print(f"치울 칸 {n_free}개 (왼쪽 {int((~wall[1:10, 1:10]).sum())}, "
      f"오른쪽 {int((~wall[1:10, 11:20]).sum())}, 문 1),  걸음 수 하한 {n_free - 1}")

s = o  (radar only) : finished  27% of runs | median steps    342 | mean coverage  83.8%
x = accumulated map : finished 100% of runs | median steps    171 | mean coverage 100.0%
치울 칸 163개 (왼쪽 81, 오른쪽 81, 문 1),  걸음 수 하한 162

"""엡실론-탐욕 전략으로 다중 슬롯머신을 1000번 당겨 본다."""
import numpy as np

np.random.seed(42)

# 팔 5개. 참 평균은 우리만 알고 에이전트는 모른다.
true_means = [1.0, 1.5, 2.0, 1.2, 0.8]
n_arms = len(true_means)
n_steps = 1000
epsilon = 0.1  # 10%는 무작위로 탐색하고, 90%는 지금까지 가장 나은 팔을 쓴다

Q = np.zeros(n_arms)       # 각 팔의 가치 추정값
N = np.zeros(n_arms)       # 각 팔을 당긴 횟수
rewards = []

for t in range(n_steps):
    if np.random.rand() < epsilon:
        action = np.random.randint(n_arms)  # 탐색: 아무 팔이나
    else:
        action = np.argmax(Q)               # 활용: 추정값이 가장 큰 팔

    reward = np.random.normal(true_means[action], 1.0)
    N[action] += 1
    # 지난 보상을 다 들고 있다가 평균을 다시 내지 않고 한 걸음씩 고쳐 나간다.
    # Q + (보상 - Q)/N 은 지금까지 받은 보상의 평균과 정확히 같다.
    Q[action] += (reward - Q[action]) / N[action]
    rewards.append(reward)

print("Estimated values:", np.round(Q, 2))
print("True means:      ", true_means)
print(f"Average reward:   {np.mean(rewards):.2f}")
print(f"Best arm chosen:  {np.argmax(N)} (pulled {int(N[np.argmax(N)])} times)")

# --- (1) 의 이론값과 맞추어 본다 ---
p_best = (1 - epsilon) + epsilon / n_arms
ev = (1 - epsilon) * max(true_means) + epsilon / n_arms * sum(true_means)
sd_count = np.sqrt(n_steps * p_best * (1 - p_best))
print(f"\n이론 최적 팔 선택확률 = {p_best:.2f},  기대 횟수 {p_best * n_steps:.0f} "
      f"(SD {sd_count:.1f}),  실제 {int(N[2])}")
print(f"이론 한 번당 기대 보상 = {ev:.3f},  한 걸음당 후회 = {max(true_means) - ev:.2f}, "
      f"1000걸음 누적 후회 = {(max(true_means) - ev) * n_steps:.0f}")
print(f"각 팔을 당긴 횟수: {N.astype(int)}")
print(f"Q 의 표준오차 1/sqrt(N): {np.round(1 / np.sqrt(N), 3)}")
print(f"z = (Q - 참값)/SE:      {np.round((Q - np.array(true_means)) * np.sqrt(N), 2)}")

Estimated values: [0.83 1.41 2.04 1.14 0.61]
True means:       [1.0, 1.5, 2.0, 1.2, 0.8]
Average reward:   1.93
Best arm chosen:  2 (pulled 903 times)

이론 최적 팔 선택확률 = 0.92,  기대 횟수 920 (SD 8.6),  실제 903
이론 한 번당 기대 보상 = 1.930,  한 걸음당 후회 = 0.07, 1000걸음 누적 후회 = 70
각 팔을 당긴 횟수: [ 35  20 903  21  21]
Q 의 표준오차 1/sqrt(N): [0.169 0.224 0.033 0.218 0.218]
z = (Q - 참값)/SE:      [-1.01 -0.39  1.16 -0.26 -0.86]

import numpy as np

rng = np.random.default_rng(0)
K, T, runs = 10, 2000, 500

def simulate(policy):
    total = np.zeros(runs)
    for r in range(runs):
        mu = rng.normal(0, 1, K)          # 각 팔의 참 평균 보상
        best = mu.max()
        Q = np.zeros(K)                   # 추정 가치
        N = np.zeros(K)                   # 당긴 횟수
        regret = 0.0
        for t in range(1, T + 1):
            if policy == "greedy":
                a = int(Q.argmax())
            elif policy == "eps":
                a = int(rng.integers(K)) if rng.random() < 0.1 else int(Q.argmax())
            elif policy == "ucb":
                a = (int(np.argmin(N)) if N.min() == 0 else
                     int(np.argmax(Q + 2 * np.sqrt(np.log(t) / N))))
            else:                          # 톰프슨 표집 (가우시안)
                a = int(np.argmax(rng.normal(Q, 1 / np.sqrt(np.maximum(N, 1)))))

            reward = rng.normal(mu[a], 1)
            N[a] += 1
            Q[a] += (reward - Q[a]) / N[a]     # 점진적 평균 갱신
            regret += best - mu[a]
        total[r] = regret
    return total.mean(), 1.96 * total.std() / np.sqrt(runs)

for key, label in [("greedy", "순수 활용"), ("eps", "엡실론-탐욕 (0.1)"),
                   ("ucb", "UCB"), ("thompson", "톰프슨 표집")]:
    m, ci = simulate(key)
    print(f"{label:>18}: T=2000 누적 후회 {m:>8.1f} ± {ci:.1f}")

             순수 활용: T=2000 누적 후회   1055.5 ± 105.9
      엡실론-탐욕 (0.1): T=2000 누적 후회    394.9 ± 14.4
               UCB: T=2000 누적 후회    180.7 ± 4.2
            톰프슨 표집: T=2000 누적 후회     86.9 ± 3.0