[Serve][5/5] Enable direct streaming for Ray Serve LLM apps

[eicherseiji]

Why this is needed

This PR introduces the Ray Serve LLM changes: an ingress request router that can pick a model replica, and a backend ASGI app that the chosen replica can serve directly. This PR is where the OpenAI/vLLM serving path plugs into those new Serve interfaces.

Summary

• add LLMRouter as the LLM-side ingress request router that serves /internal/route and returns {"replica_id": ...} for a chosen backend replica
• use the temporary DeploymentHandle._get_request_router() escape hatch so LLMRouter can reuse Serve's existing in-process replica picker instead of re-implementing replica selection logic
• add a late-bound ASGI hook, __serve_build_asgi_app__, so LLMServer can build the vendored vLLM FastAPI app after async engine startup; Serve resolves it during replica ASGI initialization before lifespan startup
• compose the direct-streaming app inside build_openai_app(...); this prototype validates direct streaming as single-model-only for now

# The direct-streaming app shape is selected internally when
# RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING=1.
app = build_openai_app({"llm_configs": [llm_config]})

Stack: #62667 -> #62680 -> #62668 -> #62669 -> this

Test plan

• Added / updated LLM builder unit coverage for the composed direct-streaming app shape
• python -m py_compile on the edited files
• git diff --check
• Repo commit hooks / linters on commit

Benchmark Results

Direct streaming, benchmarked at 17a635aa4f887a8afdf931e65a31a74bf507d400; current PR cleanup tip is 256599ac14ae7322b2222202e3c8d9a25923ae62. vLLM 0.19.0.

Workload: Qwen/Qwen3-0.6B-FP8, 8 H100 replicas, ISL=8000, OSL=50, 4096 requests, request-rate=inf.

Benchmark invocation:

cd /home/ray/default/llm-inference-stack/replica-scaling
python sweep_concurrency.py \
  --backend ray-serve-llm-direct-streaming \
  --label qwen-0.6b-fp8-h100-ray-direct-streaming-seiji-prhead-17a635-rr-8k50-c64-128-256 \
  --concurrency 64 128 256 \
  --isl-osl-pairs 8000:50

Equivalent vllm bench serve workload shape:

vllm bench serve \
  --backend openai \
  --base-url http://127.0.0.1:8000 \
  --model Qwen/Qwen3-0.6B-FP8 \
  --dataset-name random \
  --random-input-len 8000 \
  --random-output-len 50 \
  --num-prompts 4096 \
  --max-concurrency <64|128|256> \
  --request-rate inf \
  --percentile-metrics ttft,tpot,itl,e2el \
  --save-result

App deployment code:

import os

# For cluster runs, set this in the cluster/runtime env before workers start.
# In a local script, set it before importing ray.serve.llm.
os.environ["RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING"] = "1"

from ray import serve
from ray.serve.llm import LLMConfig, LLMServingArgs, ModelLoadingConfig, build_openai_app

llm_config = LLMConfig(
    model_loading_config=ModelLoadingConfig(
        model_id="Qwen/Qwen3-0.6B-FP8",
        model_source="Qwen/Qwen3-0.6B-FP8",
    ),
    engine_kwargs=dict(
        tensor_parallel_size=1,
        max_model_len=10000,
        gpu_memory_utilization=0.95,
        max_num_seqs=128,
        max_num_batched_tokens=16384,
        enable_prefix_caching=False,
        load_format="dummy",
        kv_cache_dtype="fp8",
    ),
    deployment_config=dict(
        autoscaling_config=dict(min_replicas=8, max_replicas=8),
        max_ongoing_requests=8192,
    )
)

app = build_openai_app(LLMServingArgs(llm_configs=[llm_config]))
serve.run(app)

Results:

Max concurrency	Successful	Failed	Req/s	Output tok/s	Mean TTFT (ms)	P99 TTFT (ms)	Mean TPOT (ms)	P99 TPOT (ms)	Mean E2E (ms)	P99 E2E (ms)
64	4096	0	122.20	6110.07	118.66	489.24	8.19	10.09	519.95	841.03
128	4096	0	149.83	7491.40	148.71	746.33	14.28	18.41	848.37	1389.91
256	4096	0	164.24	8211.87	231.20	1395.37	26.81	35.11	1544.79	2400.58

---

Supersedes #62670

[kouroshHakha]

Good shape for a feature-flagged prototype — the architecture is clean and the separation of concerns (LLMRouter handles replica selection, LLMServer serves traffic, builder composes them) makes sense. A few things worth discussing before the flag goes wider:

