# GPU Multi-Model Inference Service (/integrations/rayserve/scenarios/gpu-multi-model-service)



## Scenario [#scenario]

An ML platform team runs several models in production at once: a ResNet image classifier, a BERT text embedder, and an LLM completion service. Each model is GPU-bound, has its own throughput profile, and must fail independently — a stuck LLM job should never block image classification.

This scenario wires each model to its **own KubeMQ task queue** behind a Ray Serve GPU deployment. A front-end HTTP service enqueues work onto the right queue with `enqueue_task_sync`, GPU consumers process tasks in parallel, long-running jobs stream `report_progress` events to subscribers, and any task that exhausts its retries fires an `on_dlq` callback that pages an operator.

Like the [multi-language event pipeline](/connectors/cloudevents/how-to/events), the value here is isolation through queues: each model is an independent consumer of a dedicated channel, so you can scale, version, and A/B-test models without coupling them to each other.

## Architecture [#architecture]

A producer enqueues tasks onto model-specific queues. Each model is a `@serve.deployment(ray_actor_options={"num_gpus": 1})` consumer that pulls from its queue, writes results to the queue-peek backend, and publishes progress on `{queue}.progress`. The producer polls `get_task_status_sync` for the outcome and can subscribe to the progress channel for live updates.

<Mermaid
  chart="flowchart LR
    API[&#x22;HTTP Producer<br/>(FastAPI / Flask)&#x22;] -->|&#x22;enqueue_task_sync()&#x22;| QI[&#x22;Queue<br/>image-tasks&#x22;]
    API -->|&#x22;enqueue_task_sync()&#x22;| QE[&#x22;Queue<br/>embed-tasks&#x22;]
    API -->|&#x22;enqueue_task_sync()&#x22;| QL[&#x22;Queue<br/>llm-tasks&#x22;]
    QI --> DI[&#x22;ResNet Deployment<br/>num_gpus=1&#x22;]
    QE --> DE[&#x22;Embedder Deployment<br/>num_gpus=1&#x22;]
    QL --> DL[&#x22;LLM Deployment<br/>num_gpus=1&#x22;]
    DI -->|&#x22;store result&#x22;| RB[&#x22;Result Backend<br/>(queue-peek)&#x22;]
    DE -->|&#x22;store result&#x22;| RB
    DL -->|&#x22;store result&#x22;| RB
    DL -->|&#x22;report_progress()&#x22;| PG[&#x22;Events<br/>llm-tasks.progress&#x22;]
    RB -->|&#x22;get_task_status_sync()&#x22;| API
    PG -->|&#x22;SSE&#x22;| API"
/>

## GPU Deployment Binding [#gpu-deployment-binding]

In production you do not call `initialize()` yourself. You decorate a Ray Serve deployment with `@serve.deployment` to reserve a GPU, then `@task_consumer` to bind it to a KubeMQ queue through a `TaskProcessorConfig`. Ray Serve calls `initialize()` and `start_consumer()` for you; each method invocation runs as a task pulled off the queue.

```python title="image_deployment.py"
from ray import serve
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

@serve.deployment(ray_actor_options={"num_gpus": 1})
@task_consumer(
    task_processor_config=TaskProcessorConfig(
        adapter_class=KubeMQTaskProcessorAdapter,
        adapter_config=KubeMQAdapterConfig(
            address="kubemq:50000",
            auth_token="your-jwt-token",
        ),
        queue_name="inference-tasks",
        max_retries=3,
        failed_task_queue_name="inference-tasks-dlq",
    )
)
class ModelDeployment:
    def __init__(self):
        self.model = load_model()

    def __call__(self, data: str) -> dict:
        return self.model.predict(data)
```

`ray_actor_options={"num_gpus": 1}` pins one GPU per replica. `queue_name` is the channel the consumer polls, `max_retries` caps redelivery attempts before a task is dead-lettered, and `failed_task_queue_name` is where exhausted tasks land. The `auth_token` carries a JWT to the in-cluster broker at `kubemq:50000`.

<Callout type="info">
  Each model deployment uses a **distinct** `queue_name`. That single decision gives you per-model isolation, independent scaling, and independent failure handling for free — the rest of this page builds on it.
</Callout>

## Multiple GPU Models on Separate Queues [#multiple-gpu-models-on-separate-queues]

Each model gets its own adapter, its own queue, and its own consumer. The producer routes a request to the correct adapter and later collects every result by polling. The snippet below mirrors the packaged `gpu_multi_model.py` example, which simulates three GPUs (ResNet, BERT, Whisper) on three channels.

```python title="gpu_multi_model.py"
import os
import time
import uuid

from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")


def resnet_predict(image_data: str) -> dict:
    """Simulated ResNet image classification (GPU 0)."""
    time.sleep(0.1)  # Simulate 100ms GPU inference
    categories = ["cat", "dog", "bird", "car", "plane"]
    idx = hash(image_data) % len(categories)
    return {"model": "resnet50", "gpu": 0, "prediction": categories[idx]}


def bert_predict(text: str) -> dict:
    """Simulated BERT text classification (GPU 1)."""
    time.sleep(0.12)  # Simulate 120ms GPU inference
    labels = ["positive", "negative", "neutral"]
    return {"model": "bert-base", "gpu": 1, "sentiment": labels[hash(text) % 3]}


def main():
    uid = uuid.uuid4().hex[:8]
    channel_resnet = f"example-ml-gpu-resnet-{uid}"
    channel_bert = f"example-ml-gpu-bert-{uid}"

    # --- Adapter for ResNet (GPU 0) ---
    adapter_resnet = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER))

    class _CfgResnet:
        queue_name = channel_resnet
        max_retries = 1
        failed_task_queue_name = f"{channel_resnet}.dlq"
        unprocessable_task_queue_name = ""

    adapter_resnet.initialize(consumer_concurrency=2, task_processor_config=_CfgResnet())
    adapter_resnet.register_task_handler(resnet_predict, name="predict")
    adapter_resnet.start_consumer()

    # --- Adapter for BERT (GPU 1) ---
    adapter_bert = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER))

    class _CfgBert:
        queue_name = channel_bert
        max_retries = 1
        failed_task_queue_name = f"{channel_bert}.dlq"
        unprocessable_task_queue_name = ""

    adapter_bert.initialize(consumer_concurrency=2, task_processor_config=_CfgBert())
    adapter_bert.register_task_handler(bert_predict, name="predict")
    adapter_bert.start_consumer()

    try:
        # Dispatch requests to the appropriate model queue
        img = adapter_resnet.enqueue_task_sync("predict", args=["image_0.jpg"])
        txt = adapter_bert.enqueue_task_sync("predict", args=["Great product!"])

        for adapter, tid in [(adapter_resnet, img.id), (adapter_bert, txt.id)]:
            while True:
                status = adapter.get_task_status_sync(tid)
                if status.status in ("SUCCESS", "FAILURE"):
                    print(status.status, status.result)
                    break
                time.sleep(0.5)
    finally:
        adapter_resnet.stop_consumer()
        adapter_bert.stop_consumer()


if __name__ == "__main__":
    main()
```

Each handler is registered under the name `predict` on its own adapter, so the same task name routes to a different model depending on which queue it lands on.

### Chained Inference Pipeline [#chained-inference-pipeline]

When a single request needs several GPU stages — preprocess, classify, postprocess — give each stage its own queue and chain the calls, passing the result of one stage as the argument to the next. The packaged `image_classification_pipeline.py` example does exactly this across three adapters.

```python title="image_classification_pipeline.py"
def poll_result(adapter, task_id, timeout=15):
    start = time.perf_counter()
    while (time.perf_counter() - start) < timeout:
        status = adapter.get_task_status_sync(task_id)
        if status.status in ("SUCCESS", "FAILURE"):
            return {"status": status.status, "result": status.result}
        time.sleep(0.5)
    return None


# Stage 1: Preprocess (resize, normalize) on the preprocess queue
pre_task = adapter_pre.enqueue_task_sync(
    "preprocess", args=["cat_photo.jpg"], kwargs={"target_size": 224}
)
pre = poll_result(adapter_pre, pre_task.id)

# Stage 2: Classify — pass the preprocessed tensor metadata forward
cls_task = adapter_cls.enqueue_task_sync("classify", args=[pre["result"]])
cls = poll_result(adapter_cls, cls_task.id)

# Stage 3: Postprocess — format the final label
post_task = adapter_post.enqueue_task_sync("postprocess", args=[cls["result"]])
final = poll_result(adapter_post, post_task.id)
print(final["result"]["final_label"], final["result"]["confidence_level"])
```

Splitting stages across queues lets you scale the bottleneck stage independently — for example, run more classifier replicas than preprocessor replicas.

## Batch Inference with Parallel Consumers [#batch-inference-with-parallel-consumers]

For high-throughput batch jobs, raise `consumer_concurrency` so a single deployment processes many tasks in parallel. The producer enqueues the whole batch up front, then drains results as they complete. The packaged `batch_inference.py` example runs five concurrent consumers.

```python title="batch_inference.py"
adapter.initialize(consumer_concurrency=5, task_processor_config=_Cfg())
adapter.register_task_handler(classify_item, name="classify_item")
adapter.start_consumer()

# Enqueue the full batch up front
task_ids = []
start = time.perf_counter()
for i in range(BATCH_SIZE):
    result = adapter.enqueue_task_sync("classify_item", args=[i, f"sample text {i}"])
    task_ids.append(result.id)

# Drain results as the parallel consumers complete them
completed = {}
poll_start = time.perf_counter()
while len(completed) < BATCH_SIZE and (time.perf_counter() - poll_start) < 60:
    for tid in task_ids:
        if tid not in completed:
            status = adapter.get_task_status_sync(tid)
            if status.status in ("SUCCESS", "FAILURE"):
                completed[tid] = status
    time.sleep(0.5)

total_time = time.perf_counter() - start
print(f"Throughput: {BATCH_SIZE / total_time:.1f} tasks/sec")
```

<Callout type="info">
  `consumer_concurrency` controls how many tasks one deployment processes at once. Combine it with the queue-depth autoscaling policy to add replicas as the backlog grows — see the [autoscaling guide](/integrations/rayserve/how-to/autoscaling).
</Callout>

## Progress for Long-Running Jobs [#progress-for-long-running-jobs]

LLM completion and text-to-speech jobs run for seconds, not milliseconds. Inside the handler, call `report_progress(task_id, pct, detail)` to publish progress events on the `{queue_name}.progress` channel. Subscribers — like the SSE endpoint shown later — receive each update in real time.

The handler needs the task id, so the producer passes it as an explicit `task_id` kwarg. This pattern comes straight from the packaged `llm_inference.py` example.

```python title="llm_inference.py"
adapter: KubeMQTaskProcessorAdapter | None = None  # module-level for handler closure


def generate_completion(prompt: str, max_tokens: int = 50, task_id: str = "") -> dict:
    """Simulated LLM generation with per-step progress reporting."""
    assert adapter is not None

    if task_id:
        adapter.report_progress(task_id, 0, "Tokenizing prompt")

    generated_tokens = []
    for i in range(max_tokens):
        time.sleep(0.02)  # Simulate 20ms per token
        generated_tokens.append(next_token(prompt, i))
        pct = int((i + 1) / max_tokens * 100)
        if (i + 1) % 10 == 0 and task_id:
            adapter.report_progress(task_id, pct, f"Generated {i + 1}/{max_tokens} tokens")

    if task_id:
        adapter.report_progress(task_id, 100, "Generation complete")

    return {
        "prompt": prompt,
        "completion": " ".join(generated_tokens),
        "tokens_generated": max_tokens,
        "finish_reason": "length",
    }


# Producer side: pass an explicit task_id so the handler can report against it
known_task_id = f"llm-task-{uuid.uuid4().hex[:8]}"
result = adapter.enqueue_task_sync(
    "generate",
    kwargs={"prompt": "Explain message queues", "max_tokens": 40, "task_id": known_task_id},
)
```

### Binary Results [#binary-results]

Text-to-speech and other models that emit binary output return it base64-encoded so the result stays JSON-safe through the queue-peek backend. The packaged `text_to_speech.py` example shows the round trip.

```python title="text_to_speech.py"
import base64


def synthesize_speech(text: str, voice: str = "default", speed: float = 1.0) -> dict:
    audio_bytes = run_tts(text, voice, speed)  # raw WAV bytes
    return {
        "text": text,
        "voice": voice,
        "sample_rate": 22050,
        "audio_format": "wav",
        "audio_size_bytes": len(audio_bytes),
        "audio_base64": base64.b64encode(audio_bytes).decode("ascii"),
    }


# Producer side: decode the base64 payload back into bytes
outcome = poll_result(adapter, result.id)
audio = base64.b64decode(outcome["result"]["audio_base64"])
```

## Operational Concerns [#operational-concerns]

<Accordions>
  <Accordion title="A/B model testing on separate queues">
    Run two model versions side by side, each on its own queue, and route a fraction of traffic to each. The producer compares latency and confidence across both. This follows the packaged `ab_model_testing.py` example.

    ```python title="ab_model_testing.py"
    import random

    # Two adapters, two queues, same task name "predict"
    adapter_a.register_task_handler(model_a_predict, name="predict")  # accurate, slower
    adapter_b.register_task_handler(model_b_predict, name="predict")  # faster, lighter

    for i in range(NUM_REQUESTS):
        text = f"Sample review text number {i}"
        use_a = random.random() < 0.5
        adapter = adapter_a if use_a else adapter_b
        result = adapter.enqueue_task_sync("predict", args=[text])
        # ...track (result.id, "model-a" if use_a else "model-b") and poll both adapters
    ```

    Because each model owns a queue, you can shift the split, drain one side, or retire a model without touching the other.
  </Accordion>

  <Accordion title="Hot-swap model versions without downtime">
    `register_task_handler` overwrites the handler bound to a name. Re-registering the same name with a new function swaps the model in place — the consumer never stops, and in-flight queue messages start hitting the new version on their next dispatch. This mirrors the packaged `model_versioning.py` example.

    ```python title="model_versioning.py"
    # Phase 1: serve v1
    adapter.register_task_handler(predict_v1, name="predict")
    adapter.start_consumer()

    # Phase 2: hot-swap to v2 — consumer still running
    adapter.register_task_handler(predict_v2, name="predict")
    time.sleep(0.5)  # let the swap take effect

    result_v2 = adapter.enqueue_task_sync("predict", args=["This product could be better"])
    # result_v2 is now served by predict_v2 — zero downtime, no consumer restart
    ```
  </Accordion>
</Accordions>

## DLQ Alerting [#dlq-alerting]

When a task fails and exhausts `max_retries`, the adapter stores a `FAILURE` result, nacks the message to the dead letter queue, and fires the `on_dlq` callback you set on `KubeMQAdapterConfig`. Use that callback to page an operator the moment a model starts producing permanent failures. The callback receives the `task_id` and the formatted error string, exactly as the adapter invokes it on retry exhaustion.

```python title="dlq_monitoring_callback.py"
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter


def on_dlq_callback(task_id: str, error: str) -> None:
    """Called when a task exhausts retries and moves to the DLQ."""
    print(f"[DLQ ALERT] task_id={task_id} error={error}")
    # In production: page on-call, push to PagerDuty / Slack, increment an alert metric


config = KubeMQAdapterConfig(
    address="localhost:50000",
    on_dlq=on_dlq_callback,
)
adapter = KubeMQTaskProcessorAdapter(config)


class _Cfg:
    queue_name = "inference-tasks"
    max_retries = 2
    failed_task_queue_name = "inference-tasks.dlq"
    unprocessable_task_queue_name = ""


adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
```

<Callout type="warn">
  The `on_dlq` callback runs on the consumer thread. Keep it fast and non-blocking — enqueue the alert and return. The adapter wraps the call in a try/except and logs any exception it raises, so a failing alert never crashes the consumer, but it can still slow down processing if it blocks.
</Callout>

## Front-End API [#front-end-api]

Expose enqueue, status, cancel, progress (SSE), and metrics over HTTP so clients never talk to KubeMQ directly. The packaged `fastapi_app.py` example manages the adapter lifecycle in a FastAPI lifespan and exposes the full surface.

<Tabs items="[&#x22;FastAPI&#x22;, &#x22;Flask&#x22;]">
  <Tab value="FastAPI">
    ```python title="fastapi_app.py"
    import asyncio
    import json
    import uuid
    from contextlib import asynccontextmanager

    from fastapi import FastAPI
    from fastapi.responses import StreamingResponse
    from kubemq import (
        CancellationToken,
        ClientConfig,
        EventReceived,
        EventsSubscription,
        PubSubClient,
    )
    from pydantic import BaseModel, Field

    from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

    BROKER = "localhost:50000"
    CHANNEL = f"example-fastapi-{uuid.uuid4().hex[:8]}"

    config = KubeMQAdapterConfig(address=BROKER)
    adapter = KubeMQTaskProcessorAdapter(config)


    def echo_handler(text: str = "") -> dict:
        return {"echo": text, "source": "fastapi-app"}


    class EnqueueRequest(BaseModel):
        task_name: str
        args: list | None = None
        kwargs: dict | None = None


    @asynccontextmanager
    async def lifespan(app: FastAPI):
        class _Cfg:
            queue_name = CHANNEL
            max_retries = 3
            failed_task_queue_name = f"{CHANNEL}.dlq"
            unprocessable_task_queue_name = f"{CHANNEL}.unprocessable"

        adapter.initialize(consumer_concurrency=2, task_processor_config=_Cfg())
        adapter.register_task_handler(echo_handler, name="echo")
        adapter.start_consumer()
        yield
        adapter.stop_consumer()


    app = FastAPI(title="KubeMQ Ray Serve Example", lifespan=lifespan)


    @app.get("/health")
    def health():
        return {"healthy": adapter.health_check()}


    @app.post("/tasks")
    def enqueue_task(req: EnqueueRequest):
        result = adapter.enqueue_task_sync(
            task_name=req.task_name, args=req.args, kwargs=req.kwargs
        )
        return {"task_id": result.id, "status": result.status}


    @app.get("/tasks/{task_id}")
    def get_task_status(task_id: str):
        status = adapter.get_task_status_sync(task_id)
        return {"task_id": task_id, "status": status.status, "result": status.result}


    @app.post("/tasks/{task_id}/cancel")
    def cancel_task(task_id: str):
        return {"task_id": task_id, "cancelled": adapter.cancel_task(task_id)}


    @app.get("/tasks/{task_id}/progress")
    async def task_progress_sse(task_id: str):
        """SSE stream: subscribe to {CHANNEL}.progress, filter by task_id."""
        progress_queue: asyncio.Queue[dict] = asyncio.Queue()
        cancel_token = CancellationToken()
        loop = asyncio.get_running_loop()

        def on_event(event: EventReceived) -> None:
            try:
                data = json.loads(event.body)
                if data.get("task_id") == task_id:
                    loop.call_soon_threadsafe(progress_queue.put_nowait, data)
            except (json.JSONDecodeError, TypeError):
                pass

        pubsub = PubSubClient(
            config=ClientConfig(address=BROKER, client_id="fastapi-progress-subscriber")
        )
        subscription = EventsSubscription(
            channel=f"{CHANNEL}.progress",
            on_receive_event_callback=on_event,
            on_error_callback=lambda err: None,
        )
        pubsub.subscribe_to_events(subscription, cancel_token)

        async def event_stream():
            try:
                timeout = 120
                while timeout > 0:
                    try:
                        data = await asyncio.wait_for(progress_queue.get(), timeout=5.0)
                        yield f"data: {json.dumps(data)}\n\n"
                        if data.get("pct", 0) >= 100:
                            break
                    except asyncio.TimeoutError:
                        yield f"data: {json.dumps({'heartbeat': True})}\n\n"
                    timeout -= 5
            finally:
                cancel_token.cancel()
                pubsub.close()

        return StreamingResponse(event_stream(), media_type="text/event-stream")


    @app.get("/metrics")
    def get_metrics():
        return adapter.get_metrics()
    ```

    The SSE endpoint subscribes to the same `{CHANNEL}.progress` channel the handler publishes to, filters events by `task_id`, and streams `data:` frames until the job reports `100`. The 8-metric `get_metrics()` dict (queue depth, in-flight, DLQ depth, counters, durations) feeds dashboards and the autoscaler.
  </Tab>

  <Tab value="Flask">
    ```python title="flask_app.py"
    import atexit
    import uuid

    from flask import Flask, jsonify, request

    from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter

    BROKER = "localhost:50000"
    CHANNEL = f"example-flask-{uuid.uuid4().hex[:8]}"

    config = KubeMQAdapterConfig(address=BROKER)
    adapter = KubeMQTaskProcessorAdapter(config)
    _adapter_started = False


    def echo_handler(text: str = "") -> dict:
        return {"echo": text, "source": "flask-app"}


    def _ensure_adapter_started():
        global _adapter_started
        if _adapter_started:
            return

        class _Cfg:
            queue_name = CHANNEL
            max_retries = 3
            failed_task_queue_name = f"{CHANNEL}.dlq"
            unprocessable_task_queue_name = f"{CHANNEL}.unprocessable"

        adapter.initialize(consumer_concurrency=2, task_processor_config=_Cfg())
        adapter.register_task_handler(echo_handler, name="echo")
        adapter.start_consumer()
        _adapter_started = True


    app = Flask(__name__)


    @app.before_request
    def _startup():
        _ensure_adapter_started()


    @app.get("/health")
    def health():
        return jsonify(healthy=adapter.health_check())


    @app.post("/tasks")
    def enqueue_task():
        body = request.get_json(force=True)
        result = adapter.enqueue_task_sync(
            task_name=body["task_name"],
            args=body.get("args"),
            kwargs=body.get("kwargs"),
        )
        return jsonify(task_id=result.id, status=result.status)


    @app.get("/tasks/<task_id>")
    def get_task_status(task_id):
        status = adapter.get_task_status_sync(task_id)
        return jsonify(task_id=task_id, status=status.status, result=status.result)


    @app.post("/tasks/<task_id>/cancel")
    def cancel_task(task_id):
        return jsonify(task_id=task_id, cancelled=adapter.cancel_task(task_id))


    @app.get("/metrics")
    def get_metrics():
        return jsonify(adapter.get_metrics())


    atexit.register(lambda: adapter.stop_consumer())
    ```

    Flask exposes the same enqueue, status, cancel, health, and metrics surface, starting the adapter on the first request and stopping it on process exit. For progress SSE, prefer the FastAPI variant — async streaming maps cleanly onto FastAPI's `StreamingResponse`.
  </Tab>
</Tabs>

## Running the Service [#running-the-service]

<Steps>
  <Step>
    ### Start KubeMQ [#start-kubemq]

    Run the broker locally with both ports exposed:

    <RunKubeMQ ports="[50000, 9090]" />

    Port `50000` is the gRPC endpoint the adapters connect to. Port `9090` exposes the shared HTTP server used by connector endpoints such as KEDA-based scaling.
  </Step>

  <Step>
    ### Install dependencies [#install-dependencies]

    ```bash title="terminal"
    uv pip install kubemq-rayserve fastapi uvicorn
    ```
  </Step>

  <Step>
    ### Run the front-end API [#run-the-front-end-api]

    ```bash title="terminal"
    python fastapi_app.py
    ```

    The app serves on `http://0.0.0.0:8000`. Override the broker address with `KUBEMQ_ADDRESS` if it is not on `localhost:50000`.
  </Step>

  <Step>
    ### Enqueue and poll [#enqueue-and-poll]

    ```bash title="terminal"
    curl -s -X POST http://localhost:8000/tasks \
      -H "Content-Type: application/json" \
      -d '{"task_name": "echo", "kwargs": {"text": "hello gpu"}}'
    # {"task_id":"...","status":"PENDING"}

    curl -s http://localhost:8000/tasks/<task_id>
    # {"task_id":"...","status":"SUCCESS","result":{"echo":"hello gpu","source":"fastapi-app"}}
    ```

    Stream progress for a long job with `curl -N http://localhost:8000/tasks/<task_id>/progress`.
  </Step>
</Steps>

## Key Takeaway [#key-takeaway]

One queue per model is the organizing principle. It buys you **independent scaling** (raise `consumer_concurrency` and let `kubemq_queue_depth_policy` add GPU replicas where the backlog is), **independent failure handling** (`on_dlq` alerts fire per model), and **independent lifecycle** (hot-swap or A/B-test a model without touching its neighbors) — all backed by a single KubeMQ broker, with no separate result store or scaler to operate.

## Next Steps [#next-steps]

<Cards>
  <Card title="Autoscaling Guide" href="/integrations/rayserve/how-to/autoscaling" description="Scale GPU replicas from live KubeMQ queue depth with kubemq_queue_depth_policy and KEDA." />

  <Card title="Kubernetes Production Deployment" href="/integrations/rayserve/scenarios/kubernetes-production-deployment" description="Run the broker, Ray cluster, RayService, and queue-depth autoscaling end to end on Kubernetes." />

  <Card title="Progress Tracking" href="/integrations/rayserve/how-to/progress-tracking" description="Deep dive into report_progress events and subscribing to the {queue}.progress channel." />
</Cards>