[Misc] Added validation for Cohere /v2/embed input field exclusivity (#45640)

Signed-off-by: Taneem Ibrahim <taneem.ibrahim@gmail.com>

tests/entrypoints/pooling/embed/test_io_processor.py

@@ -3,7 +3,7 @@
 """Unit tests for EmbedIOProcessor."""
 
 import pytest
-from pydantic import TypeAdapter
+from pydantic import TypeAdapter, ValidationError
 
 from vllm import PoolingParams
 from vllm.entrypoints.pooling.embed.io_processor import EmbedIOProcessor
@@ -105,6 +105,59 @@ class TestEmbeddingRequestParsing:
         assert request.chat_template_kwargs == {"instruction": "Represent the query: "}
 
 
+class TestCohereEmbedRequestParsing:
+    """Unit tests for Cohere embed request parsing."""
+
+    @pytest.mark.parametrize(
+        "request_body",
+        [
+            {"model": "test"},
+            {"model": "test", "texts": ["hello"], "images": ["image-uri"]},
+            {
+                "model": "test",
+                "texts": ["hello"],
+                "inputs": [
+                    {"content": [{"type": "text", "text": "hello"}]},
+                ],
+            },
+            {
+                "model": "test",
+                "images": ["image-uri"],
+                "inputs": [
+                    {"content": [{"type": "text", "text": "hello"}]},
+                ],
+            },
+            {"model": "test", "texts": []},
+            {"model": "test", "images": []},
+            {"model": "test", "inputs": []},
+        ],
+    )
+    def test_rejects_invalid_input_field_combinations(self, request_body):
+        with pytest.raises(
+            ValidationError,
+            match="Exactly one of texts, images, or inputs must be provided",
+        ):
+            CohereEmbedRequest(**request_body)
+
+    @pytest.mark.parametrize(
+        "request_body",
+        [
+            {"model": "test", "texts": ["hello"]},
+            {"model": "test", "images": ["image-uri"]},
+            {
+                "model": "test",
+                "inputs": [
+                    {"content": [{"type": "text", "text": "hello"}]},
+                ],
+            },
+        ],
+    )
+    def test_accepts_exactly_one_non_empty_input_field(self, request_body):
+        request = CohereEmbedRequest(**request_body)
+
+        assert request.model == "test"
+
+
 class TestResolveTruncation:
     """Unit tests for EmbedIOProcessor._resolve_cohere_truncation."""
 

vllm/entrypoints/pooling/embed/protocol.py

@@ -224,6 +224,17 @@ class CohereEmbedRequest(BaseModel):
     max_tokens: int | None = None
     priority: int = 0
 
+    @model_validator(mode="after")
+    def validate_input_fields(self):
+        input_fields = (self.texts, self.images, self.inputs)
+        provided_fields = [field for field in input_fields if field is not None]
+        if len(provided_fields) != 1 or not provided_fields[0]:
+            raise ValueError(
+                "Exactly one of texts, images, or inputs must be provided, "
+                "and it must be non-empty"
+            )
+        return self
+
 
 # ---------------------------------------------------------------------------
 # Cohere /v2/embed — response models