Модуль 12. Инфраструктура RL для LLM: память, роллауты и асинхронность

Четыре основные роли:

1. тренер вычисляет функцию потерь, градиенты и шаг оптимизатора;

2. движок генерации порождает действия политики и обслуживает кэш ключей и значений (key–value cache, KV-кэш);

3. проверяющая система или среда вычисляет исход и служебный статус;

4. оркестратор выбирает запросы, формирует группы, управляет очередями, версиями и восстановлением.

Они не обязаны быть отдельными процессами. На одном графическом процессоре (graphics processing unit, GPU) один процесс может по очереди выполнять все роли; на кластере одна роль может быть распределена между десятками работников.

Инвариант: каждая обучающая запись однозначно связывает действия с фактической политикой поведения и награду — с версией проверяющей системы или среды.

Строка и одно число не сохраняют токенизацию, маску токенов политики, логарифмы вероятностей поведения, версию весов, параметры декодирования, причину остановки и происхождение награды. Без них невозможно надёжно восстановить функцию потерь и причину сбоя.

1. Размещение раздельное: генерация работает на отдельном сервере.

2. Логический режим синхронный: между пакетами стоит строгий барьер.

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

4. Совместное размещение с буферизацией: один GPU по очереди генерирует и обучается, пока центральный процессор (central processing unit, CPU) готовит следующий набор запросов. Раздельное размещение со строгим барьером: отдельный пул vLLM генерирует пакет, затем ждёт завершения обучения и загрузки новой версии.

Распространённый 16-байтовый профиль может состоять из:

• 2 байт bfloat16-параметра;
• 2 байт bfloat16-градиента;
• 8 байт двух float32-моментов Adam;
• 4 байт отдельной float32-мастер-копии.

Профиль меняется из-за точности градиентов, наличия мастер-копии, вида оптимизатора, квантования состояний, выгрузки на CPU, обучения только адаптеров, шардирования и особенностей реализации.

Деление на число участников полностью шардированного параллелизма данных (Fully Sharded Data Parallel, FSDP) не учитывает реплицируемые объекты, временную сборку модулей, коммуникационные буферы, активации, логиты, фрагментацию и момент синхронизации.

Низкоранговая адаптация (Low-Rank Adaptation, LoRA) уменьшает число параметров с градиентами и состояниями оптимизатора. Она не устраняет базовые веса, активации прямого прохода, KV-кэш, временные буферы и необходимость доставить согласованную версию адаптеров движку генерации.

1. Для групповой относительной оптимизации политики (Group Relative Policy Optimization, GRPO) в идеальной формуле память вырастет в четыре раза: число одновременно активных последовательностей увеличилось с 8 до 32.

2. Реальный рост зависит от завершившихся последовательностей, распределения длин, страниц распределителя памяти, тензорного параллелизма, резервов и повторного использования префикса.

3. PagedAttention уменьшает фрагментацию управляемого KV-кэша; повторное использование префиксов или RadixAttention переиспользует общую запросную часть.

4. Ключи и значения зависят от параметров модели. После обновления весов старый кэш представляет скрытые состояния старой политики и не является кэшем новой версии.

В журнале одна версия приписана ответу, который фактически породила смесь двух наборов весов. Поэтому знаменатель отношения вероятностей и воспроизведение политики поведения неоднозначны.

Безопасные режимы:

• завершить уже начатые ответы версией 120 и переключать только новые запросы;
• хранить версию и логарифмы вероятностей каждого действия или хода, если смена между ходами является частью протокола.

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

Температура, top-p и маски логитов меняют распределение после весов. Поэтому одинаковый трансформер не означает одинаковую политику поведения.

Первая проблема — рассогласование обучения и генерации (training–inference mismatch, TIM): версия одна, численные вероятности различаются.

Вторая — задержка политики (policy lag): данные породила более старая версия.

Контроль: сразу после синхронизации, при нулевой задержке, прогнать одни и те же токены через движок и тренер и сравнить логарифмы вероятностей токен за токеном. Затем отдельно изменить возраст версии.

Эффективный размер выборки (effective sample size, ESS) показывает концентрацию весов: равные веса дают полный размер выборки, доминирующий вес уменьшает его. Но высокий ESS не доказывает правильность знаменателя, отсутствие смещения из-за клиппирования, корректность масок или валидность выбранного оценивателя для данных другой политики.

1. Нарушено условие $\lambda<\mu$: скорость поступления 14 млн токенов/мин выше скорости потребления 10 млн.

2. Обратное давление ограничивает генераторы, когда очередь достигает порога: уменьшает выдачу задач, приостанавливает работников или меняет приоритет.

3. Две записи могут отличаться по длине в сотни раз. Число записей плохо описывает память, вычислительную стоимость и возраст данных; токены или оценочная стоимость ближе к физической нагрузке.

