feat: add standard PPO training with GAE and value critic (Transformers/FSDP)

.gitattributes

@@ -0,0 +1 @@
+* text=auto

README.md

@@ -93,6 +93,7 @@ sh INSTALL_MEGATRON.sh
 | pp/tp/cp MoE finetuning              | megatron        | [Script](cookbook/megatron/tp_moe.py)                  |
 | Multimodal FSDP finetuning           | transformers    | [Script](cookbook/mm/fsdp2.py)                         |
 | GRPO RL training                     | megatron        | [Script](cookbook/rl/grpo/grpo.py)                          |
+| PPO RL training                      | transformers    | [Script](cookbook/rl/ppo/ppo.py)                            |
 | GRPO Multimodal RL training          | megatron        | [Script](cookbook/rl/grpo/grpo_mm.py)                       |
 | GRPO Math RL training                | megatron        | [Script](cookbook/rl/grpo/short_math_grpo.py)               |
 | DPO full-parameter training          | transformers    | [Script](cookbook/rl/dpo/dpo_full.py)                      |

README_ZH.md

@@ -87,6 +87,7 @@ sh INSTALL_MEGATRON.sh
 | pp/tp/cp MoE 微调                   | megatron              | [脚本](cookbook/megatron/tp_moe.py)                    |
 | 多模态 FSDP 微调                    | transformers          | [脚本](cookbook/mm/fsdp2.py)                           |
 | GRPO 强化学习训练                    | megatron              | [脚本](cookbook/rl/grpo/grpo.py)                            |
+| PPO 强化学习训练                     | transformers          | [脚本](cookbook/rl/ppo/ppo.py)                              |
 | GRPO 多模态强化学习训练             | megatron              | [脚本](cookbook/rl/grpo/grpo_mm.py)                         |
 | GRPO 数学强化学习训练               | megatron              | [脚本](cookbook/rl/grpo/short_math_grpo.py)                 |
 | DPO 全参数训练                      | transformers          | [脚本](cookbook/rl/dpo/dpo_full.py)                        |

cookbook/rl/ppo/ppo.py

