o `۷io@snddlZddlZddlZddlmZmZmZmZmZm Z m Z m Z ddl Z ddlZddlZddlmZddlmZmZddlmZddlmZddlmZmZddlmZdd lm Z dd l!m"Z"m#Z#dd l$m%Z%erpdd l&m'Z'ej(ej(ej)ej)ej*ej*ej+ej+ej,ej,ej-ej-ej.ej.ej/ej/ej0ej0ej1ej1ej2ej2d d ddddiZ3e4e5Z6e%ddGdddZ7dS)N) TYPE_CHECKINGAnyDictListOptionalSetTupleUnion)Columns) MultiRLModuleMultiRLModuleSpec)SingleAgentEpisode) flatten_dict)OverrideToImplementCustomLogic5OverrideToImplementCustomLogic_CallToSuperRecommended)unpack_if_needed) ReplayBuffer) EpisodeTypeModuleID) PublicAPI)AlgorithmConfig agent_indexdones unroll_idalpha) stabilityc@seZdZdZedddddddeeejejfdee dee e e fd e e e ff d d Zed e e ejfd e e ejffddZed efddZed e e e ffddZdeed eefddZdeed eefddZdeed eefddZd)d efddZeee dddddfded e e e!e"ejffde e e fd ed!eee d"eed#ejd$ejd e e e fd e e ee#ffd%d&Z$eee ddfded e e e!e"ejffde e e fd ed!eee d e e ee#ff d'd(Z%dS)*OfflinePreLearnera?Class that coordinates data transformation from dataset to learner. This class is an essential part of the new `Offline RL API` of `RLlib`. It is a callable class that is run in `ray.data.Dataset.map_batches` when iterating over batches for training. It's basic function is to convert data in batch from rows to episodes (`SingleAGentEpisode`s for now) and to then run the learner connector pipeline to convert further to trainable batches. These batches are used directly in the `Learner`'s `update` method. The main reason to run these transformations inside of `map_batches` is for better performance. Batches can be pre-fetched in `ray.data` and therefore batch trransformation can be run highly parallelized to the `Learner''s `update`. This class can be overridden to implement custom logic for transforming batches and make them 'Learner'-ready. When deriving from this class the `__call__` method and `_map_to_episodes` can be overridden to induce custom logic for the complete transformation pipeline (`__call__`) or for converting to episodes only ('_map_to_episodes`). Custom `OfflinePreLearner` classes can be passed into `AlgorithmConfig.offline`'s `prelearner_class`. The `OfflineData` class will then use the custom class in its data pipeline. N)spaces module_spec module_stateconfigrrrrkwargscKs||_|jj|_|jj|_||_|j||pd\|_|_|jj|j|jd|_ |jj |_ |j |_ d|_|jsB|jsB|jrZ|jjpH|j}|j|jjB}|di||_dSdS)N)NN)input_observation_spaceinput_action_spacer)r input_read_episodesinput_read_sample_batchesbuild_module set_stateobservation_space action_spacebuild_learner_connector_learner_connectorpolicies_to_train_policies_to_trainis_multi_agent_is_multi_agentiter_since_last_module_update is_statefulprelearner_buffer_classdefault_prelearner_buffer_class default_prelearner_buffer_kwargsprelearner_buffer_kwargsepisode_buffer)selfr rrrr!r4r7r$r$Z/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/rllib/offline/offline_prelearner.py__init__Qs<      zOfflinePreLearner.__init__batchreturnc s|jrddlddlfdd|dD}||}n3|jr9tj|j|dt|j j B|j j dd}||}n|j |j|t|j j Bd |j j |j |jd d}|j|ji|idd }|D] }|||sh||=q]t|S) aqPrepares plain data batches for training with `Learner`'s. Args: batch: A dictionary of numpy arrays containing either column data with `self.config.input_read_schema`, `EpisodeType` data, or `BatchType` data. Returns: A `MultiAgentBatch` that can be passed to `Learner.update` methods. rNcs"g|] }tj|jdqS)) object_hook)r from_stateunpackbdecode).0statemnpmsgpackr$r: s z.OfflinePreLearner.__call__..itemT)to_numpyschemainput_compress_columnsepisodesF)rJrIrKr*r+) rl_moduler<rL shared_datametrics)r%rF msgpack_numpy_postprocess_and_sampler&r_map_sample_batch_to_episoder1SCHEMAr input_read_schemarK_map_to_episodesr*r+r-r(_should_module_be_updatedr)r9r<rL module_idr$rDr:__call__sT      & zOfflinePreLearner.__call__cCsddlm}|S)zSets the default replay buffer.r)EpisodeReplayBuffer)4ray.rllib.utils.replay_buffers.episode_replay_bufferrY)r9rYr$r$r:r5s z1OfflinePreLearner.default_prelearner_buffer_classcCs|jjd|jjdS)zSets the default arguments for the replay buffer. Note, the `capacity` might vary with the size of the episodes or sample batches in the offline dataset. )capacity batch_size_B)r train_batch_size_per_learner)r9r$r$r:r6s z2OfflinePreLearner.default_prelearner_buffer_kwargsrLcCsbtdd|Ds tdt}t}|D]}|j|vr.|j|jjvr.||j||q|S)aValidate episodes sampled from the dataset. Note, our episode buffers cannot handle either duplicates nor non-ordered fragmentations, i.e. fragments from episodes that do not arrive in timestep order. Args: episodes: A list of `SingleAgentEpisode` instances sampled from a dataset. Returns: A set of `SingleAgentEpisode` instances. Raises: ValueError: If not all episodes are `done`. css|]}|jVqdSN)is_done)rBepsr$r$r: sz7OfflinePreLearner._validate_episodes..zWhen sampling from episodes (`input_read_episodes=True`) all recorded episodes must be done (i.e. either `terminated=True`) or `truncated=True`).)all ValueErrorsetid_r8episode_id_to_indexadd)r9rLunique_episode_idscleaned_episodesrar$r$r:_validate_episodess   z$OfflinePreLearner._validate_episodescCs:|D]}tj|jvr|jtj=tj|jvr|jtj=q|S)a]Removes states from episodes. This is necessary, if the module is stateful and we want to enable the offline RLModule to learn its own state representations. Args: episodes: A list of `SingleAgentEpisode` instances. Returns: A list of `SingleAgentEpisode` instances without states. )r STATE_OUTextra_model_outputsSTATE_IN)r9rLrar$r$r:_remove_states_from_episodes/s    z.OfflinePreLearner._remove_states_from_episodesc Cs||}|jr|jjs||}|j||jr&|jj ddnd}|jj |jj ||j dddd|jj t |jdddS) zPostprocesses episodes and samples from the buffer. Args: episodes: A list of `SingleAgentEpisode` instances. Returns: A list of `SingleAgentEpisode` instances sampled from the buffer. max_seq_lenrNn_stepT burnin_len) num_itemsbatch_length_Trqsample_episodesrIlookbackmin_batch_length_T)rkr(r3r %prelearner_use_recorded_module_statesror8rh model_configgetsampler^episode_lookback_horizongetattr)r9rLrur$r$r:rQEs(     z)OfflinePreLearner._postprocess_and_samplecCs.|jsdSt|js|t|jvS|||S)z:Checks which modules in a MultiRLModule should be updated.T)r/callablere)r9rWmulti_agent_batchr$r$r:rVos   z+OfflinePreLearner._should_module_be_updatedFr0rJrIrKignore_final_observationr*r+c spgg} t|tjD]\} |r4tj|vr#|tjnd|vr1|dnd} nd} |r9qtjvrBt| n| } |rRtddt| } ntjvrbt|tj n|tj } t tj |vr~t |tj nt j| | | gitj|vr|tjnigtjvrt|tjn|tjg|tjg|tj|vrȈtjndtj|vr|tjndfdd|Dd d }|r|| |qd | iS) z!Maps a batch of data to episodes.rNcSsd|S)Nrr$)xr$r$r:sz4OfflinePreLearner._map_to_episodes..rFcsLi|]"\}}|vr$|vr$|dvr||vrt|n|gqS)rrtypevaluesrrBkvirKrJr$r: s   z6OfflinePreLearner._map_to_episodes..r rfagent_id observationsinfosactionsrewards terminated truncatedrmlen_lookback_bufferrL) enumerater OBSAGENT_IDrtree map_structurecopydeepcopyNEXT_OBSr EPS_IDstruuiduuid4hexINFOSACTIONSREWARDS TERMINATEDS TRUNCATEDSitemsrIappend)r0r<rJrIrKrr*r+r!rLobsr unpacked_obsunpacked_next_obsepisoder$rr:rUysx         8 z"OfflinePreLearner._map_to_episodesc spgg}t|tjD]l\|r)d|vr&|ddnd}nd}|r.qttrFtfddtjdDn"ttj r^tjvrYt n n t dt dtj |vrtjvrt|tj dn |tj dndtj|vrtj|vr|tjd}|tjd}nLtj|vrڈtj|vr|tjd}d }n0tj|vrtj|vr|tjd}d }nd |vr|d d}d }nd }d }ttj|vr t|tjdntj|tj|vr7|tjnigttjvrNt|tjn|tj|tj||fd d |Ddd } |rw| || qd|iS)z6Maps an old stack `SampleBatch` to new stack episodes.rrNcsg|]}|dfqS).r$)rBr)rr$r:rG$szBOfflinePreLearner._map_sample_batch_to_episode..zUnknown observation type: z. When mapping from old recorded `SampleBatches` batched observations should be either of type `np.array` or - if the column is compressed - of `str` type.FdoneTcsJi|]!\}}|vr#|vr#|dvr||vrt|n|qSrrrrr$r:rs  zBOfflinePreLearner._map_sample_batch_to_episode..rrL)rr r isinstancerrrangeshapenpndarraytolist TypeErrorrrrrrr rrrrrlenrrrrI) r0r<rJrIrKrLrrrrr$)rrKrrJr:rRs           . z.OfflinePreLearner._map_sample_batch_to_episoder_)&__name__ __module__ __qualname____doc__rrrgymSpacer rrrrr;rrrrXpropertyrr5r6rr rrkrorQboolrV staticmethodrSr listrrUrRr$r$r$r:r5s  :(e   '  *     x  r)8rloggingrtypingrrrrrrrr gymnasiumrnumpyrrray.rllib.core.columnsr (ray.rllib.core.rl_module.multi_rl_moduler r "ray.rllib.env.single_agent_episoder ray.rllib.utilsrray.rllib.utils.annotationsrrray.rllib.utils.compressionr,ray.rllib.utils.replay_buffers.replay_bufferrray.rllib.utils.typingrrray.util.annotationsr%ray.rllib.algorithms.algorithm_configrrr MODULE_IDrrrrrrrTrS getLoggerrloggerrr$r$r$r:sH(