File size: 2,889 Bytes
1faccd4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | # Copyright 2024 Bytedance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import warnings
from enum import Enum
from omegaconf import DictConfig
from verl.single_controller.base import Worker
from verl.trainer.ppo.core_algos import AdvantageEstimator
WorkerType = type[Worker]
class Role(Enum):
"""
To create more roles dynamically, you can subclass Role and add new members
"""
Actor = 0
Rollout = 1
ActorRollout = 2
Critic = 3
RefPolicy = 4
RewardModel = 5
ActorRolloutRef = 6
Env = 7
def __str__(self):
return self._get_role_string()
def _get_role_string(self):
role_mapping = {
Role.Actor: "actor",
Role.Rollout: "rollout",
Role.ActorRollout: "actor_rollout",
Role.Critic: "critic",
Role.RefPolicy: "ref",
Role.RewardModel: "rm",
Role.ActorRolloutRef: "actor_rollout_ref",
}
return role_mapping.get(self, self.name.lower())
@classmethod
def from_string(cls, name: str):
string_mapping = {
"actor": cls.Actor,
"rollout": cls.Rollout,
"actor_rollout": cls.ActorRollout,
"critic": cls.Critic,
"ref": cls.RefPolicy,
"rm": cls.RewardModel,
"actor_rollout_ref": cls.ActorRolloutRef,
}
role = string_mapping.get(name.lower())
if role is None:
raise ValueError(f"No Role found for string: {name}")
return role
def need_reference_policy(
config: DictConfig,
) -> bool:
"""Given the config, do we need ref policy."""
return config.algorithm.use_kl_in_reward or config.actor_rollout_ref.actor.use_kl_loss
def need_reward_model(
config: DictConfig,
) -> bool:
"""Given the config, do we need reward model."""
return config.reward.reward_model.enable
def need_critic(config: DictConfig) -> bool:
"""Given a config, do we need critic."""
if config.critic.enable is not None:
return bool(config.critic.enable)
elif config.algorithm.adv_estimator == AdvantageEstimator.GAE:
return True
else:
warnings.warn(
"Disabled critic as algorithm.adv_estimator != gae. If it is not intended, please set critic.enable=True",
stacklevel=2,
)
return False
|