@@ -0,0 +1,233 @@
+"""Standard PPO training on GSM8K with a LoRA policy and full-parameter critic.
+
+The first implementation supports the Transformers/Accelerate-FSDP backend. Policy,
+critic, and vLLM sampler use separate GPU groups. The frozen policy base model is
+used as the reference policy.
+"""
+import random
+from typing import Any, Dict, List, Tuple
+
+from peft import LoraConfig
+
+import twinkle
+from twinkle import DeviceGroup, DeviceMesh, get_device_placement, get_logger
+from twinkle.advantage import GAEAdvantage
+from twinkle.checkpoint_engine import CheckpointEngineManager
+from twinkle.cli import CLI
+from twinkle.data_format import SamplingParams
+from twinkle.dataloader import DataLoader
+from twinkle.dataset import Dataset, DatasetMeta
+from twinkle.metric import CompletionRewardMetric, PPOMetric, PPOValueMetric
+from twinkle.model import TransformersModel, TransformersValueModel
+from twinkle.processor import InputProcessor
+from twinkle.preprocessor.llm import GSM8KProcessor
+from twinkle.reward import GSM8KAccuracyReward, GSM8KFormatReward
+from twinkle.sampler import vLLMSampler
+
+logger = get_logger()
+args = CLI.from_args()
+
+MODEL_ID = args.model.model_id or 'ms://Qwen/Qwen3.5-4B'
+POLICY_GPUS = args.infra.model_gpus or 4
+CRITIC_GPUS = args.infra.critic_model_gpus or 4
+SAMPLER_GPUS = args.infra.sampler_gpus or 4
+NUM_GPUS = POLICY_GPUS + CRITIC_GPUS + SAMPLER_GPUS
+NUM_GENERATIONS = args.rl.num_generations or 4
+MAX_NEW_TOKENS = args.sampling.max_tokens or 1024
+POLICY_LR = args.optimizer.learning_rate or 1e-5
+CRITIC_LR = args.rl.critic_learning_rate
+MAX_STEPS = args.training.max_steps or 200
+BATCH_SIZE = args.training.batch_size or 4
+MINI_BATCH_SIZE = args.training.mini_batch_size or 4
+MICRO_BATCH_SIZE = args.training.micro_batch_size or 1
+PPO_EPOCHS = args.rl.ppo_epochs
+SAVE_STEPS = args.training.save_steps or 50
+ADAPTER_NAME = args.lora.adapter_name or 'default'
+
+
+def create_gsm8k_dataset():
+    dataset = Dataset(DatasetMeta('ms://modelscope/gsm8k', subset_name='main', split='train'))
+    dataset.set_template('Qwen3_5Template', model_id=MODEL_ID, max_length=400)
+    dataset.map(GSM8KProcessor())
+    dataset.encode(add_generation_prompt=True)
+    return dataset
+
+
+def compute_rewards(trajectories: List[Dict[str, Any]]) -> Tuple[List[float], List[float], List[float]]:
+    accuracy = GSM8KAccuracyReward()(trajectories)
+    formatting = GSM8KFormatReward()(trajectories)
+    return [a + f for a, f in zip(accuracy, formatting)], formatting, accuracy
+
+
+def response_rows(full_values, trajectories) -> List[List[float]]:
+    """Extract response-token rows from collected model outputs."""
+    import torch
+
+    value_rows = []
+    tensors = full_values if isinstance(full_values, list) else [full_values]
+    for tensor in tensors:
+        if tensor is None:
+            continue
+        tensor = torch.as_tensor(tensor)
+        if tensor.dim() == 1:
+            tensor = tensor.unsqueeze(0)
+        value_rows.extend(tensor)
+    if len(value_rows) != len(trajectories):
+        raise ValueError(f'model output batch mismatch: {len(value_rows)} rows for {len(trajectories)} trajectories')
+
+    rows = []
+    for value_row, trajectory in zip(value_rows, trajectories):
+        mask = torch.as_tensor(trajectory['labels'], device=value_row.device) != -100
+        rows.append(value_row[:mask.numel()][mask].detach().float().cpu().tolist())
+    return rows
+
+
+def main():
+    critic_start = POLICY_GPUS
+    sampler_start = POLICY_GPUS + CRITIC_GPUS
+    groups = [
+        DeviceGroup(name='policy', ranks=list(range(POLICY_GPUS)), device_type='GPU'),
+        DeviceGroup(name='critic', ranks=list(range(critic_start, sampler_start)), device_type='GPU'),
+        DeviceGroup(name='sampler', ranks=list(range(sampler_start, NUM_GPUS)), device_type='GPU'),
+    ]
+    policy_mesh = DeviceMesh.from_sizes(world_size=POLICY_GPUS, fsdp_size=POLICY_GPUS)
+    critic_mesh = DeviceMesh.from_sizes(world_size=CRITIC_GPUS, fsdp_size=CRITIC_GPUS)
+    sampler_mesh = DeviceMesh.from_sizes(world_size=SAMPLER_GPUS, dp_size=SAMPLER_GPUS)
+    twinkle.initialize(mode='ray', nproc_per_node=NUM_GPUS, groups=groups, lazy_collect=False)
+
+    policy = TransformersModel(
+        model_id=MODEL_ID, device_mesh=policy_mesh, remote_group='policy')
+    lora_config = LoraConfig(
+        target_modules=['q_proj', 'k_proj', 'v_proj', 'o_proj', 'gate_proj', 'up_proj', 'down_proj'],
+        r=32,
+        lora_alpha=64,
+        lora_dropout=0.05,
+    )
+    policy.add_adapter_to_model(ADAPTER_NAME, lora_config, gradient_accumulation_steps=1)
+    policy.set_optimizer('AdamW', lr=POLICY_LR)
+    policy.set_lr_scheduler('CosineAnnealingLR', T_max=MAX_STEPS, eta_min=0)
+    policy.set_loss('PPOLoss', epsilon=args.loss.epsilon, entropy_coef=args.loss.entropy_coef)
+    policy.add_metric(PPOMetric, epsilon=args.loss.epsilon)
+    policy.set_processor(InputProcessor)
+    policy.set_template('Qwen3_5Template', model_id=MODEL_ID)
+
+    critic = TransformersValueModel(
+        model_id=MODEL_ID, device_mesh=critic_mesh, remote_group='critic')
+    critic.set_optimizer('AdamW', lr=CRITIC_LR)
+    critic.set_lr_scheduler('CosineAnnealingLR', T_max=MAX_STEPS, eta_min=0)
+    critic.set_loss('PPOValueLoss', epsilon=args.loss.value_clip)
+    critic.add_metric(PPOValueMetric, epsilon=args.loss.value_clip)
+    critic.set_processor(InputProcessor)
+    critic.set_template('Qwen3_5Template', model_id=MODEL_ID)
+
+    sampler = vLLMSampler(
+        model_id=MODEL_ID,
+        engine_args={
+            'gpu_memory_utilization': 0.8,
+            'max_model_len': 400 + MAX_NEW_TOKENS,
+            'max_lora_rank': 32,
+            'enable_lora': True,
+            'tensor_parallel_size': 1,
+        },
+        device_mesh=sampler_mesh,
+        remote_group='sampler',
+    )
+    sampler.set_template('Qwen3_5Template', model_id=MODEL_ID)
+    checkpoint_manager = CheckpointEngineManager(model=policy, sampler=sampler)
+    dataloader = DataLoader(
+        dataset=create_gsm8k_dataset,
+        batch_size=BATCH_SIZE,
+        min_batch_size=BATCH_SIZE,
+        device_mesh=policy_mesh,
+        remote_group='policy',
+    )
+    gae = GAEAdvantage(args.rl.gamma, args.rl.gae_lambda, args.rl.normalize_advantages)
+    reward_metric = CompletionRewardMetric()
+    sampling_params = SamplingParams(max_tokens=MAX_NEW_TOKENS, num_samples=1, logprobs=1)
+
+    optim_step = 0
+    rollout_step = 0
+    logger.info(get_device_placement())
+    while optim_step < MAX_STEPS:
+        for batch in dataloader:
+            if optim_step >= MAX_STEPS:
+                break
+            reward_metric.reset()
+            prompts = batch if isinstance(batch, list) else [batch]
+            checkpoint_manager.sync_weights(merge_and_sync=False)
+            sampler.reset_prefix_cache()
+            expanded = [prompt for prompt in prompts for _ in range(NUM_GENERATIONS)]
+            samples = sampler.sample(expanded, sampling_params)
+
+            trajectories, old_logps, lengths = [], [], []
+            for response in samples:
+                for sequence in response.sequences:
+                    trajectories.append(sequence.new_input_feature)
+                    old_logps.append([entry[0][1] for entry in sequence.logprobs])
+                    lengths.append(len(sequence.tokens))
+            rewards, format_rewards, accuracy_rewards = compute_rewards(trajectories)
+            reward_metric.accumulate(
+                completion_lengths=lengths,
+                rewards={'total': rewards, 'format': format_rewards, 'accuracy': accuracy_rewards},
+            )
+
+            reference = policy.forward_only(inputs=trajectories, disable_lora=True)
+            ref_logps = response_rows(reference['logps'], trajectories)
+            critic_outputs = critic.forward_only(inputs=trajectories)
+            old_values = response_rows(critic_outputs['values'], trajectories)
+            token_rewards = gae.build_token_rewards(
+                rewards, lengths, old_logps=old_logps, ref_logps=ref_logps, kl_coef=args.rl.kl_coef)
+            max_len = max(lengths)
+            padded_rewards = [row + [0.0] * (max_len - len(row)) for row in token_rewards]
+            padded_values = [row + [0.0] * (max_len - len(row)) for row in old_values]
+            masks = [[True] * length + [False] * (max_len - length) for length in lengths]
+            advantages, returns = gae(padded_rewards, padded_values, masks=masks)
+            advantages = [advantages[i, :length].tolist() for i, length in enumerate(lengths)]
+            returns = [returns[i, :length].tolist() for i, length in enumerate(lengths)]
+
+            indices = list(range(len(trajectories)))
+            for _ in range(PPO_EPOCHS):
+                random.shuffle(indices)
+                for start in range(0, len(indices), MINI_BATCH_SIZE):
+                    chosen = indices[start:start + MINI_BATCH_SIZE]
+                    mb_inputs = [trajectories[i] for i in chosen]
+                    mb_old_logps = [old_logps[i] for i in chosen]
+                    mb_old_values = [old_values[i] for i in chosen]
+                    mb_advantages = [advantages[i] for i in chosen]
+                    mb_returns = [returns[i] for i in chosen]
+                    policy.forward_backward(
+                        inputs=mb_inputs,
+                        old_logps=mb_old_logps,
+                        advantages=mb_advantages,
+                        micro_batch_size=MICRO_BATCH_SIZE,
+                    )
+                    policy.clip_grad_and_step()
+                    critic.forward_backward(
+                        inputs=mb_inputs,
+                        old_values=mb_old_values,
+                        returns=mb_returns,
+                        advantages=mb_advantages,
+                        micro_batch_size=MICRO_BATCH_SIZE,
+                    )
+                    critic.clip_grad_and_step()
+                    optim_step += 1
+                    if optim_step % SAVE_STEPS == 0:
+                        policy.save(f'ppo-policy-checkpoint-{optim_step}')
+                        critic.save(f'ppo-critic-checkpoint-{optim_step}')
+                    if optim_step >= MAX_STEPS:
+                        break
+                if optim_step >= MAX_STEPS:
+                    break
+
+            logs = reward_metric.calculate()
+            logs.update(policy.calculate_metric(is_training=True))
+            logs.update(critic.calculate_metric(is_training=True))
+            rollout_step += 1
+            logger.info(f'[Rollout {rollout_step}, optim step {optim_step}/{MAX_STEPS}] {logs}')
+
+    policy.save('ppo-policy-final')
+    critic.save('ppo-critic-final')
+
+
+if __name__ == '__main__':
+    main()

