RL для LLM1 / 13

Модуль 1. Введение: Reinforcement Learning для LLM

Здесь можно прочитать пояснения и код имеющихся ноутбуков или скачать файлы. Код на сайте не запускается.

Скачать файлы курса (.zip) 382,1 КБ

Архив содержит файлы на обоих языках и вспомогательные материалы с сохранением структуры каталогов.

Модуль 1 · Минимальная механика RL

Основной ноутбук модуля

Скачать .ipynb module-01-rl-basics.ru.ipynb · 22,5 КБ

Показать текст ноутбука

Модуль 1 · Минимальная механика RL

Этот ноутбук — практический спутник Модуля 1. Он не пытается воспроизвести большой конвейер RLHF. Его задача проще и важнее: сделать исполняемыми первые объекты курса — политику как распределение, численно устойчивый softmax, энтропию, траекторию, накопленную награду и оценивание методом Монте-Карло.

Профиль: A — CPU, NumPy, Matplotlib.
Воспроизводимость: фиксированные seed и детерминированные вспомогательные функции.
Ожидаемое время: менее одной минуты на обычном CPU.
Связь с лекцией: рисунки §1.4 строятся отдельным генератором модуля; здесь та же механика дана в редактируемой форме.

Ноутбук самодостаточен: знакомство с другими курсами серии полезно, но не требуется. В курсе Modern LLMs ту же идею распределения следующего токена можно увидеть внутри полного Transformer; здесь она рассматривается как политика RL.

0. Подготовка среды

В Colab можно раскомментировать строку установки. В локальной среде с установленными numpy и matplotlib её следует пропустить.

Python

# При необходимости раскомментируйте установку в Colab.
# %pip -q install numpy matplotlib
from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Iterable

import matplotlib.pyplot as plt
import numpy as np

OUTPUT_DIR = Path("outputs/module-01/ru")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

np.set_printoptions(precision=6, suppress=True)
SEED = 7
rng = np.random.default_rng(SEED)

1. Накопленная награда: сумма будущих наград

Для конечной траектории используем

$G_0 = r_1 + \gamma r_2 + \gamma^2 r_3 + \cdots.$

Код намеренно короткий, но проверяет входы. В RL ошибка индекса или незамеченный nan легко маскируется под «нестабильное обучение», хотя причина лежит в самой бухгалтерии траектории.

Python

def discounted_return(rewards: list[float], gamma: float = 1.0) -> float:
    """Вычислить G_0 обратной рекурсией для непустой конечной последовательности."""
    reward_array = np.asarray(rewards, dtype=float)
    if reward_array.ndim != 1 or reward_array.size == 0:
        raise ValueError("rewards must be a non-empty 1D sequence")
    if not np.isfinite(reward_array).all():
        raise ValueError("rewards must be finite")
    if not np.isfinite(gamma) or not 0.0 <= gamma <= 1.0:
        raise ValueError("gamma must be finite and lie in [0, 1]")
    total = 0.0
    for reward in reward_array[::-1]:
        total = float(reward) + gamma * total
    return float(total)


assert discounted_return([0, 0, 1], gamma=1.0) == 1.0
assert np.isclose(discounted_return([1, 1, 1], gamma=0.9), 1 + 0.9 + 0.81)

discounted_return([0, 0, 0, 1], gamma=0.95)

2. Устойчивый softmax и энтропия

Для LLM политика следующего токена задаётся распределением вероятностей. Softmax нужно считать в устойчивой форме: сначала вычесть максимум масштабированных логитов, затем экспоненцировать. Энтропия измеряет ширину распределения, но сама по себе не является мерой качества ответа.

Python

def softmax_with_temperature(logits: np.ndarray, temperature: float = 1.0) -> np.ndarray:
    """Численно устойчивый softmax(logits / temperature)."""
    logits = np.asarray(logits, dtype=float)
    if logits.ndim != 1 or logits.size == 0:
        raise ValueError("logits должен быть непустым одномерным массивом")
    if not np.all(np.isfinite(logits)):
        raise ValueError("все логиты должны быть конечными")
    if not np.isfinite(temperature) or temperature <= 0:
        raise ValueError("temperature должна быть конечной и положительной")
    scaled = logits / float(temperature)
    scaled = scaled - np.max(scaled)
    weights = np.exp(scaled)
    return weights / np.sum(weights)


def categorical_entropy(probs: np.ndarray) -> float:
    """Энтропия категориального распределения в натах."""
    probs = np.asarray(probs, dtype=float)
    if probs.ndim != 1 or probs.size == 0:
        raise ValueError("probs должен быть непустым одномерным массивом")
    if np.any(probs < 0) or not np.all(np.isfinite(probs)):
        raise ValueError("вероятности должны быть конечными и неотрицательными")
    if not np.isclose(probs.sum(), 1.0, atol=1e-8):
        raise ValueError("вероятности должны суммироваться в единицу")
    positive = probs > 0
    return float(-np.sum(probs[positive] * np.log(probs[positive])))


