fix: fix test

Author: Irvingwangjr

tests/workers/rollout/test_sglang_async_rollout_sf_tools.py

@@ -34,7 +34,7 @@
 from verl.workers.rollout.schemas import AsyncRolloutRequest, AsyncRolloutRequestStateEnum, Message
 from verl.workers.rollout.sglang_rollout.async_sglang_rollout import AsyncSGLangRollout
 
-sandbox_url = ""
+sandbox_url = "https://sd04qmtd8e6v9i08l9l00.apigateway-cn-beijing.volceapi.com/run_code"
 
 
 def get_sandbox_fusion_data():
@@ -185,27 +185,28 @@ def test_rollout_req_creation(self, mock_env, mock_engine, mock_sampling, sandbo
         req_list = rollout._preprocess_prompt_to_async_rollout_requests(sandbox_data_proto, n=1)
         assert len(req_list) == 1
         assert req_list[0].state == AsyncRolloutRequestStateEnum.PENDING
-        assert req_list[0].tools == [
-            OpenAIFunctionToolSchema(
-                type="function",
-                function=OpenAIFunctionSchema(
-                    name="code_interpreter",
-                    description="A tool for executing code.",
-                    parameters=OpenAIFunctionParametersSchema(
-                        type="object",
-                        properties={
-                            "code": OpenAIFunctionPropertySchema(
-                                type="string",
-                                description="The code to execute.",
-                                enum=None,
-                            )
-                        },
-                        required=["code"],
-                    ),
-                    strict=False,
+        assert len(req_list[0].tools) == 1
+        print("------------")
+        print(type(req_list[0].tools[0]))
+        assert req_list[0].tools[0] == OpenAIFunctionToolSchema(
+            type="function",
+            function=OpenAIFunctionSchema(
+                name="code_interpreter",
+                description="A tool for executing code.",
+                parameters=OpenAIFunctionParametersSchema(
+                    type="object",
+                    properties={
+                        "code": OpenAIFunctionPropertySchema(
+                            type="string",
+                            description="The code to execute.",
+                            enum=None,
+                        )
+                    },
+                    required=["code"],
                 ),
-            )
-        ]
+                strict=False,
+            ),
+        )
 
     @patch.object(AsyncSGLangRollout, "_init_distributed_env", return_value=None)
     @patch.object(AsyncSGLangRollout, "_init_inference_engine", return_value=None)

verl/tools/base_tool.py

@@ -12,7 +12,7 @@
 # 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.
-from typing import Any, Optional, Tuple
+from typing import Any, Optional, Protocol, Tuple, runtime_checkable
 from uuid import uuid4
 
 from .schemas import OpenAIFunctionToolSchema
@@ -84,3 +84,9 @@ async def release(self, instance_id: str, **kwargs) -> None:
             instance_id: The instance id of the tool.
         """
         pass
+
+
+@runtime_checkable
+class DatasetIrrelevantTool(Protocol):
+    def dataset_irrelevant(self) -> bool:
+        return False

verl/tools/sandbox_fusion_tools.py

@@ -26,7 +26,7 @@
 
 from verl.utils.reward_score.sandbox_fusion.utils import _process_single_case
 
-from .base_tool import BaseTool
+from .base_tool import BaseTool, DatasetIrrelevantTool
 from .schemas import OpenAIFunctionToolSchema
 
 logger = logging.getLogger(__name__)
@@ -93,7 +93,7 @@ def init_execution_pool(num_workers: int, enable_global_rate_limit=True, rate_li
         # return ray.util.multiprocessing.Pool(processes=num_workers)
 
 
-class SandboxFusionTool(BaseTool):
+class SandboxFusionTool(BaseTool, DatasetIrrelevantTool):
     """A tool for executing the code using sanbox fusion image.
 
     - `to_openai_function_tool_schema`: return the tool schema in OpenAI format.
@@ -141,6 +141,9 @@ def __init__(self, config: dict, tool_schema: OpenAIFunctionToolSchema):
     def get_openai_tool_schema(self) -> OpenAIFunctionToolSchema:
         return self.tool_schema
 
+    def dataset_irrelevant(self):
+        return True
+
     async def create(self, instance_id: Optional[str] = None, ground_truth: Optional[str] = None, **kwargs) -> str:
         if instance_id is None:
             instance_id = str(uuid4())

verl/workers/rollout/sglang_rollout/async_sglang_rollout.py

@@ -39,7 +39,7 @@
 
 from verl import DataProto
 from verl.third_party.sglang import parallel_state as sglang_ps
-from verl.tools.base_tool import BaseTool
+from verl.tools.base_tool import BaseTool, DatasetIrrelevantTool
 from verl.tools.schemas import OpenAIFunctionCallSchema, OpenAIFunctionParsedSchema, OpenAIFunctionToolCall
 from verl.utils.debug import GPUMemoryLogger
 from verl.utils.model import compute_position_id_with_mask
@@ -121,7 +121,7 @@ def _init_distributed_env(self, device_mesh_cpu, **kwargs):
             os.environ["CUDA_TIMER_STREAM_KAFKA_ENABLE"] = "0"
             os.environ["MEGATRON_IMPORT_TIMERS"] = "0"
             train_tp = kwargs.get("train_tp", None)
-            num_tp_per_train_tp = train_tp // tensor_parallel_size
+            num_tp_per_train_tp = train_tp // self.tensor_parallel_size
             sglang_ps.initialize_parallel_state(
                 tensor_model_parallel_size=self.tensor_parallel_size,
                 num_tp_per_train_tp=num_tp_per_train_tp,
@@ -555,7 +555,7 @@ async def calc_reward_and_release_fn(name: str, tool: BaseTool):
         return _req
 
     async def _handle_engine_call(self, _req: AsyncRolloutRequest, do_sample: bool, is_validate: bool, **kwargs) -> dict:
-        generation_prompt = _req.get_generation_prompt(self.tokenizer)
+        generation_prompt_ids = _req.get_generation_prompt(self.tokenizer)
         max_new_tokens = min(self.config.response_length, self.config.max_model_len - len(generation_prompt_ids) - 1)
         if not do_sample:
             kwargs = dict(
@@ -737,8 +737,10 @@ def _preprocess_prompt_to_async_rollout_requests(self, prompts: DataProto, n: in
                         _tools_kwargs[k] = data_tools_kwargs[k]
                     # add for dataset-irrelevant tools
                     for tool_key in self._tool_map.keys():
-                        _tools_kwargs[tool_key] = {}
-                        _tool_schemas.append(self._tool_map[tool_key].get_openai_tool_schema())
+                        # TODO: redesign this logic
+                        if tool_key not in _tools_kwargs and isinstance(self._tool_map[tool_key], DatasetIrrelevantTool) and self._tool_map[tool_key].dataset_irrelevant():
+                            _tools_kwargs[tool_key] = {}
+                            _tool_schemas.append(self._tool_map[tool_key].get_openai_tool_schema())
                     prompt_with_chat_template = self.tokenizer.apply_chat_template(
                         conversation=raw_prompt,
                         tools=[tool.model_dump() for tool in _tool_schemas],