cookbook/rl/ppo/ppo.sh

@@ -0,0 +1,29 @@
+#!/bin/sh
+set -eu
+
+# Standard PPO on GSM8K via Ray.
+# Transformers/Accelerate-FSDP: 4 policy + 4 full-parameter critic + 4 sampler GPUs.
+# Override any option after the defaults, for example:
+#   sh ppo.sh --max-steps 20 --ppo-epochs 1
+
+python ppo.py \
+    --model-id ms://Qwen/Qwen3.5-4B \
+    --model-gpus 4 \
+    --critic-model-gpus 4 \
+    --sampler-gpus 4 \
+    --num-generations 2 \
+    --max-tokens 1024 \
+    --batch-size 4 \
+    --mini-batch-size 4 \
+    --micro-batch-size 1 \
+    --ppo-epochs 4 \
+    --gamma 1.0 \
+    --gae-lambda 0.95 \
+    --kl-coef 0.01 \
+    --value-clip 0.2 \
+    --lr 1e-5 \
+    --critic-learning-rate 1e-5 \
+    --max-steps 200 \
+    --save-steps 50 \
+    --adapter-name default \
+    "$@"

src/twinkle/advantage/__init__.py

@@ -1,10 +1,12 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
 from .base import Advantage
+from .gae import GAEAdvantage
 from .grpo import GRPOAdvantage
 from .rloo import RLOOAdvantage
 
 __all__ = [
     'Advantage',
+    'GAEAdvantage',
     'GRPOAdvantage',
     'RLOOAdvantage',
 ]

