o 1 iXo@szddlZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z mZmZmZddlmZddlmZmZddlmZddlmZddlmZmZddlmZdd lm Z dd l!m"Z"dd l#m$Z$m%Z%dd l&m'Z'ervdd l(m)Z)ej*ej*ej+ej+ej,ej,ej-ej-ej.ej.ej/ej/ej0ej0ej1ej1ej2ej2ej3ej3ej4ej4ddddddiZ5e6e7Z8e'ddGdddZ9dS)N)AnyDictListOptionalUnionSetTuple TYPE_CHECKING)Columns)MultiRLModuleSpec MultiRLModule)SingleAgentEpisode) flatten_dict)OverrideToImplementCustomLogic5OverrideToImplementCustomLogic_CallToSuperRecommended)unpack_if_needed) ReplayBuffer)from_jsonable_if_needed) EpisodeTypeModuleID) PublicAPI)AlgorithmConfig agent_indexdonesZ unroll_idalpha)Z stabilityc@seZdZdZedddddddeeejejfdee dee e e fd e e e ff d d Zed e e ejfd e e ejffddZed efddZed e e e ffddZdeed eefddZd%d efddZeeedddddfded e e ee ejffde e e fdedeee deedejd ejd e e e fd e e ee!ffd!d"Z"eeeddfded e e ee ejffde e e fdedeee 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)Zinput_observation_spaceZinput_action_spacer)rinput_read_episodesinput_read_sample_batchesbuild_moduleZ set_stateobservation_space action_spaceZbuild_learner_connector_learner_connectorZpolicies_to_train_policies_to_trainis_multi_agent_is_multi_agentZiter_since_last_module_update is_statefulprelearner_buffer_classdefault_prelearner_buffer_class default_prelearner_buffer_kwargsprelearner_buffer_kwargsepisode_buffer)selfrrrrr r-r0r!r!p/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/rllib/offline/offline_prelearner.py__init__Rs<      zOfflinePreLearner.__init__batchreturnc sv|jrDddlddlfdd|dD}||}|j||jj|jj|j r5|jj ddnd|j ddp>dd d d }nZ|j rtj|j|d t|jjB|jjd d }||}|j||jj|jj|j rx|jj ddnd|j ddpdd d d }n|j|j|t|jjBd |jj|j|jdd }|j|j i|idd}|D] }|||s||=qt|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 Z from_stateunpackbdecode).0stateZmnpmsgpackr!r3 s z.OfflinePreLearner.__call__..itemZ max_seq_lenn_stepT) num_itemsZbatch_length_Tr@Zsample_episodesto_numpy)rCschemainput_compress_columnsepisodesF)rDrCrEr&r')Z rl_moduler5rFZ shared_dataZmetrics)r"r=Z msgpack_numpy_validate_episodesr1addsamplertrain_batch_size_per_learnerr%r,Z model_configgetr#r_map_sample_batch_to_episoder+SCHEMAZinput_read_schemarE_map_to_episodesr&r'r(_should_module_be_updatedr)r2r5rF module_idr!r<r3__call__s|         & zOfflinePreLearner.__call__cCsddlm}|S)zSets the default replay buffer.r)EpisodeReplayBuffer)Z4ray.rllib.utils.replay_buffers.episode_replay_bufferrR)r2rRr!r!r3r. s 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. )capacityZ batch_size_B)rrJ)r2r!r!r3r/s z2OfflinePreLearner.default_prelearner_buffer_kwargsrFcs8tdd|Ds tdtfdd|D}|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)Zis_doner:Zepsr!r!r3 7sz7OfflinePreLearner._validate_episodes..zWhen sampling from episodes (`input_read_episodes=True`) all recorded episodes must be done (i.e. either `terminated=True`) or `truncated=True`).cs8h|]}|jvr|js|jjjvr|qSr!)id_rHr1Zepisode_id_to_indexkeysrVr2Zunique_episode_idsr!r3 @s  z7OfflinePreLearner._validate_episodes..)all ValueErrorset)r2rFr!rZr3rG#s z$OfflinePreLearner._validate_episodescCs.|jsdSt|js|t|jvS|||S)z:Checks which modules in a MultiRLModule should be updated.T)r)callabler^)r2rPZmulti_agent_batchr!r!r3rOIs   z+OfflinePreLearner._should_module_be_updatedFr*rDrCrEignore_final_observationr&r'c s6pg|r |r t} ndd} g} t|tjD]\} |r?tj|vr.|tjnd|vr<|dnd} nd} |rDqtjvrP| t| |n| | |} |rctddt | }ntjvrv| t|tj |n | |tj |}t tj |vrt |tj ntj| | |gitj|vr|tjnigtjvr| t|tj|n | |tj|g|tjg|tj|vrtjndtj|vr|tjndfd d |Dd d }|r|| |qd | iS)z!Maps a batch of data to episodes.cSs|SrUr!)rIspacer!r!r3convertlsz3OfflinePreLearner._map_to_episodes..convertrNcSsd|S)Nrr!)xr!r!r3sz4OfflinePreLearner._map_to_episodes..rFcsLi|]"\}}|vr$|vr$|dvr||vrt|n|gqS)rrtypevaluesrr:kvirErDr!r3 s   z6OfflinePreLearner._map_to_episodes..r rXagent_idZ observationsinfosactionsZrewards terminated truncatedZextra_model_outputsZlen_lookback_bufferrF)r enumerater OBSAGENT_IDrtreeZ map_structurecopydeepcopyNEXT_OBSr EPS_IDstruuiduuid4hexINFOSACTIONSREWARDS TERMINATEDS TRUNCATEDSitemsrCappend)r*r5rDrCrEr`r&r'r rbrFobsrpZ unpacked_obsZunpacked_next_obsepisoder!rlr3rNSs          ; 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!)r:rm)rr!r3r>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|qSrergrirlr!r3rnms  zBOfflinePreLearner._map_sample_batch_to_episode..rorF)rur rv isinstancer}rrangeshapenpndarraytolist TypeErrorrfr{rrrr r|r~rrrlenrrrrC) r*r5rDrCrErFrprtrsrr!)rmrErrDr3rLs           . z.OfflinePreLearner._map_sample_batch_to_episoderU)$__name__ __module__ __qualname____doc__rrrgymZSpacer rrrr}r4rrrrQpropertyrr.r/rr rrGboolrO staticmethodrMrlistrrNrLr!r!r!r3r6s  :(   &        r):ryZ gymnasiumrloggingnumpyrrxr~typingrrrrrrrr Zray.rllib.core.columnsr Z(ray.rllib.core.rl_module.multi_rl_moduler r Z"ray.rllib.env.single_agent_episoder Zray.rllib.utilsrZray.rllib.utils.annotationsrrZray.rllib.utils.compressionrZ,ray.rllib.utils.replay_buffers.replay_bufferrZ"ray.rllib.utils.spaces.space_utilsrZray.rllib.utils.typingrrZray.util.annotationsrZ%ray.rllib.algorithms.algorithm_configrr|rwZ MODULE_IDrvrrrr{rrTrM getLoggerrloggerrr!r!r!r3sJ(