feat: add vLLM serving, streaming, and Ray Serve configuration

README.md

@@ -61,6 +61,28 @@ Successful PR CI runs are merged automatically only for trusted same-repository
 and Dependabot. Forks, drafts, and untrusted author associations are deliberately skipped;
 repository branch-protection and review requirements continue to apply.
 
+## Serving backends
+
+`ROUTER_BACKEND=mock` (the default) keeps CI deterministic and GPU-free.
+`ROUTER_BACKEND=vllm` dispatches to a vLLM OpenAI-compatible server; a selected LoRA
+adapter is served by name over the shared base model. An unreachable or failing engine
+returns `502` with retry guidance, and `/readyz` fails while the engine is unhealthy.
+
+Set `"stream": true` to receive OpenAI-compatible `text/event-stream` chunks. Streamed
+results are cached under the same eligibility rules and replayed as chunks on a hit.
+
+### Ray Serve deployment
+
+[`config/ray-serve.yaml`](config/ray-serve.yaml) is generated from the catalog, never
+hand-edited, and verified by a test:
+
+```bash
+python -m llm_router.serving > config/ray-serve.yaml
+```
+
+It carries per-tier autoscaling (latency-sensitive tiers keep a warm replica), GPU pool
+placement, tensor parallelism, prefix caching, quantization, and Multi-LoRA settings.
+
 ## Model registry
 
 [`config/registry.yaml`](config/registry.yaml) is the governed source of truth for what may
@@ -127,6 +149,9 @@ All settings use the `ROUTER_` prefix.
 | `ROUTER_ADMISSION_TIMEOUT_SECONDS` | `0.25` | Time allowed to wait for capacity. |
 | `ROUTER_QUOTA_REQUESTS_PER_MINUTE` | `120` | Per-token sliding-window quota. |
 | `ROUTER_EXTERNAL_FALLBACK_ENABLED` | `false` | Operator gate for external fallback. |
+| `ROUTER_BACKEND` | `mock` | `mock` or `vllm`. |
+| `ROUTER_VLLM_BASE_URL` | `http://127.0.0.1:8001` | vLLM OpenAI-compatible endpoint. |
+| `ROUTER_BACKEND_TIMEOUT_SECONDS` | `60` | Per-request engine timeout. |
 | `ROUTER_REGISTRY_PATH` | `config/registry.yaml` | Governed model catalog; built-in profiles are used if absent. |
 | `ROUTER_ROUTING_POLICY_VERSION` | `v1` | Invalidates router and response caches when changed. |
 | `ROUTER_CACHE_ENABLED` | `true` | Master switch for all cache tiers. |

config/ray-serve.yaml

@@ -0,0 +1,99 @@
+policy_version: v1
+applications:
+- model_id: small-specialist
+  model_revision: mock-small@sha256:dev
+  stage: production
+  model_loading_config:
+    model_id: small-specialist
+    model_source: registry://small-specialist@mock-small@sha256:dev
+    tokenizer: mock-small-tokenizer
+  accelerator_type: nvidia-l4
+  deployment_config:
+    autoscaling_config:
+      min_replicas: 1
+      max_replicas: 4
+      target_ongoing_requests: 8
+      upscale_delay_s: 10
+      downscale_delay_s: 300
+    max_ongoing_requests: 16
+    ray_actor_options:
+      num_gpus: 1
+      resources:
+        gpu_pool_nvidia-l4: 0.001
+  engine_kwargs:
+    max_model_len: 8192
+    tensor_parallel_size: 1
+    enable_prefix_caching: true
+    enable_chunked_prefill: true
+    quantization: awq
+    enable_lora: true
+    max_loras: 3
+    max_lora_rank: 32
+  lora_config:
+    dynamic_lora_loading_path: registry://adapters/small-specialist
+    adapters:
+    - id: claims-extraction-lora
+      revision: claims-lora@sha256:dev
+      domain: claims
+      quantized: true
+      stage: production
+    - id: claims-extraction-lora-next
+      revision: claims-lora@sha256:next
+      domain: claims
+      quantized: false
+      stage: staging
+    - id: support-classification-lora
+      revision: support-lora@sha256:dev
+      domain: support
+      quantized: false
+      stage: production
+- model_id: general-local
+  model_revision: mock-general@sha256:dev
+  stage: production
+  model_loading_config:
+    model_id: general-local
+    model_source: registry://general-local@mock-general@sha256:dev
+    tokenizer: mock-general-tokenizer
+  accelerator_type: nvidia-a10g
+  deployment_config:
+    autoscaling_config:
+      min_replicas: 1
+      max_replicas: 4
+      target_ongoing_requests: 8
+      upscale_delay_s: 10
+      downscale_delay_s: 300
+    max_ongoing_requests: 16
+    ray_actor_options:
+      num_gpus: 1
+      resources:
+        gpu_pool_nvidia-a10g: 0.001
+  engine_kwargs:
+    max_model_len: 32768
+    tensor_parallel_size: 1
+    enable_prefix_caching: true
+    enable_chunked_prefill: true
+- model_id: high-capability
+  model_revision: mock-high@sha256:dev
+  stage: production
+  model_loading_config:
+    model_id: high-capability
+    model_source: registry://high-capability@mock-high@sha256:dev
+    tokenizer: mock-high-tokenizer
+  accelerator_type: nvidia-a100
+  deployment_config:
+    autoscaling_config:
+      min_replicas: 0
+      max_replicas: 2
+      target_ongoing_requests: 8
+      upscale_delay_s: 10
+      downscale_delay_s: 60
+    max_ongoing_requests: 16
+    ray_actor_options:
+      num_gpus: 2
+      resources:
+        gpu_pool_nvidia-a100: 0.001
+  engine_kwargs:
+    max_model_len: 65536
+    tensor_parallel_size: 2
+    enable_prefix_caching: true
+    enable_chunked_prefill: true