4. Длинные эпизоды могут быть систематически сложнее. Их отбрасывание меняет распределение задач и оставляет тренеру преимущественно простые примеры.

Это новый запуск с инициализацией из старых весов, а не продолжение прежнего процесса оптимизации. Сброшены моменты Adam, планировщик и контракт награды.

Семантически полная контрольная точка обычно включает:

• веса модели и адаптеров;
• состояние оптимизатора;
• состояние планировщика скорости обучения;
• глобальный шаг и версию политики;
• состояния генераторов случайных чисел;
• положение в наборе данных;
• состояние учебной программы или планировщика запросов;
• параметры нормировки наград и преимуществ;
• версии токенизатора и шаблона диалога;
• версию проверяющей системы и среды;
• конфигурацию декодирования;
• схему роллаута;
• распределённую раскладку и рецепт восстановления очереди.

Проверка восстановления: сохранить состояние, принудительно остановить процессы, восстановить систему, выполнить контрольный пакет и сравнить версии, токенизацию, конечность потерь и статистику с допустимым диапазоном.

Библиотеки и системы оптимизируют разные уровни: готовый тренер, оркестрацию ролей, полностью асинхронный поток, агентные среды или конкретную систему распределённого обучения. Сравнение без модели, масштаба, синхронности, типа среды, требований к отказоустойчивости и навыков команды не имеет однозначного смысла.

На обучаемый параметр приходится

Полное обучение:

Базовые bfloat16-веса:

Для LoRA базовые веса занимают 14 GB. На каждый из 20 млн обучаемых параметров учитываются сам вес адаптера, градиент, два момента и мастер-копия:

Дополнительная память:

Итого:

Пик дополнительно содержит активации, временные буферы, логиты, KV-кэш, фрагментацию, реплики и фазу передачи весов.

На один токен:

Поскольку один кибибайт (KiB) равен $1024$ байтам,

Число токенов:

Память:

Здесь GiB — гибибайт, равный $2^{30}$ байтам. При идеальном равномерном шардировании по четырём участникам:

Это полезная нагрузка без страниц, таблиц блоков, выравнивания и временных тензоров.

Один последовательный пакет:

Период после заполнения:

Двадцать последовательных пакетов:

Идеальный конвейер:

Ускорение по полному времени:

Предельное ускорение по периоду:

Этапы несбалансированы: 90 секунд генерации нельзя скрыть десятисекундным обучением. Почти весь период определяется генерацией.

Суммарная работа:

Группа завершается через

Ёмкость:

Утилизация:

Простой:

После ограничения последнего роллаута 8 секундами работа равна 43 GPU-с, время группы — 8 с, ёмкость — 64 GPU-с:

Утилизация выросла, но если длинные ответы соответствуют трудным задачам, правило усечения изменяет распределение данных и наблюдаемую награду.

Веса:

Сумма равна 13, сумма квадратов —

Поэтому

Нормированный ESS:

Для четырёх одинаковых весов ESS равен 4, нормированный — 1.

После клиппирования последнего логарифма сверху $\log2$ все веса равны $(1,2,2,2)$:

нормированный ESS равен примерно 0.942.

Клиппирование уменьшило концентрацию, но изменило оцениватель. ESS описывает распределение весов, а не гарантирует корректную целевую функцию или хорошее качество политики.

Вычислим:

Минимум среди кандидатов достигается при $K=4$.

Стационарная последовательность задержек равна $(0,1,2,3)$, поэтому максимум — 3, среднее —

Для 10 обновлений:

Сумма равна 13, точное конечновыборочное среднее — 1.3.

Оптимальный интервал между контрольными точками:

В минутах:

Модель не учитывает время перезапуска, асинхронную запись, коррелированные отказы, стоимость сети и хранилища, частично потерянные очереди, несовместимость распределённой раскладки и изменение риска во времени.

def model_state_bytes(
    total_parameters,
    trainable_parameters,
    parameter_bytes,
    gradient_bytes,
    optimizer_state_bytes,
    master_parameter_bytes,
):
    integers = [
        total_parameters,
        trainable_parameters,
        parameter_bytes,
        gradient_bytes,
        optimizer_state_bytes,
        master_parameter_bytes,
    ]
    if any(isinstance(x, bool) or not isinstance(x, int) for x in integers):
        raise ValueError("all arguments must be integers")
    if total_parameters <= 0 or not 0 <= trainable_parameters <= total_parameters:
        raise ValueError("invalid parameter counts")
    if min(integers[2:]) < 0:
        raise ValueError("byte counts must be non-negative")

    base = total_parameters * parameter_bytes
    trainable_extra = trainable_parameters * (
        gradient_bytes + optimizer_state_bytes + master_parameter_bytes
    )
    return base + trainable_extra


