[reward] feat: deterministic reward inference and VLM reward model support

tests/reward_loop/test_visual_reward_manager.py

@@ -108,6 +108,63 @@ def test_reward_model_genrm():
     ray.shutdown()
 
 
+def test_deterministic_reward_reproducibility():
+    ray.init(
+        runtime_env={
+            "env_vars": {
+                "TOKENIZERS_PARALLELISM": "true",
+                "NCCL_DEBUG": "WARN",
+                "VLLM_LOGGING_LEVEL": "INFO",
+                "VLLM_USE_V1": "1",
+            }
+        }
+    )
+    with initialize_config_dir(config_dir=os.path.abspath("verl_omni/trainer/config")):
+        config = compose(config_name="diffusion_trainer")
+
+    rollout_model_name = os.path.expanduser("~/models/tiny-random/Qwen-Image")
+    reward_model_name = os.path.expanduser("~/models/tiny-random/qwen3-vl")
+    reward_model_gpus, tp_size = resolve_reward_loop_gpu_topology()
+
+    config.actor_rollout_ref.model.path = rollout_model_name
+    config.actor_rollout_ref.model.tokenizer_path = os.path.join(rollout_model_name, "tokenizer")
+    config.reward.custom_reward_function.path = "verl_omni/utils/reward_score/genrm_ocr.py"
+    config.reward.custom_reward_function.name = "compute_score_ocr"
+    config.reward.num_workers = 1
+    config.reward.reward_model.enable = True
+    config.reward.reward_model.enable_resource_pool = True
+    config.reward.reward_model.n_gpus_per_node = reward_model_gpus
+    config.reward.reward_model.nnodes = 1
+    config.reward.reward_model.model_path = reward_model_name
+    config.reward.reward_model.rollout.name = os.getenv("ROLLOUT_NAME", "vllm")
+    config.reward.reward_model.rollout.gpu_memory_utilization = 0.9
+    config.reward.reward_model.rollout.tensor_model_parallel_size = tp_size
+    config.reward.reward_model.rollout.skip_tokenizer_init = False
+    config.reward.reward_model.rollout.prompt_length = 2048
+    config.reward.reward_model.rollout.response_length = 32
+    config.reward.reward_model.deterministic = True
+
+    reward_loop_manager = RewardLoopManager(config)
+
+    rollout_tokenizer = hf_tokenizer(config.actor_rollout_ref.model.tokenizer_path)
+    torch.manual_seed(42)
+    data = create_data_samples(rollout_tokenizer)
+
+    run1 = reward_loop_manager.compute_rm_score(data)
+    run2 = reward_loop_manager.compute_rm_score(data)
+
+    for o1, o2 in zip(run1, run2, strict=False):
+        assert torch.equal(o1.batch["rm_scores"], o2.batch["rm_scores"]), (
+            f"rm_scores differ: {o1.batch['rm_scores']} vs {o2.batch['rm_scores']}"
+        )
+        assert o1.non_tensor_batch["genrm_response"] == o2.non_tensor_batch["genrm_response"], (
+            f"genrm_response differs: "
+            f"{o1.non_tensor_batch['genrm_response']} vs {o2.non_tensor_batch['genrm_response']}"
+        )
+
+    ray.shutdown()
+
+
 def test_rule_reward():
     ray.init(
         runtime_env={

verl_omni/reward_loop/reward_manager/visual.py

@@ -58,11 +58,23 @@ async def run_single(self, data: DataProto) -> dict:
         extra_info["num_turns"] = num_turns
         extra_info["rollout_reward_scores"] = rollout_reward_scores
 
+        rm_rollout = self.config.reward.reward_model.rollout
+        sampling_params = {
+            "temperature": rm_rollout.temperature,
+            "do_sample": rm_rollout.do_sample,
+            "top_k": rm_rollout.top_k,
+            "top_p": rm_rollout.top_p,
+        }
+        deterministic = self.config.reward.reward_model.get("deterministic", False)
+        if deterministic:
+            sampling_params["seed"] = self.config.reward.reward_model.get("seed", 42)
+
         extra_reward_kwargs = (
             {
                 "reward_router_address": self.reward_router_address,
                 "reward_model_tokenizer": self.reward_model_tokenizer,
                 "model_name": self.config.reward.reward_model.model_path,
+                "sampling_params": sampling_params,
             }
             if self.reward_router_address is not None
             else {}

verl_omni/trainer/config/_generated_diffusion_trainer.yaml

@@ -356,6 +356,8 @@ reward:
       path: pkg://verl_omni.reward_loop.reward_manager
   reward_model:
     enable: false
+    deterministic: false
+    seed: 42
     enable_resource_pool: false
     n_gpus_per_node: 8
     nnodes: 0

verl_omni/trainer/config/reward/reward.yaml

@@ -32,6 +32,8 @@ reward_manager:
 # support both discriminative and generative models
 reward_model:
   enable: False
+  deterministic: false
+  seed: 42
 
   # Whether to deploy the model to a separate resource pool.
   # If true, n_gpus_per_node & nnodes will be used to determine the resource node.

verl_omni/trainer/diffusion/ray_diffusion_trainer.py

@@ -632,6 +632,78 @@ def _init_colocated_workers(self):
 
     def _init_online_rollout_stack(self, actor_rollout_resource_pool):
         """Initialize rollout, reward, and checkpoint engines (online sampling only)."""
+        # Deterministic reward model inference when `deterministic=true`
+        from verl.workers.rollout.vllm_rollout.vllm_async_server import vLLMHttpServer, vLLMReplica
+
+        class _DeterministicRMHttpServer(vLLMHttpServer):
+            """vLLMHttpServer subclass that applies full determinism in _post_init for RM actors."""
+
+            def _post_init(self, cuda_visible_devices: str) -> None:
+                seed = int(os.environ.get("VERL_OMNI_RM_SEED", "42"))
+                from verl.workers.engine.utils import enable_full_determinism
+
+                enable_full_determinism(seed)
+                super()._post_init(cuda_visible_devices)
+
+        deterministic = self.config.reward.reward_model.get("deterministic", False)
+        seed = self.config.reward.reward_model.get("seed", 42)
+        if deterministic and self.use_rm:
+            # Monkey-patch vLLMReplica to inject deterministic env vars into RM actor
+            # runtime_env and swap server_class. The is_reward_model guard ensures
+            # rollout actors are unaffected.
+            if not hasattr(vLLMReplica, "_reward_deterministic"):
+                vLLMReplica._reward_deterministic = False
+                _original_launch_servers = vLLMReplica.launch_servers
+
+                async def _deterministic_launch_servers(self):
+                    if self.is_reward_model and self._reward_deterministic:
+                        original_server_class = self.server_class
+
+                        class _DeterministicServerProxy:
+                            """Inject PYTHONHASHSEED + seed signal into runtime_env and swap to deterministic server."""
+
+                            def __init__(self, deterministic_server_class, rm_seed):
+                                self.original = deterministic_server_class
+                                self._rm_seed = rm_seed
+
+                            def options(self, **kwargs):
+                                runtime_env = kwargs.setdefault("runtime_env", {})
+                                env_vars = runtime_env.setdefault("env_vars", {})
+                                env_vars["PYTHONHASHSEED"] = str(self._rm_seed)
+                                env_vars["VERL_OMNI_RM_SEED"] = str(self._rm_seed)
+                                return self.original.options(**kwargs)
+
+                            def __getattr__(self, name):
+                                return getattr(self.original, name)
+
+                        self.server_class = _DeterministicServerProxy(
+                            ray.remote(_DeterministicRMHttpServer), self._rm_seed
+                        )
+                        try:
+                            await _original_launch_servers(self)
+                        finally:
+                            self.server_class = original_server_class
+                    else:
+                        await _original_launch_servers(self)
+
+                vLLMReplica.launch_servers = _deterministic_launch_servers
+
+            vLLMReplica._reward_deterministic = True
+            vLLMReplica._rm_seed = seed
+        else:
+            if hasattr(vLLMReplica, "_reward_deterministic"):
+                vLLMReplica._reward_deterministic = False
+
+        # Auto-set VLM scoring function when RM enabled + no custom_reward_function provided
+        if self.use_rm:
+            custom_fn_cfg = self.config.reward.custom_reward_function
+            if custom_fn_cfg.path is None:
+                from omegaconf import OmegaConf
+
+                with OmegaConf.open_dict(custom_fn_cfg):
+                    custom_fn_cfg.path = "verl_omni/utils/reward_score/genrm_ocr.py"
+                    custom_fn_cfg.name = "compute_score_ocr"
+
         # create reward loop manager
         from verl.experimental.reward_loop import RewardLoopManager
 

verl_omni/utils/reward_score/genrm_ocr.py

@@ -86,6 +86,7 @@ async def compute_score_ocr(
     reward_router_address: str,
     reward_model_tokenizer: PreTrainedTokenizer = None,
     model_name: Optional[str] = None,
+    sampling_params: dict = None,
 ):
     """Compute an image OCR score via a generative reward model (GRM).
 
@@ -138,11 +139,11 @@ async def compute_score_ocr(
                 ],
             },
         ]
-        # TODO: make sampling params configurable
+        params = sampling_params or DEFAULT_SAMPLING_PARAMS
         chat_complete_request = {
             "messages": messages,
             "model": model_name,
-            **DEFAULT_SAMPLING_PARAMS,
+            **params,
         }
         result = await _chat_complete(
             router_address=reward_router_address,