def _validate(self, identifier):
        """
        The training loop of PPO with global metric computation.
        Accumulates metrics across all batches before computing final statistics.
        """
        import torch
        reward_tensor_lst_em = []
        reward_tensor_lst_f1 = []
        data_source_lst_em = []
        data_source_lst_f1 = []

        gen_config = GenerationConfig(
            max_turns=self.config.trainer.max_turns,
            max_start_length=self.config.data.max_start_length,
            max_prompt_length=self.config.data.max_prompt_length,
            max_response_length=self.config.data.max_response_length,
            max_obs_length=self.config.data.max_obs_length,
            num_gpus=self.config.trainer.n_gpus_per_node,
            no_think_rl=self.config.algorithm.no_think_rl,
            simulate_llm=self.config.retriever.simulate_llm,
            llm_ip=self.config.retriever.llm_ip,
            start_threshold=self.config.retriever.start_threshold,
            end_threshold=self.config.retriever.end_threshold,
            search_mode=self.config.retriever.search_mode,
            topk = self.config.retriever.topk
        )

        # Agent config preparation
        generation_manager = LLMGenerationManager(
            tokenizer=self.tokenizer,
            actor_rollout_wg=self.actor_rollout_wg,
            config=gen_config,
            is_validation = True,
        )
        all_inference_data = []
        if not self.config.trainer.do_search:
            for test_data in tqdm(self.val_dataloader, desc="Testing"):
                test_batch = DataProto.from_single_dict(test_data)

                # we only do validation on rule-based rm
                if self.config.reward_model.enable and test_batch[0].non_tensor_batch['reward_model']['style'] == 'model':
                    return {}

                test_gen_batch = test_batch.pop(batch_keys=['input_ids', 'attention_mask', 'position_ids'], non_tensor_batch_keys=['question', 'golden_answers'])
                test_gen_batch.meta_info = {
                    'eos_token_id': self.tokenizer.eos_token_id,
                    'pad_token_id': self.tokenizer.pad_token_id,
                    'recompute_log_prob': False,
                    'do_sample': False,
                    'validate': True,
                }

                # pad to be divisible by dp_size
                test_gen_batch_padded, pad_size = pad_dataproto_to_divisor(test_gen_batch, self.actor_rollout_wg.world_size)
                test_output_gen_batch_padded = self.actor_rollout_wg.generate_sequences(test_gen_batch_padded)
                # unpad
                test_output_gen_batch = unpad_dataproto(test_output_gen_batch_padded, pad_size=pad_size)
                print('validation generation end')

                test_batch = test_batch.union(test_output_gen_batch)

                # evaluate using reward_function
                # for certain reward function (e.g. sandbox), the generation can overlap with reward
                reward_tensor_em, reward_tensor_f1, inference_data = self.val_reward_fn(test_batch)
                all_inference_data.extend(inference_data)

                reward_tensor_lst_em.append(reward_tensor_em)
                reward_tensor_lst_f1.append(reward_tensor_f1)
                data_source_lst_em.append(
                    test_batch.non_tensor_batch.get('data_source', ['unknown'] * reward_tensor_em.shape[0]))
                data_source_lst_f1.append(
                    test_batch.non_tensor_batch.get('data_source', ['unknown'] * reward_tensor_f1.shape[0]))

        else:
            for batch_dict in tqdm(self.val_dataloader, desc="Testing"):
                timing_raw = {}
                test_batch: DataProto = DataProto.from_single_dict(batch_dict)

                test_gen_batch = test_batch.pop(batch_keys=['input_ids', 'attention_mask', 'position_ids'], non_tensor_batch_keys=['question', 'golden_answers'])
                test_gen_batch.meta_info = {
                    'eos_token_id': self.tokenizer.eos_token_id,
                    'pad_token_id': self.tokenizer.pad_token_id,
                    'recompute_log_prob': False,
                    'do_sample': False,
                    'validate': True,
                }
                with _timer('step', timing_raw):
                    first_input_ids = test_gen_batch.batch['input_ids'][:, -gen_config.max_start_length:].clone()
                    with _timer('gen', timing_raw):
                        generation_manager.timing_raw = timing_raw
                        final_gen_batch_output, trajectory_turns = generation_manager.run_llm_loop(
                            gen_batch=test_gen_batch,
                            search_mode='google',
                            initial_input_ids=first_input_ids,
                            current_step=self.global_steps,
                            total_steps=self.config.trainer.total_training_steps,
                        )
                    
                    test_batch = test_batch.union(final_gen_batch_output)
                    
                    for key in test_batch.batch.keys():
                        test_batch.batch[key] = test_batch.batch[key].long()
                    
                    # evaluate using reward_function
                    # for certain reward function (e.g. sandbox), the generation can overlap with reward
                    reward_tensor_em, reward_tensor_f1, inference_data = self.val_reward_fn(test_batch)
                    all_inference_data.extend(inference_data)

                    reward_tensor_lst_em.append(reward_tensor_em)
                    reward_tensor_lst_f1.append(reward_tensor_f1)
                    data_source_lst_em.append(test_batch.non_tensor_batch.get('data_source', ['unknown'] * reward_tensor_em.shape[0]))
                    data_source_lst_f1.append(test_batch.non_tensor_batch.get('data_source', ['unknown'] * reward_tensor_f1.shape[0]))

        reward_tensor_em = torch.cat([rw.sum(-1) for rw in reward_tensor_lst_em], dim=0).cpu()  # (batch_size,)
        reward_tensor_f1 = torch.cat([rw.sum(-1) for rw in reward_tensor_lst_f1], dim=0).cpu()  # (batch_size,)
        # reward_tensor = torch.cat(reward_tensor_lst, dim=0).sum(-1).cpu()  # (batch_size,)
        data_sources_em = np.concatenate(data_source_lst_em, axis=0)
        data_sources_f1 = np.concatenate(data_source_lst_f1, axis=0)
        # evaluate test_score based on data source
        data_source_reward_em = {}
        for i in range(reward_tensor_em.shape[0]):
            data_source = data_sources_em[i]
            if data_source not in data_source_reward_em:
                data_source_reward_em[data_source] = []
            data_source_reward_em[data_source].append(reward_tensor_em[i].item())

        data_source_reward_f1 = {}
        for i in range(reward_tensor_f1.shape[0]):
            data_source = data_sources_f1[i]
            if data_source not in data_source_reward_f1:
                data_source_reward_f1[data_source] = []
            data_source_reward_f1[data_source].append(reward_tensor_f1[i].item())


        metric_dict = {}
        for data_source, rewards in data_source_reward_em.items():
            metric_dict[f'val/test_score_EM/{data_source}'] = np.mean(rewards)

        for data_source, rewards in data_source_reward_f1.items():
            metric_dict[f'val/test_score_F1/{data_source}'] = np.mean(rewards)

        os.makedirs(os.path.dirname(identifier), exist_ok=True)

        import json
        with open(identifier, 'w') as file:
            json.dump(all_inference_data, file)

        return metric_dict

        The training loop of PPO with global metric computation.
        Accumulates metrics across all batches before computing final statistics.
        
verl/trainer/ppo/ray_trainer.py