Direct ASGI path bypasses Serve lifecycle (replica.py:2538): The TODO correctly lists what's missing (backpressure, timeouts, cancellation, tracing, per-route metrics). Can you link this to a tracking issue? Missing cancellation propagation in particular means a client disconnect won't abort the vLLM generation, which can waste GPU time under load.

LLMRouter reimplements routing policy: _ready_replicas, _round_robin_counter, and _pick_replica duplicate logic that already lives in RequestRouter.choose_replicas(). The right shape is to call the existing policy via _get_request_router().choose_replicas(...) and extract backend_http_endpoint — routing policy stays in one place and becomes configurable. When #60865 lands with DeploymentHandle.choose_replica(), this collapses to a single call. Since this is a fully internal change with no API or contract impact, it can land as a follow-up once the flag is validated.

Naming: Consider renaming router.py → llm_ingress_request_router.py and LLMRouter → LLMIngressRequestRouter. The current names are ambiguous alongside Serve's own ray/serve/_private/router.py.

Note

This review was co-written with AI assistance (Claude Code).

[kouroshHakha]

Important (can land as follow-up): _ready_replicas, _cached_replica_signature, _round_robin_counter, _replica_id_sort_key, and _pick_replica reimplement replica selection that already exists in RequestRouter.choose_replicas(). The right shape is to call the existing policy via _get_request_router().choose_replicas(...) and extract backend_http_endpoint from the result — all the custom caching and counter logic goes away, and the routing policy becomes whatever Serve is already configured with rather than always being a bespoke round-robin. When #60865 lands with DeploymentHandle.choose_replica(), this becomes a single call with no internal API access at all. Since this is a fully internal change with no API or contract impact, it can land as a follow-up once the flag is validated.

[kouroshHakha]

Important: num_replicas=1 and max_ongoing_requests=1000 are hardcoded with no user override path. Introduce an IngressRequestRouterConfig (small Pydantic model in LLMServingArgs) with these as fields and appropriate defaults. For now validate that only num_replicas is user-configurable, with an explicit error if autoscaling is set (since LLMRouter doesn't support it yet).

[eicherseiji]

Proposing to defer this until the root cause of ingress request router scaling issue is better understood

[cursor]

build_asgi_app skips vLLM version compatibility checks

Medium Severity

build_asgi_app unconditionally passes supported_tasks to build_app and init_app_state, and omits the vllm_config parameter entirely. The existing start() method guards both with inspect.signature checks for cross-version compatibility and passes vllm_config when the parameter exists. The new method will crash on vLLM versions that don't accept supported_tasks, and will produce an incompletely initialized app state on versions that require vllm_config.

 

^{Reviewed by Cursor Bugbot for commit 363f0c3. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{Reviewed by Cursor Bugbot for commit 120dab7. Configure here.}

[cursor]

Stale cache when replica dies via in-place dict mutation

High Severity

_ready_endpoints uses id(curr_replicas) == self._cached_dict_id as a fast "nothing changed" shortcut. However, RequestRouter.on_replica_actor_died() mutates self._replicas in place via pop(), so the dict object identity remains unchanged after a replica is removed. The cache continues returning the stale endpoint list that includes the dead replica, routing requests to it until the next full controller broadcast replaces the dict wholesale.

 

^{Reviewed by Cursor Bugbot for commit 120dab7. Configure here.}

[kouroshHakha]

Approving. We need to also follow up with a release test. @eicherseiji

[eicherseiji]

llm-gpu failures unrelated

 Test: //doc:.../working-with-llms/omni_audio_example                            
  Type: Ray Data batch inference (Qwen2-Omni audio)                            
  Time: TIMEOUT @ 911s                                                            
  ────────────────────────────────────────                                     
  Test: //doc:.../llm/doc_code/serve/transcription/transcription_example          
  Type: Ray Data batch inference (Whisper)                                        
  Time: TIMEOUT @ 905s                                                         
  ────────────────────────────────────────                                        
  Test: //doc:.../working-with-llms/vlm_image_example                             
  Type: Ray Data batch inference (VLM)                                            
  Time: TIMEOUT @ 911s               

"Cluster resources are not enough to run any task from TaskPoolMapOperator[ReadParquet]. The job may hang forever unless the cluster scales up"
https://buildkite.com/ray-project/premerge/builds/66007#019e002b-23ef-4854-a61e-5e31bed60dad