src/twinkle/advantage/gae.py

@@ -0,0 +1,103 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+from typing import TYPE_CHECKING, List, Optional, Tuple, Union
+
+from .base import Advantage
+
+if TYPE_CHECKING:
+    import torch
+
+
+class GAEAdvantage(Advantage):
+    """Token-level generalized advantage estimation for terminal completions."""
+
+    def __init__(self, gamma: float = 1.0, gae_lambda: float = 0.95, normalize: bool = True):
+        if not 0.0 <= gamma <= 1.0:
+            raise ValueError('gamma must be in [0, 1]')
+        if not 0.0 <= gae_lambda <= 1.0:
+            raise ValueError('gae_lambda must be in [0, 1]')
+        self.gamma = gamma
+        self.gae_lambda = gae_lambda
+        self.normalize = normalize
+
+    @staticmethod
+    def build_token_rewards(
+        rewards: Union['torch.Tensor', List[float]],
+        lengths: List[int],
+        *,
+        old_logps: Optional[List[List[float]]] = None,
+        ref_logps: Optional[List[List[float]]] = None,
+        kl_coef: float = 0.0,
+    ) -> List[List[float]]:
+        import torch
+
+        rewards = torch.as_tensor(rewards, dtype=torch.float32).flatten().tolist()
+        if len(rewards) != len(lengths):
+            raise ValueError('rewards and lengths must have the same batch size')
+        if (old_logps is None) != (ref_logps is None):
+            raise ValueError('old_logps and ref_logps must be provided together')
+
+        token_rewards = []
+        for i, (reward, length) in enumerate(zip(rewards, lengths)):
+            if length <= 0:
+                raise ValueError('completion lengths must be positive')
+            values = [0.0] * length
+            if old_logps is not None:
+                if len(old_logps[i]) != length or len(ref_logps[i]) != length:
+                    raise ValueError(f'log-prob length mismatch at sample {i}')
+                values = [-kl_coef * (float(old) - float(ref)) for old, ref in zip(old_logps[i], ref_logps[i])]
+            values[-1] += float(reward)
+            token_rewards.append(values)
+        return token_rewards
+
+    def __call__(
+        self,
+        rewards: Union['torch.Tensor', List[List[float]]],
+        values: Union['torch.Tensor', List[List[float]]],
+        *,
+        masks: Optional[Union['torch.Tensor', List[List[bool]]]] = None,
+        normalize: Optional[bool] = None,
+        **kwargs,
+    ) -> Tuple['torch.Tensor', 'torch.Tensor']:
+        import torch
+
+        rewards = torch.as_tensor(rewards, dtype=torch.float32)
+        values = torch.as_tensor(values, dtype=torch.float32, device=rewards.device)
+        if rewards.dim() == 1:
+            rewards = rewards.unsqueeze(0)
+        if values.dim() == 1:
+            values = values.unsqueeze(0)
+        if rewards.shape != values.shape:
+            raise ValueError(f'rewards and values must have identical shapes, got {rewards.shape} and {values.shape}')
+
+        if masks is None:
+            masks = torch.ones_like(rewards, dtype=torch.bool)
+        else:
+            masks = torch.as_tensor(masks, dtype=torch.bool, device=rewards.device)
+            if masks.shape != rewards.shape:
+                raise ValueError('masks must have the same shape as rewards')
+
+        advantages = torch.zeros_like(rewards)
+        for batch_idx in range(rewards.shape[0]):
+            valid = masks[batch_idx].nonzero(as_tuple=True)[0]
+            last_gae = rewards.new_zeros(())
+            for j in range(len(valid) - 1, -1, -1):
+                pos = valid[j]
+                if j + 1 < len(valid):
+                    next_value = values[batch_idx, valid[j + 1]]
+                else:
+                    next_value = rewards.new_zeros(())
+                delta = rewards[batch_idx, pos] + self.gamma * next_value - values[batch_idx, pos]
+                last_gae = delta + self.gamma * self.gae_lambda * last_gae
+                advantages[batch_idx, pos] = last_gae
+
+        returns = advantages + values
+        should_normalize = self.normalize if normalize is None else normalize
+        if should_normalize:
+            valid_advantages = advantages[masks]
+            if valid_advantages.numel() > 1:
+                mean = valid_advantages.mean()
+                std = valid_advantages.std(unbiased=False)
+                advantages = torch.where(masks, (advantages - mean) / (std + 1e-8), advantages)
+        advantages = advantages.masked_fill(~masks, 0.0)
+        returns = returns.masked_fill(~masks, 0.0)
+        return advantages, returns