| name | truncated-ppo |
| title | Truncated Proximal Policy Optimization |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2506.15050 |
| keywords | ["policy-optimization","PPO","training-efficiency","reasoning","truncated-rollouts"] |
| description | T-PPO improves training efficiency via truncated rollouts and extended GAE, enabling batch continuity without waiting for full sequence completion. |
Truncated Proximal Policy Optimization
Core Concept
Truncated PPO (T-PPO) enhances training efficiency for reasoning LLMs through truncated rollouts and extended generalized advantage estimation (EGAE). Rather than waiting for complete generation sequences before updating, T-PPO divides long sequences into fixed-window chunks and applies EGAE to compute advantages from incomplete trajectories. This approach maintains constant batch size through replacement of finished sequences while achieving 60% wall-clock time reduction and 2.5x training efficiency improvement without sacrificing convergence on mathematical reasoning tasks.
Architecture Overview
- Truncated Rollout Strategy: Divide response generation into fixed-length windows rather than waiting for completion
- Extended GAE (EGAE): Generalize advantage estimation to incomplete trajectories by assuming terminal value equals penultimate state value
- Batch Continuity: Replace finished sequences with new samples, maintaining constant batch size and GPU utilization
- Token Filtering: Selectively filter training tokens to enable independent policy/value optimization
- Successive Batching: Continuously feed new prompts while earlier ones still generate
Implementation
Step 1: Implement Truncated Rollout Strategy
Divide long sequences into manageable chunks:
import torch
import torch.nn as nn
from typing import List, Dict
class TruncatedRolloutBuffer:
"""
Manages truncated rollouts: divides generation into windows.
"""
def __init__(self, window_size=128, max_total_length=512):
self.window_size = window_size
self.max_total_length = max_total_length
def create_windows(self, prompt_ids, max_new_tokens):
"""
Divide expected generation into fixed-length windows.
Args:
prompt_ids: [batch, prompt_len] token IDs
max_new_tokens: maximum tokens to generate per prompt
Returns:
window_config: dict with window boundaries and settings
"""
batch_size = prompt_ids.shape[0]
num_windows = (max_new_tokens + self.window_size - 1) // self.window_size
window_config = {
'window_size': self.window_size,
'num_windows': num_windows,
'batch_size': batch_size,
'max_new_tokens': max_new_tokens
}
return window_config
def process_truncated_generation(self, prompt_ids, generation_outputs,
window_idx):
"""
Extract tokens for a specific generation window.
Args:
prompt_ids: [batch, prompt_len]
generation_outputs: full generated sequences (may not be complete)
window_idx: which window to extract
Returns:
window_tokens: [batch, window_size] tokens in this window
is_complete: [batch] boolean indicating if sequence completed
"""
start_idx = window_idx * .window_size
end_idx = (window_idx + ) * .window_size
window_tokens = generation_outputs[:, start_idx:end_idx]
window_tokens.shape[] < .window_size:
padding = torch.full(
(window_tokens.shape[], .window_size - window_tokens.shape[]),
fill_value=
)
window_tokens = torch.cat([window_tokens, padding], dim=)
is_complete = (window_tokens != ).(dim=) < .window_size
window_tokens, is_complete
Step 2: Implement Extended GAE
Generalize GAE to incomplete trajectories:
class ExtendedGeneralizedAdvantageEstimation:
"""
Extended GAE: handles incomplete (truncated) trajectories.
Key assumption: V(s_l) = V(s_{l-1}) for ungenerated states.
"""
def __init__(self, gamma=0.99, gae_lambda=0.95):
self.gamma = gamma
self.gae_lambda = gae_lambda
def compute_advantages_truncated(self, rewards, values, dones,
trajectory_length):
"""
Compute advantages assuming incomplete trajectories.
Args:
rewards: [batch, seq_len] rewards at each position
values: [batch, seq_len] state value estimates
dones: [batch, seq_len] episode termination flags
trajectory_length: list of actual lengths per batch
Returns:
advantages: [batch, seq_len] advantage estimates
returns: [batch, seq_len] return estimates
"""
batch_size, seq_len = rewards.shape
advantages = torch.zeros_like(rewards)
gae = 0
for t in reversed(range(seq_len)):
if t == seq_len - 1:
next_value = values[:, t].clone()
next_non_terminal = 1 - dones[:, t]
else:
next_value = values[:, t + 1]
next_non_terminal = 1 - dones[:, t]
delta = (
rewards[:, t] +
self.gamma * next_value * next_non_terminal -
values[:, t]
)
gae = (
delta +
.gamma * .gae_lambda * next_non_terminal * gae
)
advantages[:, t] = gae
returns = advantages + values
advantages, returns
():
batch_size, seq_len = rewards.shape
advantages = torch.zeros_like(rewards)
gae =
next_value = values[:, -].clone()
t ((seq_len)):
t == seq_len - :
next_non_terminal = - dones[:, t]
next_value = values[:, t]
:
next_non_terminal = - dones[:, t]
next_value = values[:, t + ]
delta = (
rewards[:, t] +
.gamma * next_value * next_non_terminal -
values[:, t]
)
gae = (
delta +
.gamma * .gae_lambda * next_non_terminal * gae
)
advantages[:, t] = gae
returns = advantages + values
advantages, returns
Step 3: Implement Token Filtering
Selectively filter tokens for optimization:
class TokenFilter:
"""
Filters training tokens to improve efficiency.
"""
@staticmethod
def filter_high_advantage(tokens, advantages, log_probs,
percentile_threshold=50):
"""
Focus on tokens with high absolute advantage.
Args:
tokens: [batch, seq_len] token IDs
advantages: [batch, seq_len] advantage estimates
log_probs: [batch, seq_len] log probabilities
percentile_threshold: keep tokens above this percentile
Returns:
filtered_mask: [batch, seq_len] boolean mask
"""
advantage_threshold = torch.kthvalue(
advantages.reshape(-1),
int(advantages.numel() * (100 - percentile_threshold) / 100)
).values
filtered_mask = torch.abs(advantages) > advantage_threshold
return filtered_mask
@staticmethod
def filter_redundant_tokens(log_probs, entropy_threshold=0.1):
"""
Filter low-entropy (confident) tokens to reduce redundancy.
Args:
log_probs: [batch, seq_len, vocab_size]
entropy_threshold: keep if entropy > threshold
Returns:
filtered_mask: [batch, seq_len] boolean mask
"""
probs = torch.softmax(log_probs, dim=-1)
entropy = -(probs * torch.log(probs + 1e-8)).sum(dim=-1)
filtered_mask = entropy > entropy_threshold
return filtered_mask
@staticmethod
def filter_composite():
mask_advantage = TokenFilter.filter_high_advantage(
, advantages, , percentile_threshold=
)
mask_entropy = TokenFilter.filter_redundant_tokens(
log_probs, entropy_threshold=
)
combined_mask = (
weight_advantage * mask_advantage.() +
( - weight_advantage) * mask_entropy.()
) >
combined_mask
Step 4: Implement Successive Batching
Maintain constant batch size with continuous sampling:
class SuccessiveBatchSampler:
"""
Samples batches continuously, replacing finished sequences.
Maintains constant batch size for GPU efficiency.
"""
def __init__(self, prompt_dataset, batch_size=32, window_size=128):
self.prompt_dataset = prompt_dataset
self.batch_size = batch_size
self.window_size = window_size
self.current_idx = 0
self.active_sequences = []
def get_next_batch(self):
"""
Return batch with new samples replacing finished ones.
Returns:
batch: dict with prompts, generation_states
"""
batch = {
'new_prompts': [],
'in_progress': self.active_sequences.copy()
}
num_new_needed = self.batch_size - len(self.active_sequences)
for _ in range(num_new_needed):
if self.current_idx >= len(self.prompt_dataset):
self.current_idx = 0
prompt = self.prompt_dataset[self.current_idx]
batch['new_prompts'].append(prompt)
self.active_sequences.append({
: prompt,
: [],
:
})
.current_idx +=
batch
():
.active_sequences = [
seq i, seq (.active_sequences)
finished_mask[i]
]
Step 5: Implement T-PPO Training Loop
Complete training with truncation and successive batching:
class TruncatedPPOTrainer:
"""
Full T-PPO trainer with truncated rollouts and batch continuity.
"""
def __init__(self, model, value_model, window_size=128, batch_size=32):
self.model = model
self.value_model = value_model
self.window_size = window_size
self.batch_size = batch_size
self.egae = ExtendedGeneralizedAdvantageEstimation()
self.rollout_buffer = TruncatedRolloutBuffer(window_size)
self.batch_sampler = SuccessiveBatchSampler(
[], batch_size=batch_size, window_size=window_size
)
def training_step(self, batch_prompts, batch_responses, window_idx,
verification_fn):
"""
Single training step on a generation window.
Args:
batch_prompts: [batch, prompt_len]
batch_responses: [batch, total_len] responses generated so far
window_idx: which window to train on
verification_fn: function to verify answer correctness
"""
window_tokens, is_complete = self.rollout_buffer.process_truncated_generation(
batch_prompts, batch_responses, window_idx
)
rewards = torch.zeros(self.batch_size, self.window_size)
for i in range(self.batch_size):
response_text = self.model.tokenizer.decode(batch_responses[i])
is_correct = verification_fn(response_text)
is_complete[i] (window_idx == .rollout_buffer.window_size - ):
rewards[i, -] = is_correct
torch.no_grad():
value_outputs = .value_model(window_tokens)
values = value_outputs.squeeze(-)
advantages, returns = .egae.compute_advantages_truncated(
rewards,
values,
~is_complete.unsqueeze(),
.window_size
)
policy_outputs = .model(window_tokens)
log_probs = policy_outputs.log_probs
token_filter = TokenFilter()
filter_mask = token_filter.filter_composite(
advantages, policy_outputs.logits
)
ratio = torch.exp(log_probs - policy_outputs.old_log_probs)
clipped_ratio = torch.clamp(ratio, - , + )
loss_actor = -torch.(
ratio * advantages,
clipped_ratio * advantages
)[filter_mask].mean()
loss_value = torch.nn.functional.mse_loss(values, returns)
loss = loss_actor + * loss_value
loss, {
: loss_actor.item(),
: loss_value.item(),
: filter_mask.().mean().item()
}
():
optimizer = torch.optim.Adam(
(.model.parameters()) +
(.value_model.parameters()),
lr=
)
epoch (num_epochs):
epoch_loss =
epoch_metrics = {}
batch_idx, (prompts, responses) (dataloader):
max_len = responses.shape[]
num_windows = (max_len + .window_size - ) // .window_size
window_idx (num_windows):
loss, metrics = .training_step(
prompts, responses, window_idx,
verification_fn=._verify_correctness
)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(
(.model.parameters()) +
(.value_model.parameters()),
)
optimizer.step()
epoch_loss += loss.item()
k, v metrics.items():
k epoch_metrics:
epoch_metrics[k] = []
epoch_metrics[k].append(v)
avg_loss = epoch_loss / ((dataloader) * num_windows)
()
k, v epoch_metrics.items():
()
():
Practical Guidance
- Window Size: Larger windows (256) reduce overhead but increase memory; start with 128
- EGAE Assumption: V(s_l) = V(s_{l-1}) works well empirically; validate on your tasks
- Token Filtering: Apply selectively; too aggressive filtering can bias updates
- Batch Continuity: Requires prompt dataset large enough to avoid repetition
- Speedup Target: Expect 2-3x speedup; measure actual wall-clock time, not just theoretical
- Hyperparameters: Use standard PPO settings (clip=0.2, GAE lambda=0.95); adjust window size first
- Evaluation: Benchmark on long-sequence reasoning tasks (AIME, MATH500, code generation)
- Comparison: Always compare to standard PPO on same tasks to validate efficiency gains
Reference
Paper: arXiv:2506.15050
Key metrics: 60% wall-clock reduction, 2.5x training efficiency on math reasoning
EGAE improvement: Handles incomplete trajectories enabling continuous batching
Related work: PPO, GAE, policy optimization, training efficiency
