fix issues when running deepeyes using tq

Signed-off-by: jianjunzhong <jianjunzhong@foxmail.com>

Author: jianjunzhong

recipe/deepeyes/deepeyes.py

@@ -179,7 +179,7 @@ def __getitem__(self, item):
         return row_dict
 
 
-def compute_score(data_source: str, solution_str: str, ground_truth: str, extra_info=None) -> float:
+def compute_score(data_source: str, solution_str: str, ground_truth: str, extra_info=None, **kwargs) -> float:
     """
     Compute reward score for model solutions with robust handling of various formats.
 

recipe/transfer_queue/agent_loop.py

@@ -30,12 +30,11 @@ def generate_sequences(self, prompts: BatchMeta) -> BatchMeta:
             BatchMeta: Output batch metadata.
         """
 
-        if self.rm_micro_batch_size and len(prompts) % self.rm_micro_batch_size != 0:
-            raise ValueError(
-                f"The length of prompts {len(prompts)} cannot divide the world size of rm_wg {self.rm_micro_batch_size}"
-            )
         if self.config.actor_rollout_ref.rollout.free_cache_engine:
             self.wake_up()
+        if self.reward_model_manager and self.config.reward_model.rollout.free_cache_engine:
+            self.reward_model_manager.wake_up()
+
         chunkes = prompts.chunk(len(self.agent_loop_workers))
         outputs = ray.get(
             [
@@ -46,6 +45,8 @@ def generate_sequences(self, prompts: BatchMeta) -> BatchMeta:
         output = BatchMeta.concat(outputs)
         if self.config.actor_rollout_ref.rollout.free_cache_engine:
             self.sleep()
+        if self.reward_model_manager and self.config.reward_model.rollout.free_cache_engine:
+            self.reward_model_manager.sleep()
 
         # calculate performance metrics
         metrics = [output.extra_info.pop("metrics") for output in outputs]  # List[List[Dict[str, str]]]

verl/experimental/agent_loop/agent_loop.py

@@ -734,6 +734,11 @@ def _initialize_llm_servers(self):
     def _init_agent_loop_workers(self):
         self.agent_loop_workers = []
         num_workers = self.config.actor_rollout_ref.rollout.agent.num_workers
+        runtime_env = {
+            "env_vars": {
+                "TRANSFER_QUEUE_ENABLE": "1" if self.config.transfer_queue.enable else "0",
+            }
+        }
 
         node_ids = [node["NodeID"] for node in ray.nodes() if node["Alive"] and node["Resources"].get("CPU", 0) > 0]
         for i in range(num_workers):
@@ -745,6 +750,7 @@ def _init_agent_loop_workers(self):
                     scheduling_strategy=ray.util.scheduling_strategies.NodeAffinitySchedulingStrategy(
                         node_id=node_id, soft=True
                     ),
+                    runtime_env=runtime_env,
                 ).remote(self.config, self.server_handles, self.reward_router_address)
             )
 

verl/utils/transferqueue_utils.py

@@ -39,7 +39,7 @@ class BatchMeta:
 
 _TRANSFER_QUEUE_CLIENT = None
 
-is_transferqueue_enabled = os.environ.get("TRANSFER_QUEUE_ENABLE", False)
+is_transferqueue_enabled = os.environ.get("TRANSFER_QUEUE_ENABLE", "0") == "1"
 
 
 def create_transferqueue_client(