def kv_cache_bytes(
    num_sequences,
    tokens_per_sequence,
    num_layers,
    num_kv_heads,
    head_dim,
    element_bytes,
):
    values = [
        num_sequences,
        tokens_per_sequence,
        num_layers,
        num_kv_heads,
        head_dim,
        element_bytes,
    ]
    if any(isinstance(x, bool) or not isinstance(x, int) or x <= 0 for x in values):
        raise ValueError("all dimensions must be positive integers")
    return (
        2
        * num_sequences
        * tokens_per_sequence
        * num_layers
        * num_kv_heads
        * head_dim
        * element_bytes
    )

Проверка:

assert model_state_bytes(7_000_000_000, 7_000_000_000, 2, 2, 8, 4) == 112_000_000_000
assert model_state_bytes(7_020_000_000, 20_000_000, 2, 2, 8, 4) == 14_320_000_000
assert kv_cache_bytes(64, 4096, 32, 8, 128, 2) == 32 * 1024**3

import math

REQUIRED = {
    "sample_id", "prompt_id", "policy_version",
    "prompt_tokens", "response_tokens", "behavior_logprobs",
    "decoding", "reward", "stop_reason", "verifier_version",
}


def rollout_contract_errors(record):
    if not isinstance(record, dict):
        return ["record must be a dict"]

    errors = []
    missing = sorted(REQUIRED - set(record))
    if missing:
        errors.append("missing fields: " + ", ".join(missing))

    version = record.get("policy_version")
    if isinstance(version, bool) or not isinstance(version, int) or version < 0:
        errors.append("policy_version must be a non-negative integer")

    for key in ("prompt_tokens", "response_tokens"):
        tokens = record.get(key)
        if not isinstance(tokens, list) or any(
            isinstance(token, bool) or not isinstance(token, int) or token < 0
            for token in (tokens or [])
        ):
            errors.append(f"invalid {key}")

    logps = record.get("behavior_logprobs")
    if not isinstance(logps, list) or any(
        not isinstance(value, (int, float))
        or not math.isfinite(value)
        or value > 0
        for value in (logps or [])
    ):
        errors.append("invalid behavior_logprobs")
    elif isinstance(record.get("response_tokens"), list) and len(logps) != len(record["response_tokens"]):
        errors.append("logprobs do not align with response_tokens")

    return errors

Полная эталонная версия дополнительно проверяет строки, вложенные отображения и выдаёт более точные сообщения. В М11 token_ids, policy_mask и turn_ids описывают всю траекторию, а policy_versions — версии её ходов. Здесь выберите одно продолжение, созданное одной поведенческой версией: prompt_tokens — контекст перед его генерацией, response_tokens — выбранные политикой токены этого продолжения, behavior_logprobs выровнены именно с ними. Одиночная числовая policy_version требует явного реестра версий; строку версии нельзя молча приводить к числу. prompt_id задаётся при формировании запроса. Преобразование rewards в reward и выбор stop_reason требуют объявленного правила кредитования и остановки данного продолжения; итог эпизода не следует автоматически приписывать каждому ходу. Идентификаторы среды, действия, наблюдения, происхождение токенов и флаги всей траектории остаются в исходном журнале, связанном с инфраструктурной записью. Это проекция на другую единицу данных, а не замена полного агентного контракта.

def diagnose_pipeline(metrics):
    required = {
        "trainer_idle_fraction",
        "rollout_idle_fraction",
        "verifier_fraction",
        "weight_sync_fraction",
        "queue_growth_per_minute",
        "normalized_ess",
        "oom_events",
    }
    missing = required - set(metrics)
    if missing:
        raise ValueError("missing metrics: " + ", ".join(sorted(missing)))

    bounded = required - {"queue_growth_per_minute", "oom_events"}
    for key in bounded:
        if not 0 <= metrics[key] <= 1:
            raise ValueError(f"{key} must lie in [0, 1]")
    if metrics["oom_events"] < 0:
        raise ValueError("oom_events must be non-negative")

    hypotheses = []
    if metrics["trainer_idle_fraction"] > 0.25 and metrics["queue_growth_per_minute"] <= 0:
        hypotheses.append("check rollout, verifier, and input supply")
    if metrics["verifier_fraction"] > 0.30:
        hypotheses.append("profile the verifier critical path")
    if metrics["weight_sync_fraction"] > 0.20:
        hypotheses.append("profile weight transfer and conversion")
    if metrics["normalized_ess"] < 0.50:
        hypotheses.append("separate policy lag from training-inference mismatch")
    if metrics["oom_events"] > 0:
        hypotheses.append("measure phase-specific peaks, KV cache, activations, and transfer buffers")
    if not hypotheses:
        hypotheses.append("inspect a stage-level trace")
    return hypotheses

Главный принцип решения: выводы сформулированы как гипотезы, которые можно проверить измерением. По одной агрегированной метрике нельзя надёжно определить причину.