pyproject.toml

@@ -10,6 +10,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
   "fastapi>=0.141.1,<1",
+  "httpx>=0.28.1,<1",
   "prometheus-client>=0.26.0,<1",
   "pyyaml>=6.0.2,<7",
   "pydantic-settings>=2.15.0,<3",
@@ -18,7 +19,6 @@ dependencies = [
 
 [project.optional-dependencies]
 dev = [
-  "httpx>=0.28.1,<1",
   "mypy>=2.3.1,<3",
   "pytest>=9.1.1,<10",
   "pytest-asyncio>=1.4.0,<2",

src/llm_router/app.py

@@ -1,21 +1,29 @@
 import hashlib
+import json
 import secrets
 import time
 import uuid
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from pathlib import Path
 
+import httpx
 from fastapi import Depends, FastAPI, Header, HTTPException, Request, Response, status
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 
 from llm_router.admission import (
     AdmissionController,
     AdmissionRejectedError,
     QuotaExceededError,
     SlidingWindowQuota,
 )
-from llm_router.backends import BackendResult, InferenceBackend, MockInferenceBackend
+from llm_router.backends import (
+    BackendResult,
+    BackendUnavailableError,
+    InferenceBackend,
+    MockInferenceBackend,
+    VLLMBackend,
+)
 from llm_router.caching import (
     CachedCompletion,
     CacheStore,
@@ -74,7 +82,18 @@ def create_app(
         runtime_settings.admission_timeout_seconds,
     )
     quota = SlidingWindowQuota(runtime_settings.quota_requests_per_minute)
-    inference_backend = backend or MockInferenceBackend()
+    engine_client = (
+        httpx.AsyncClient() if backend is None and runtime_settings.backend == "vllm" else None
+    )
+    inference_backend: InferenceBackend = backend or (
+        VLLMBackend(
+            base_url=runtime_settings.vllm_base_url,
+            client=engine_client,
+            request_timeout_seconds=runtime_settings.backend_timeout_seconds,
+        )
+        if engine_client is not None
+        else MockInferenceBackend()
+    )
     telemetry = metrics if metrics is not None else Metrics()
     exact_cache: CacheStore = cache_store or InMemoryCacheStore(
         max_entries=runtime_settings.cache_max_entries,
@@ -95,6 +114,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         app.state.ready = True
         yield
         app.state.ready = False
+        if engine_client is not None:
+            await engine_client.aclose()
 
     app = FastAPI(
         title="Local LLM Inference Router",
@@ -136,6 +157,15 @@ async def admission_handler(_: Request, error: AdmissionRejectedError) -> JSONRe
             content={"error": {"message": str(error), "type": "overloaded"}},
         )
 
+    @app.exception_handler(BackendUnavailableError)
+    async def backend_handler(_: Request, error: BackendUnavailableError) -> JSONResponse:
+        telemetry.record_rejection("backend_unavailable")
+        return JSONResponse(
+            status_code=502,
+            headers={"Retry-After": "5"},
+            content={"error": {"message": str(error), "type": "backend_unavailable"}},
+        )
+
     @app.exception_handler(QuotaExceededError)
     async def quota_handler(_: Request, error: QuotaExceededError) -> JSONResponse:
         telemetry.record_rejection("quota_exceeded")
@@ -153,6 +183,8 @@ async def health() -> dict[str, str]:
     async def readiness(request: Request) -> dict[str, str]:
         if not getattr(request.app.state, "ready", False):
             raise HTTPException(status_code=503, detail="not ready")
+        if not await inference_backend.healthy():
+            raise HTTPException(status_code=503, detail="inference backend is unhealthy")
         return {"status": "ready"}
 
     @app.get("/metrics")
@@ -258,6 +290,26 @@ def _completion_response(
             routing=routing,
         )
 
+    def _cached_stream(entry: CachedCompletion, hit_name: str) -> StreamingResponse:
+        completion_id = f"chatcmpl-{uuid.uuid4().hex}"
+        created = int(time.time())
+
+        async def iterator() -> AsyncIterator[str]:
+            yield _chunk(completion_id, created, entry.model_id, delta={"role": "assistant"})
+            yield _chunk(completion_id, created, entry.model_id, delta={"content": entry.text})
+            yield _chunk(completion_id, created, entry.model_id, delta={}, finish_reason="stop")
+            yield "data: [DONE]\n\n"
+
+        return StreamingResponse(
+            iterator(),
+            media_type="text/event-stream",
+            headers={
+                "X-Cache": hit_name,
+                "X-Route-Model": entry.model_id,
+                "X-Route-Revision": entry.model_revision,
+            },
+        )
+
     async def _lookup_cache(
         payload: ChatCompletionRequest, prompt: str, cache_key: str, tenant: str
     ) -> tuple[str, CachedCompletion] | None:
@@ -305,12 +357,75 @@ async def _store_cache(
             scope = semantic_cache.scope(payload, tenant=tenant, model_revision=catalog_version)
             semantic_cache.store(scope, prompt, entry)
 
-    @app.post("/v1/chat/completions", response_model=ChatCompletionResponse)
+    def _route_headers(decision: RouteDecision, cache_state: str) -> dict[str, str]:
+        headers = {
+            "X-Cache": cache_state,
+            "X-Route-Model": decision.profile.id,
+            "X-Route-Revision": decision.profile.revision,
+            "X-Route-Reason": decision.reason,
+        }
+        if decision.adapter_id is not None:
+            headers["X-Route-Adapter"] = decision.adapter_id
+        return headers
+
+    def _chunk(completion_id: str, created: int, model_id: str, **choice: object) -> str:
+        document = {
+            "id": completion_id,
+            "object": "chat.completion.chunk",
+            "created": created,
+            "model": model_id,
+            "choices": [{"index": 0, **choice}],
+        }
+        return f"data: {json.dumps(document)}\n\n"
+
+    async def _stream_completion(
+        payload: ChatCompletionRequest,
+        prompt: str,
+        cache_key: str,
+        subject: str,
+        decision: RouteDecision,
+        started: float,
+        queue_seconds: float,
+    ) -> AsyncIterator[str]:
+        completion_id = f"chatcmpl-{uuid.uuid4().hex}"
+        created = int(time.time())
+        model_id = decision.profile.id
+        collected: list[str] = []
+
+        yield _chunk(completion_id, created, model_id, delta={"role": "assistant"})
+        async for delta in inference_backend.stream(payload, decision):
+            collected.append(delta)
+            yield _chunk(completion_id, created, model_id, delta={"content": delta})
+        yield _chunk(completion_id, created, model_id, delta={}, finish_reason="stop")
+        yield "data: [DONE]\n\n"
+
+        text = "".join(collected)
+        prompt_tokens = max(1, len(prompt) // 4)
+        completion_tokens = max(1, len(text) // 4)
+        telemetry.record_completion(
+            decision,
+            latency_seconds=time.perf_counter() - started,
+            queue_seconds=queue_seconds,
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+        )
+        await _store_cache(
+            payload,
+            prompt,
+            cache_key,
+            subject,
+            decision,
+            BackendResult(
+                text=text, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens
+            ),
+        )
+
+    @app.post("/v1/chat/completions", response_model=None)
     async def chat_completions(
         payload: ChatCompletionRequest,
         response: Response,
         subject: str = Depends(authenticate),
-    ) -> ChatCompletionResponse:
+    ) -> ChatCompletionResponse | StreamingResponse:
         started = time.perf_counter()
         await quota.consume(subject)
         prompt = payload.prompt
@@ -321,6 +436,8 @@ async def chat_completions(
         cached = await _lookup_cache(payload, prompt, cache_key, subject)
         if cached is not None:
             hit_name, entry = cached
+            if payload.stream:
+                return _cached_stream(entry, hit_name)
             response.headers["X-Cache"] = hit_name
             response.headers["X-Route-Model"] = entry.model_id
             response.headers["X-Route-Revision"] = entry.model_revision
@@ -356,6 +473,20 @@ async def chat_completions(
                 queue_seconds = time.perf_counter() - started
                 telemetry.inflight_requests.inc()
                 try:
+                    if payload.stream:
+                        return StreamingResponse(
+                            _stream_completion(
+                                payload,
+                                prompt,
+                                cache_key,
+                                subject,
+                                decision,
+                                started,
+                                queue_seconds,
+                            ),
+                            media_type="text/event-stream",
+                            headers=_route_headers(decision, "miss"),
+                        )
                     result = await inference_backend.generate(payload, decision)
                 finally:
                     telemetry.inflight_requests.dec()