feat(video): support video inputs in vLLM responses (#2324)

## Summary

Add video-input support to the NeMo Gym Responses API and vLLM model
backend for multimodal RL training.

  - Accept `input_video` and `video_url` content parts.
- Convert Responses API video inputs into vLLM-compatible chat content.
  - Reject video parts that do not contain a valid URL.
- Prefer prompt and generation token IDs returned by the generation
request.
- Preserve multimodal processor arguments when falling back to the
tokenize endpoint.
- Validate that generation token IDs and log probabilities have matching
lengths.
- Propagate token IDs, log probabilities, and optional training metadata
through Gym responses.
- Ensure component processes load their owning Gym checkout instead of a
stale container installation.

  ## Motivation

Video GRPO requires Gym to preserve video content and the exact prompt
and generation tokenization used by vLLM.

Using token IDs produced by a separate or differently configured
tokenization request can cause rollout and policy log probabilities to
reference different token sequences, resulting in
  incorrect training data and elevated TMPE.

  ## Validation

  - Added coverage for `input_video` and `video_url` conversion.
  - Added coverage for missing video URLs.
  - Added coverage for native vLLM prompt and generation token IDs.
- Added coverage for the tokenize fallback and multimodal processor
arguments.
  - Added coverage for token-ID and log-probability propagation.
- Added coverage ensuring component processes prefer the owning Gym
checkout.
- Exercised through downstream synchronous and asynchronous NeMo RL
video GRPO training.

  ## Dependencies

  None.

  ## Related integrations

- NVIDIA-NeMo/Megatron-Bridge#5304 `feat(video): enable canonical
Nemotron Omni training for V2 MoE checkpoints`
- NVIDIA-NeMo/RL#3500 `feat(video): add Gym support for sync and async
GRPO`

---------

Signed-off-by: Ehsan Hosseini Asl <ehsan.hosseiniasl@gmail.com>

Author: ehosseiniasl

nemo_gym/openai_utils.py

@@ -82,15 +82,13 @@
 from openai.types.responses.response_function_call_output_item_list_param import (
     ResponseFunctionCallOutputItemListParam,
 )
+from openai.types.responses.response_input_content_param import ResponseInputContentParam
 from openai.types.responses.response_input_item import (
     ComputerCallOutput,
     LocalShellCallOutput,
     McpApprovalResponse,
     ResponseCustomToolCallOutput,
 )
-from openai.types.responses.response_input_param import (
-    ResponseInputMessageContentListParam,
-)
 from openai.types.responses.response_output_item import (
     ImageGenerationCall,
     LocalShellCall,
@@ -224,14 +222,46 @@ class NeMoGymResponseOutputMessage(BaseModel):
     type: Literal["message"] = "message"
 
 
+class NeMoGymInputVideoPart(TypedDict, total=False):
+    """Video content accepted by the Responses-compatible Gym API."""
+
+    type: Required[Literal["input_video"]]
+    video_url: Union[str, Dict[str, Any]]
+    video: Union[str, Dict[str, Any]]
+
+
+def _validate_input_video_part(value: Any) -> Any:
+    """Require one non-empty source without changing the TypedDict representation."""
+
+    if not isinstance(value, dict) or value.get("type") != "input_video":
+        return value
+
+    source_keys = [key for key in ("video_url", "video") if key in value]
+    if len(source_keys) != 1:
+        raise ValueError("input_video requires exactly one of video_url or video")
+
+    source = value[source_keys[0]]
+    url = source.get("url") if isinstance(source, dict) else source
+    if not isinstance(url, str) or not url.strip():
+        raise ValueError(f"input_video.{source_keys[0]} must contain a non-empty URL")
+    return value
+
+
+NeMoGymResponseInputContentPart: TypeAlias = Annotated[
+    Union[ResponseInputContentParam, NeMoGymInputVideoPart],
+    BeforeValidator(_validate_input_video_part),
+]
+NeMoGymResponseInputContentList: TypeAlias = List[NeMoGymResponseInputContentPart]
+
+
 class NeMoGymEasyInputMessage(BaseModel):
-    content: Union[str, ResponseInputMessageContentListParam]
+    content: Union[str, NeMoGymResponseInputContentList]
     role: Literal["user", "assistant", "system", "developer"]
     type: Literal["message"] = "message"
 
 
 class NeMoGymMessage(BaseModel):
-    content: ResponseInputMessageContentListParam
+    content: NeMoGymResponseInputContentList
     role: Literal["user", "system", "developer"]
     status: Literal["in_progress", "completed", "incomplete"] = "completed"
     type: Literal["message"] = "message"
@@ -678,11 +708,17 @@ class NeMoGymChatCompletionContentPartFileParam(ChatCompletionContentPartFilePar
     pass
 
 
+class NeMoGymChatCompletionContentPartVideoUrlParam(TypedDict, total=False):
+    video_url: Required[Union[str, Dict[str, Any]]]
+    type: Required[Literal["video_url"]]
+
+
 NeMoGymChatCompletionContentPartParam = Union[
     NeMoGymChatCompletionContentPartTextParam,
     NeMoGymChatCompletionContentPartImageParam,
     NeMoGymChatCompletionContentPartInputAudioParam,
     NeMoGymChatCompletionContentPartFileParam,
+    NeMoGymChatCompletionContentPartVideoUrlParam,
 ]
 
 

nemo_gym/responses_converter.py

@@ -319,10 +319,22 @@ def _format_message(
                         converted_parts.append({"type": "text", "text": part_param["text"]})
                     case "input_image":
                         image_url = part_param.get("image_url", "")
+                        if isinstance(image_url, dict):
+                            image_url = image_url.get("url", "")
+                        if not image_url:
+                            raise ValueError(f"{part_param['type']} requires a non-empty image_url")
                         detail = part_param.get("detail", "auto")
                         converted_parts.append(
                             {"type": "image_url", "image_url": {"url": image_url, "detail": detail}}
                         )
+                    case "input_video":
+                        source_key = "video_url" if "video_url" in part_param else "video"
+                        video_url = part_param[source_key]
+                        if isinstance(video_url, dict):
+                            video_url = video_url.get("url", "")
+                        if not video_url:
+                            raise ValueError(f"input_video.{source_key} requires a non-empty URL")
+                        converted_parts.append({"type": "video_url", "video_url": {"url": video_url}})
                     case _:
                         raise NotImplementedError(f"Unsupported part param type: {part_param['type']}")
             content = converted_parts

tests/unit_tests/test_responses_converter.py

@@ -244,7 +244,9 @@ def test_responses_to_chat_completion_no_instructions_adds_no_message(converter:
     assert [m["role"] for m in params.messages] == ["user"]
 
 
-def test_responses_to_chat_completion_input_image_part(converter: ResponsesConverter):
+def test_responses_to_chat_completion_input_image_part(
+    converter: ResponsesConverter,
+):
     params = converter.responses_to_chat_completion_create_params(
         NeMoGymResponseCreateParamsNonStreaming(
             input=[
@@ -264,6 +266,141 @@ def test_responses_to_chat_completion_input_image_part(converter: ResponsesConve
     assert {"type": "image_url", "image_url": {"url": "http://img", "detail": "high"}} in parts
 
 
+def test_responses_to_chat_completion_empty_image_url_raises(
+    converter: ResponsesConverter,
+):
+    with pytest.raises(ValueError, match="requires a non-empty image_url"):
+        converter._format_message(
+            {
+                "role": "user",
+                "content": [{"type": "input_image", "image_url": ""}],
+            },
+            ResponsesConverterState(return_token_id_information=False),
+        )
+
+
+@pytest.mark.parametrize(
+    ("video_field", "video_url"),
+    [
+        ("video_url", "file:///videos/example.mp4"),
+        ("video_url", {"url": "https://example.com/video.mp4"}),
+        ("video", "file:///videos/example.mp4"),
+    ],
+)
+def test_responses_to_chat_completion_video_part(
+    converter: ResponsesConverter,
+    video_field: str,
+    video_url: object,
+):
+    params = converter.responses_to_chat_completion_create_params(
+        NeMoGymResponseCreateParamsNonStreaming(
+            input=[
+                {
+                    "role": "user",
+                    "type": "message",
+                    "content": [
+                        {"type": "input_text", "text": "what happens?"},
+                        {"type": "input_video", video_field: video_url},
+                    ],
+                }
+            ]
+        )
+    )
+
+    expected_url = video_url["url"] if isinstance(video_url, dict) else video_url
+    assert params.messages[0]["content"] == [
+        {"type": "text", "text": "what happens?"},
+        {"type": "video_url", "video_url": {"url": expected_url}},
+    ]
+
+
+def test_responses_to_chat_completion_empty_video_url_raises(
+    converter: ResponsesConverter,
+):
+    with pytest.raises(ValueError, match="requires a non-empty URL"):
+        converter._format_message(
+            {
+                "role": "user",
+                "content": [{"type": "input_video", "video_url": ""}],
+            },
+            ResponsesConverterState(return_token_id_information=False),
+        )
+
+
+def test_responses_video_schema_preserves_mixed_sdk_items_as_dicts():
+    request = NeMoGymResponseCreateParamsNonStreaming(
+        input=[
+            {
+                "role": "user",
+                "type": "message",
+                "content": [
+                    {"type": "input_file", "file_url": "https://example.com/context.txt"},
+                    {"type": "input_video", "video_url": "https://example.com/video.mp4"},
+                ],
+            }
+        ]
+    )
+
+    content = request.input[0].content
+    assert isinstance(content[0], dict)
+    assert isinstance(content[1], dict)
+    assert content[0]["type"] == "input_file"
+    assert content[1]["type"] == "input_video"
+
+
+@pytest.mark.parametrize(
+    "video_part",
+    [
+        {"type": "input_video"},
+        {"type": "input_video", "video_url": "", "video": "https://example.com/video.mp4"},
+        {
+            "type": "input_video",
+            "video_url": "https://example.com/a.mp4",
+            "video": "https://example.com/b.mp4",
+        },
+        {"type": "input_video", "video_url": {"url": ""}},
+    ],
+)
+def test_responses_video_schema_requires_exactly_one_nonempty_source(video_part: dict):
+    with pytest.raises(ValueError, match="exactly one|non-empty URL"):
+        NeMoGymResponseCreateParamsNonStreaming(input=[{"role": "user", "type": "message", "content": [video_part]}])
+
+
+def test_responses_schema_rejects_chat_style_media_aliases():
+    with pytest.raises(ValueError):
+        NeMoGymResponseCreateParamsNonStreaming(
+            input=[
+                {
+                    "role": "user",
+                    "type": "message",
+                    "content": [{"type": "video_url", "video_url": "https://example.com/video.mp4"}],
+                }
+            ]
+        )
+
+
+def test_chat_schema_accepts_only_canonical_video_url():
+    request = NeMoGymChatCompletionCreateParamsNonStreaming(
+        messages=[
+            {
+                "role": "user",
+                "content": [{"type": "video_url", "video_url": {"url": "https://example.com/video.mp4"}}],
+            }
+        ]
+    )
+    assert request.messages[0]["content"][0]["type"] == "video_url"
+
+    with pytest.raises(ValueError):
+        NeMoGymChatCompletionCreateParamsNonStreaming(
+            messages=[
+                {
+                    "role": "user",
+                    "content": [{"type": "input_video", "video_url": "https://example.com/video.mp4"}],
+                }
+            ]
+        )
+
+
 def test_responses_to_chat_completion_unsupported_part_raises(converter: ResponsesConverter):
     # Exercise the converter directly with an unsupported content part type. A raw
     # ResponseCreateParams would reject this at schema-validation time, so we call the