Test CPT.
(self)
| 479 | |
| 480 | class TestTrainerCPT(BaseTrainerCase): |
| 481 | def test_trainer(self): |
| 482 | """Test CPT.""" |
| 483 | # test both mode |
| 484 | self.config.mode = "train" |
| 485 | self.config.algorithm.algorithm_type = "cpt" |
| 486 | self.config.algorithm.policy_loss_fn = "sft" |
| 487 | self.config.algorithm.policy_loss_fn_args = {} |
| 488 | self.config.algorithm.kl_loss_fn = "none" |
| 489 | self.config.algorithm.entropy_loss_fn = "none" |
| 490 | self.config.buffer.train_batch_size = 8 |
| 491 | self.config.buffer.total_epochs = 2 |
| 492 | self.config.buffer.trainer_input.experience_buffer = get_unittest_dataset_config( |
| 493 | "cpt_for_countdown" |
| 494 | ) |
| 495 | self.config.check_and_update() |
| 496 | train(self.config) |
| 497 | parser = TensorBoardParser(os.path.join(self.config.monitor.cache_dir, "tensorboard")) |
| 498 | actor_metrics = parser.metric_list("actor") |
| 499 | self.assertGreater(len(actor_metrics), 0) |
| 500 | self.assertEqual(parser.metric_max_step(actor_metrics[0]), 4) |
| 501 | |
| 502 | def tearDown(self): |
| 503 | # remove dir only when the test passed |
nothing calls this directly
no test coverage detected