logits = np.array([2.0, 1.0, 0.0, -1.0])
for temperature in [0.25, 0.5, 1.0, 2.0, 4.0]:
    probs = softmax_with_temperature(logits, temperature)
    print(f"T={temperature:>4}: вероятности={probs}, H={categorical_entropy(probs):.4f}")

# Реализация без вычитания максимума переполнилась бы на этом примере.
p = softmax_with_temperature(np.array([1000.0, 999.0, 998.0]), temperature=1.0)
assert np.all(np.isfinite(p)) and np.isclose(p.sum(), 1.0)

3. Температурная кривая

Для фиксированных конечных логитов, если не применять top-k, top-p и другие фильтры, энтропия softmax-политики не убывает с температурой. График ниже — не внешний benchmark, а численная проверка формулы на одном фиксированном векторе логитов.

Python

temperatures = np.geomspace(0.1, 12.0, 400)
entropies = np.array([
    categorical_entropy(softmax_with_temperature(logits, t))
    for t in temperatures
])

fig, ax = plt.subplots(figsize=(8.5, 4.8))
ax.plot(temperatures, entropies, linewidth=2)
ax.axhline(np.log(len(logits)), linestyle="--", linewidth=1.2,
           label="максимальная энтропия ln K")
ax.set_xscale("log")
ax.set_xlabel("температура T")
ax.set_ylabel("энтропия H(pi_T), наты")
ax.set_title("Энтропия чистой softmax-политики при фиксированных логитах")
ax.legend()
fig.tight_layout()
fig.savefig(OUTPUT_DIR / "temperature_entropy.png", dpi=160)
plt.show()

4. Среда-коридор

Теперь построим маленькую среду. Она нужна не «ради игрушки»: в коридоре можно буквально увидеть состояние, действие, терминал, награду, траекторию и правило остановки. В следующих модулях вместо пяти клеток появятся токены, полные ответы, группы кандидатов и среды с инструментами, но бухгалтерия траектории останется той же.

Python

@dataclass(frozen=True)
class CorridorConfig:
    n_states: int = 5
    start: int = 2
    left_terminal: int = 0
    right_terminal: int = 4
    max_steps: int = 20


def corridor_step(
    state: int,
    action: int,
    cfg: CorridorConfig = CorridorConfig(),
) -> tuple[int, float, bool]:
    """Выполнить один переход в пяти-клеточном коридоре."""
    if action not in (0, 1):
        raise ValueError("action должен быть 0 (влево) или 1 (вправо)")
    if state in (cfg.left_terminal, cfg.right_terminal):
        return state, 0.0, True
    next_state = state - 1 if action == 0 else state + 1
    done = next_state in (cfg.left_terminal, cfg.right_terminal)
    reward = 1.0 if next_state == cfg.right_terminal else 0.0
    return next_state, reward, done


def random_policy(_: int, rng: np.random.Generator) -> int:
    """Равновероятно выбрать шаг влево или вправо."""
    return int(rng.integers(0, 2))


def run_episode(
    rng: np.random.Generator,
    cfg: CorridorConfig = CorridorConfig(),
) -> dict:
    """Собрать одну траекторию вместе с правилом остановки и накопленной наградой."""
    state = cfg.start
    states = [state]
    actions: list[int] = []
    rewards: list[float] = []
    done = False
    for _ in range(cfg.max_steps):
        action = random_policy(state, rng)
        next_state, reward, done = corridor_step(state, action, cfg)
        actions.append(action)
        rewards.append(reward)
        states.append(next_state)
        state = next_state
        if done:
            break
    return {
        "states": states,
        "actions": actions,
        "rewards": rewards,
        "done": done,
        "return": discounted_return(rewards, gamma=1.0),
    }


for i in range(5):
    episode = run_episode(np.random.default_rng(SEED + i))
    print(
        f"эпизод {i}: состояния={episode['states']}, "
        f"действия={episode['actions']}, накопленная награда={episode['return']}"
    )

Python

def plot_corridor_episode(
    episode: dict,
    cfg: CorridorConfig = CorridorConfig(),
) -> None:
    """Показать клетки среды и первые переходы одной траектории."""
    fig, ax = plt.subplots(figsize=(8.8, 2.4))
    ax.set_xlim(-0.5, cfg.n_states - 0.5)
    ax.set_ylim(-0.35, 1.2)
    ax.axis("off")

    for state in range(cfg.n_states):
        if state == cfg.left_terminal:
            face, edge, label = "#F8E2D2", "#D55E00", "терминал\n0"
        elif state == cfg.right_terminal:
            face, edge, label = "#FFF0C8", "#E69F00", "терминал\n+1"
        elif state == cfg.start:
            face, edge, label = "#D9EDF8", "#0072B2", "старт\n2"
        else:
            face, edge, label = "#E6F4EF", "#009E73", str(state)
        box = plt.Rectangle(
            (state - 0.42, 0.22), 0.84, 0.52,
            facecolor=face, edgecolor=edge, linewidth=2,
        )
        ax.add_patch(box)
        ax.text(state, 0.48, label, ha="center", va="center", fontsize=11)

    states = episode["states"]
    for t, (start, end) in enumerate(zip(states[:-1], states[1:])):
        y = 0.96 + 0.04 * (t % 2)
        ax.annotate(
            "", xy=(end, y), xytext=(start, y),
            arrowprops={"arrowstyle": "->", "linewidth": 1.6},
        )
    ax.set_title(
        f"пример эпизода: накопленная награда={episode['return']}, "
        f"шагов={len(episode['actions'])}"
    )
    fig.tight_layout()
    fig.savefig(OUTPUT_DIR / "corridor_episode.png", dpi=160)
    plt.show()


