replace submodule with pip dependencies (#35)

Author: ji-huazhong

.gitmodules

@@ -13,10 +13,10 @@
 # limitations under the License.
 import numpy as np
 import ray
+from transfer_queue import BatchMeta
 
 import verl.experimental.agent_loop.agent_loop as agent_loop
 from verl import DataProto
-from verl.experimental.transfer_queue import BatchMeta
 
 
 class AgentLoopManager(agent_loop.AgentLoopManager):

recipe/transfer_queue/agent_loop.py

@@ -39,16 +39,16 @@
 from torch.utils.data import Dataset, Sampler
 from torchdata.stateful_dataloader import StatefulDataLoader
 from tqdm import tqdm
-
-from verl import DataProto
-from verl.experimental.dataset.sampler import AbstractCurriculumSampler
-from verl.experimental.transfer_queue import (
+from transfer_queue import (
     BatchMeta,
     TransferQueueController,
     TransferQueueStorageSimpleUnit,
     get_placement_group,
     process_zmq_server_info,
 )
+
+from verl import DataProto
+from verl.experimental.dataset.sampler import AbstractCurriculumSampler
 from verl.single_controller.ray import (
     RayClassWithInitArgs,
     RayResourcePool,

recipe/transfer_queue/ray_trainer.py

@@ -24,4 +24,4 @@ fastapi
 latex2sympy2_extended
 math_verify
 tensorboard
-pyzmq
+git+https://github.com/TransferQueue/TransferQueue.git@a2ddb30

requirements.txt

@@ -42,6 +42,7 @@
     "wandb",
     "packaging>=20.0",
     "tensorboard",
+    "TransferQueue @ git+https://github.com/TransferQueue/TransferQueue.git@a2ddb30",
 ]
 
 TEST_REQUIRES = ["pytest", "pre-commit", "py-spy", "pytest-asyncio"]

setup.py

@@ -69,7 +69,8 @@ def init_predefined_execute_mode():
 
 
 def _split_args_kwargs_data_proto(chunks, *args, **kwargs):
-    from verl.experimental.transfer_queue import BatchMeta
+    from transfer_queue import BatchMeta
+
     from verl.protocol import DataProto, DataProtoFuture
 
     splitted_args = []
@@ -134,8 +135,8 @@ def collect_all_to_all(worker_group, output):
 
 def _concat_data_proto_or_future(output: list):
     import ray
+    from transfer_queue import BatchMeta
 
-    from verl.experimental.transfer_queue import BatchMeta
     from verl.protocol import DataProto, DataProtoFuture
 
     # make sure all the elements in output has the same type
@@ -265,8 +266,8 @@ def dispatch_nd_compute_dataproto(dp_rank_mapping: list[int], dp_size, worker_gr
 def collect_nd_compute_dataproto(collect_mask: list[bool], worker_group, output):
     output = collect_nd_compute(collect_mask, worker_group, output)
     import ray
+    from transfer_queue import BatchMeta
 
-    from verl.experimental.transfer_queue import BatchMeta
     from verl.protocol import DataProto
 
     for o in output:

verl/experimental/transfer_queue

@@ -19,12 +19,12 @@
 from typing import Any, Callable
 
 from tensordict import TensorDict
-
-from verl.experimental.transfer_queue import (
+from transfer_queue import (
     AsyncTransferQueueClient,
     BatchMeta,
     ZMQServerInfo,
 )
+
 from verl.protocol import DataProto
 
 _TRANSFER_QUEUE_CLIENT = None