[Serve][4/4] Add build_openai_applications for multi-model direct streaming

doc/source/serve/api/index.md

@@ -507,6 +507,7 @@ Content-Type: application/json
 
    serve.llm.build_llm_deployment
    serve.llm.build_openai_app
+   serve.llm.build_openai_router_app
 ```
 
 ### Configs

python/ray/llm/_internal/serve/core/ingress/builder.py

@@ -1,6 +1,6 @@
 import os
 import pprint
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Mapping, Optional, Type, Union
 
 from pydantic import Field, field_validator, model_validator
 
@@ -25,6 +25,7 @@
 from ray.llm._internal.serve.routing_policies.kv_aware.kv_aware_router import (
     is_kv_aware,
 )
+from ray.serve._private.constants import RAY_SERVE_ENABLE_HA_PROXY
 from ray.serve.config import RequestRouterConfig
 from ray.serve.deployment import Application
 from ray.serve.experimental.round_robin_router import RoundRobinRouter
@@ -293,3 +294,52 @@ def build_openai_app(builder_config: dict) -> Application:
         lora_paths=lora_paths,
         **ingress_cls_config.ingress_extra_kwargs,
     )
+
+
+def _validate_model_applications(
+    model_applications: Mapping[str, str],
+) -> Dict[str, str]:
+    if not isinstance(model_applications, Mapping):
+        raise TypeError(
+            "model_applications must be a mapping of model IDs to app names."
+        )
+
+    result = dict(model_applications)
+    if not result:
+        raise ValueError("model_applications must contain at least one model.")
+    for model_id, application_name in result.items():
+        if not isinstance(model_id, str) or not model_id:
+            raise ValueError("Model IDs must be nonempty strings.")
+        if not isinstance(application_name, str) or not application_name:
+            raise ValueError("Application names must be nonempty strings.")
+    if len(set(result.values())) != len(result):
+        raise ValueError("Each model must reference a different Serve application.")
+    return result
+
+
+def build_openai_router_app(
+    model_applications: Mapping[str, str],
+) -> Application:
+    """Build a router for independently deployed OpenAI model applications."""
+    if not RAY_SERVE_ENABLE_HA_PROXY:
+        raise ValueError(
+            "build_openai_router_app requires HAProxy. Set "
+            "RAY_SERVE_ENABLE_HA_PROXY=1 on the Ray cluster."
+        )
+    if not RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING:
+        raise ValueError(
+            "build_openai_router_app requires direct streaming. Set "
+            "RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING=1 on the Ray cluster."
+        )
+
+    from ray.llm._internal.serve.core.ingress.applications import RouterApplication
+
+    models = _validate_model_applications(model_applications)
+    return (
+        serve.deployment(
+            RouterApplication,
+            **RouterApplication.get_deployment_options(),
+        )
+        .bind(model_applications=models)
+        ._as_router_application()
+    )

python/ray/llm/tests/BUILD.bazel

@@ -106,6 +106,7 @@ py_test_module_list(
     files = [
         "serve/cpu/deployments/data_parallel/test_dp_direct_streaming.py",
         "serve/cpu/deployments/prefill_decode_disagg/test_pd_direct_streaming.py",
+        "serve/cpu/deployments/routers/test_openai_router_app_direct_streaming.py",
         "serve/cpu/deployments/routers/test_router_direct_streaming.py",
     ],
     tags = [

python/ray/llm/tests/serve/cpu/deployments/routers/test_builder_openai_router_app.py

@@ -0,0 +1,71 @@
+import sys
+
+import pytest
+
+from ray.llm._internal.serve.core.ingress import builder as builder_module
+from ray.llm._internal.serve.core.ingress.applications import RouterApplication
+from ray.llm._internal.serve.core.ingress.builder import build_openai_router_app
+from ray.serve._private.build_app import build_app
+
+
+@pytest.fixture(autouse=True)
+def _direct_streaming(monkeypatch):
+    monkeypatch.setattr(builder_module, "RAY_SERVE_ENABLE_HA_PROXY", True)
+    monkeypatch.setattr(builder_module, "RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING", True)
+    monkeypatch.setattr("ray.serve._private.build_app.RAY_SERVE_ENABLE_HA_PROXY", True)
+
+
+def test_builds_marked_router_application():
+    model_applications = {
+        "model-a": "llm-model-a",
+        "org/model-b": "llm-model-b",
+    }
+
+    app = build_openai_router_app(model_applications)
+
+    assert app._bound_deployment.func_or_class is RouterApplication
+    assert app._bound_deployment.init_kwargs == {
+        "model_applications": model_applications
+    }
+    assert app._is_router_application
+    assert app._ingress_request_router is None
+    built = build_app(app, name="main", route_prefix="/")
+    assert built.is_router_application
+
+
+@pytest.mark.parametrize(
+    "model_applications, error, match",
+    [
+        ({}, ValueError, "at least one"),
+        ({"": "app"}, ValueError, "Model IDs"),
+        ({1: "app"}, ValueError, "Model IDs"),
+        ({"model": ""}, ValueError, "Application names"),
+        ({"model": 1}, ValueError, "Application names"),
+        ({"a": "same", "b": "same"}, ValueError, "different"),
+        (["not", "a", "mapping"], TypeError, "must be a mapping"),
+    ],
+)
+def test_validation(model_applications, error, match):
+    with pytest.raises(error, match=match):
+        build_openai_router_app(model_applications)
+
+
+@pytest.mark.parametrize(
+    "flag", ["RAY_SERVE_ENABLE_HA_PROXY", "RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING"]
+)
+def test_requires_direct_streaming_haproxy(monkeypatch, flag):
+    monkeypatch.setattr(builder_module, flag, False)
+
+    with pytest.raises(ValueError, match=flag):
+        build_openai_router_app({"model": "app"})
+
+
+def test_public_builder():
+    from ray.serve.llm import build_openai_router_app as public_builder
+
+    app = public_builder({"model": "app"})
+    assert app._bound_deployment.func_or_class is RouterApplication
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(["-v", __file__]))