(self)
| 278 | |
| 279 | class TestExplorerGSM8k(BaseExplorerCase): |
| 280 | def test_explorer(self): |
| 281 | self.config.algorithm.repeat_times = 2 |
| 282 | self.config.buffer.total_epochs = 1 |
| 283 | self.config.buffer.explorer_input.taskset = get_unittest_dataset_config("gsm8k") |
| 284 | self.config.name = f"explore-{datetime.now().strftime('%Y%m%d%H%M%S')}" |
| 285 | # some step may be skipped due to same reward |
| 286 | self.config.algorithm.algorithm_type = "grpo" |
| 287 | self.config.algorithm.advantage_fn = "grpo" |
| 288 | self.config.algorithm.advantage_fn_args = { |
| 289 | "epsilon": 1e-6, |
| 290 | } |
| 291 | self.config.model.max_model_len = 10240 |
| 292 | self.config.model.max_response_tokens = 8192 |
| 293 | self.config.model.min_response_tokens = 8192 |
| 294 | self.config.explorer.rollout_model.ignore_eos = True |
| 295 | self.config.check_and_update() |
| 296 | explorer = Explorer.get_actor(self.config) |
| 297 | ray.get(explorer.prepare.remote()) |
| 298 | ray.get(explorer.sync_weight.remote()) |
| 299 | ray.get(explorer.explore.remote()) |
| 300 | parser = TensorBoardParser(os.path.join(self.config.monitor.cache_dir, "tensorboard")) |
| 301 | rollout_metrics = parser.metric_list("rollout") |
| 302 | self.assertTrue(len(rollout_metrics) > 0) |
| 303 | eval_metrics = parser.metric_list("eval") |
| 304 | self.assertTrue(len(eval_metrics) == 0) |
| 305 | self.assertEqual(parser.metric_max_step(rollout_metrics[0]), 4) |
| 306 | self.assertTrue(parser.metric_exist("experience_pipeline/experience_count")) |
| 307 | experience_counts = parser.metric_values("experience_pipeline/experience_count") |
| 308 | self.assertTrue(len(experience_counts) == 4) |
| 309 | for count in experience_counts: |
| 310 | self.assertTrue(count >= 0) |
| 311 | self.assertTrue(count <= 2 * 4) # repeat_times * batch_size |
| 312 | self.assertTrue(count % 2 == 0) # should be multiple of repeat_times |
| 313 | exp_save_path = self.config.buffer.trainer_input.experience_buffer.path |
| 314 | with open(exp_save_path, "r", encoding="utf-8") as f: |
| 315 | lines = f.readlines() |
| 316 | self.assertTrue(len(lines) <= 4 * 2 * 4) # step * repeat_times * batch_size |
| 317 | self.assertTrue(len(lines) % (2 * 4) == 0) |
| 318 | exp = json.loads(lines[0]) |
| 319 | self.assertEqual(exp["response_length"], 8192) |
| 320 | ray.get(explorer.shutdown.remote()) |
| 321 | |
| 322 | |
| 323 | class TestExplorerCoordinatorPath(unittest.IsolatedAsyncioTestCase): |
nothing calls this directly
no test coverage detected