plot_corridor_episode(run_episode(np.random.default_rng(SEED)))

5. Точное значение и оценка Монте-Карло

Без ограничения длины вероятность когда-нибудь дойти из центра до правого терминала равна $0.5$ . Но код обрывает эпизод после max_steps=20, поэтому сравнивать эксперимент нужно с конечным протоколом:

$V_{20}^{\pi}(2)=\Pr(T_4\le 20)=\frac{1023}{2048}.$

Метод Монте-Карло оценивает это ожидание средним по независимым эпизодам. Отдельная оценка может оказаться выше или ниже точного значения; её шум убывает в масштабе $1/\sqrt{N}$ .

Python

def exact_corridor_value(max_steps: int = 20) -> float:
    """Точно вычислить вероятность достичь правого терминала за заданный горизонт."""
    if not isinstance(max_steps, int) or isinstance(max_steps, bool) or max_steps <= 0:
        raise ValueError("max_steps должен быть положительным целым числом")
    values = np.zeros(5, dtype=float)
    values[4] = 1.0
    for _ in range(max_steps):
        updated = values.copy()
        for state in (1, 2, 3):
            updated[state] = 0.5 * values[state - 1] + 0.5 * values[state + 1]
        updated[0] = 0.0
        updated[4] = 1.0
        values = updated
    return float(values[2])


def monte_carlo_value(
    n_episodes: int,
    seed: int = SEED,
    cfg: CorridorConfig = CorridorConfig(),
) -> tuple[float, np.ndarray]:
    """Оценить ценность средним накопленных наград по независимым эпизодам."""
    if not isinstance(n_episodes, int) or isinstance(n_episodes, bool) or n_episodes <= 0:
        raise ValueError("n_episodes должен быть положительным целым числом")
    local_rng = np.random.default_rng(seed)
    returns = np.array(
        [run_episode(local_rng, cfg)["return"] for _ in range(n_episodes)],
        dtype=float,
    )
    return float(returns.mean()), returns


estimate_1000, returns_1000 = monte_carlo_value(1000, seed=SEED)
exact = exact_corridor_value(20)
print(f"Оценка Монте-Карло при N=1000: {estimate_1000:.3f}")
print(f"Точная ценность при горизонте 20: {exact:.12f}")
print(f"Приближённая стандартная ошибка: {np.sqrt(exact * (1 - exact) / 1000):.4f}")

# The Monte Carlo estimate changes with SEED; test its domain, not one draw.
assert np.isfinite(estimate_1000) and 0.0 <= estimate_1000 <= 1.0
assert np.isclose(exact, 1023 / 2048)

Python

n = 5000
_, returns = monte_carlo_value(n, seed=SEED)
ns = np.arange(1, n + 1)
running = np.cumsum(returns) / ns
exact = exact_corridor_value(20)
standard_error = np.sqrt(exact * (1 - exact) / ns)

fig, ax = plt.subplots(figsize=(8.8, 4.8))
ax.fill_between(
    ns,
    exact - 2 * standard_error,
    exact + 2 * standard_error,
    alpha=0.35,
    label="приближённый интервал ±2 стандартные ошибки",
)
ax.plot(ns, running, linewidth=1.5, label="текущая оценка Монте-Карло")
ax.axhline(exact, linestyle="--", linewidth=1.8,
           label="точная ценность при горизонте 20")
ax.set_xscale("log")
ax.set_ylim(0.42, 0.58)
ax.set_xlabel("число эпизодов N")
ax.set_ylabel("оценка ценности")
ax.set_title("Оценивание случайной политики в коридоре методом Монте-Карло")
ax.legend()
fig.tight_layout()
fig.savefig(OUTPUT_DIR / "corridor_mc_convergence.png", dpi=160)
plt.show()

6. Ключевые выводы практикума

Политика — распределение вероятностей, а не одно выбранное действие.
Softmax следует считать численно устойчиво.
Энтропия описывает разброс вероятностей, но не заменяет оценку качества.
У траектории есть точная бухгалтерия: состояния, действия, награды, правило остановки и накопленная награда.
Оценка Монте-Карло случайна; сравнивать её нужно с точной версией протокола, которая действительно реализована.
Код и рисунки используют фиксированный seed=7; это обеспечивает воспроизводимость конкретного прогона, но не устраняет статистическую случайность самого метода.

В Модуле 2 коридор станет маленьким MDP с уравнениями Беллмана. В Модулях 6–9 те же идеи вернутся на уровне токеновых log-probabilities, KL, DPO, PPO, GRPO и RLVR.