class_docstr: The DDPOTrainer uses Deep Diffusion Policy Optimization to optimise diffusion models.
  Note, this trainer is heavily inspired by the work here: https://github.com/kvablack/ddpo-pytorch
  As of now only Stable Diffusion based pipelines are supported
  
  Attributes:
      **config** (`DDPOConfig`) -- Configuration object for DDPOTrainer. Check the documentation of `PPOConfig` for more
       details.
      **reward_function** (Callable[[torch.Tensor, Tuple[str], Tuple[Any]], torch.Tensor]) -- Reward function to be used
      **prompt_function** (Callable[[], Tuple[str, Any]]) -- Function to generate prompts to guide model
      **sd_pipeline** (`DDPOStableDiffusionPipeline`) -- Stable Diffusion pipeline to be used for training.
      **image_samples_hook** (Optional[Callable[[Any, Any, Any], Any]]) -- Hook to be called to log images
docstr: Train on a batch of samples. Main training segment
  
  Args:
      inner_epoch (int): The current inner epoch
      epoch (int): The current epoch
      global_step (int): The current global step
      batched_samples (List[Dict[str, torch.Tensor]]): The batched samples to train on
  
  Side Effects:
      - Model weights are updated
      - Logs the statistics to the accelerator trackers.
  
  Returns:
      global_step (int): The updated global step
docstr_tok: ["Train","on","a","batch","of","samples",".","Main","training","segment","Args",":","inner_epoch","(","int",")",":","The","current","inner","epoch","epoch","(","int",")",":","The","current","epoch","global_step","(","int",")",":","The","current","global","step","batched_samples","(","List","[","Dict","[","str",",","torch.Tensor","]","]",")",":","The","batched","samples","to","train","on","Side","Effects",":","-","Model","weights","are","updated","-","Logs","the","statistics","to","the","accelerator","trackers",".","Returns",":","global_step","(","int",")",":","The","updated","global","step"]
code: def _train_batched_samples(
      self, inner_epoch, epoch, global_step, batched_samples
  ):
      """
      Train on a batch of samples. Main training segment
  
      Args:
          inner_epoch (int): The current inner epoch
          epoch (int): The current epoch
          global_step (int): The current global step
          batched_samples (List[Dict[str, torch.Tensor]]): The batched samples to train on
  
      Side Effects:
          - Model weights are updated
          - Logs the statistics to the accelerator trackers.
  
      Returns:
          global_step (int): The updated global step
      """
      info = defaultdict(list)
      for i, sample in enumerate(batched_samples):
          if self.config.train_cfg:
              # concat negative prompts to sample prompts to avoid two forward passes
              embeds = torch.cat(
                  [
                      sample["negative_prompt_embeds"],
                      sample["prompt_embeds"],
                  ]
              )
          else:
              embeds = sample["prompt_embeds"]
  
          for j in range(self.num_train_timesteps):
              with self.accelerator.accumulate(self.sd_pipeline.unet):
                  loss, approx_kl, clipfrac = self.calculate_loss(
                      sample["latents"][:, j],
                      sample["timesteps"][:, j],
                      sample["next_latents"][:, j],
                      sample["log_probs"][:, j],
                      sample["advantages"],
                      embeds,
                  )
                  info["approx_kl"].append(approx_kl)
                  info["clipfrac"].append(clipfrac)
                  info["loss"].append(loss)
  
                  self.accelerator.backward(loss)
                  if self.accelerator.sync_gradients:
                      self.accelerator.clip_grad_norm_(
                          self.trainable_layers.parameters()
                          if not isinstance(self.trainable_layers, list)
                          else self.trainable_layers,
                          self.config.train_max_grad_norm,
                      )
                  self.optimizer.step()
                  self.optimizer.zero_grad()
  
              # Checks if the accelerator has performed an optimization step behind the scenes
              if self.accelerator.sync_gradients:
                  # log training-related stuff
                  info = {
                      k: torch.mean(torch.stack(v))
                      for k, v in info.items()
                  }
                  info = self.accelerator.reduce(info, reduction="mean")
                  info.update(
                      {"epoch": epoch, "inner_epoch": inner_epoch}
                  )
                  self.accelerator.log(info, step=global_step)
                  global_step += 1
                  info = defaultdict(list)
      return global_step
  
imports: import os
  import warnings
  from collections import defaultdict
  from concurrent import futures
  from typing import Any, Callable, Optional, Tuple
  from warnings import warn
  import torch
  from accelerate import Accelerator
  from accelerate.logging import get_logger
  from accelerate.utils import ProjectConfiguration, set_seed
  from huggingface_hub import whoami
  from ..models import DDPOStableDiffusionPipeline
  from .None import BaseTrainer, DDPOConfig
  from .utils import PerPromptStatTracker
label_desc: Use image node_id 10 for calling the DDPOTrainer obj's underlying member method code with example usage: obj._train_batched_samples(inner_epoch, epoch, global_step, batched_samples) and returns: global_step