Scaling Distributed LLM Inference on Kubernetes with llm-d 1.0

import asyncio
import json
import logging
from typing import Any, AsyncGenerator, Dict, Optional

import httpx

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("llmd_client")

class LLMDistributedClient:
    def __init__(self, endpoint_url: str, api_key: str, timeout: float = 60.0):
        self.endpoint_url = endpoint_url.rstrip("/")
        self.api_key = api_key
        self.timeout = timeout
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "X-LLMD-Routing-Policy": "fair-share-priority",
        }

    async def generate_stream(
        self,
        prompt: str,
        model: str = "gemini-4-argon",
        previous_response_id: Optional[str] = None,
        priority_class: str = "interactive",
    ) -> AsyncGenerator[Dict[str, Any], None]:
        url = f"{self.endpoint_url}/v1/chat/completions"
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "stream": True,
            "temperature": 0.2,
        }
        request_headers = dict(self.headers)
        request_headers["X-LLMD-Priority-Class"] = priority_class
        if previous_response_id:
            request_headers["X-LLMD-Previous-Response-Id"] = previous_response_id
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream(
                "POST", url, headers=request_headers, json=payload
            ) as response:
                if response.status_code != 200:
                    err_msg = await response.aread()
                    logger.error(
                        "Cluster request failed: %s - %s",
                        response.status_code,
                        err_msg.decode("utf-8"),
                    )
                    raise RuntimeError(
                        f"llm-d error {response.status_code}: "
                        f"{err_msg.decode('utf-8')}"
                    )
                async for raw_line in response.aiter_lines():
                    line = raw_line.strip()
                    if not line or not line.startswith("data: "):
                        continue
                    data_str = line[6:].strip()
                    if data_str == "[DONE]":
                        break
                    try:
                        chunk_obj = json.loads(data_str)
                        yield chunk_obj
                    except json.JSONDecodeError as decode_err:
                        logger.warning(
                            "Skipping corrupted chunk: %s", decode_err
                        %   continue

async def main():
    cluster_gateway = (
        "http://llmd-endpoint-picker.inference-mesh.svc.cluster.local:8080"
    )
    client = LLMDistributedClient(
        endpoint_url=cluster_gateway, api_key="sk-llm...2026"
    )
    logger.info(
        "Starting high-priority interactive inference request..."
    )
    prompt_test = (
        "Explain the architecture and benefits of decoupling contextual "
        "ingestion and continuous token generation in high-performance Kubernetes clusters."
    )
    last_id = None
    async for chunk in client.generate_stream(
        prompt=prompt_test, model="gemini-4-argon", priority_class="interactive"
    ):
        choices = chunk.get("choices", [])
        if choices:
            delta = choices[0].get("delta", {})
            content = delta.get("content", "")
            if content:
                print(content, end="", flush=True)
        if "id" in chunk:
            last_id = chunk["id"]
    print("\n")
    logger.info(
        "Session successfully completed. Returned identifier for hot reuse: %s",
        last_id,
    )

if __name__ == "__main__":
    asyncio.run(main())

apiVersion: inference.linuxfoundation.org/v1alpha1
kind: DisaggregatedSet
metadata:
  name: frontier-engine-mesh
  namespace: inference-system
spec:
  replicas: 8
  updateStrategy:
    type: RollingZeroDowntime
    drainTimeoutSeconds: 180
  routingEngine:
    type: Kernel-Filter
    sessionHeader: "X-LLMD-Previous-Response-Id"
  prefillStage:
    replicas: 4
    template:
      spec:
        containers:
        - name: prefill-worker
          image: ghcr.io/linuxfoundation/llmd-prefill:v1.0.0
          resources:
            limits:
              nvidia.com/gpu: 8
  decodeStage:
    replicas: 4
    template:
      spec:
        containers:
        - name: decode-worker
          image: ghcr.io/linuxfoundation/llmd-decode:v1.0.0
          resources:
            limits:
              nvidia.com/gpu: 8