Distributed LLM Cluster

Source: showcase/distributed_llm_cluster See in Playground

Builds directly on the distributed computation tutorial step’s distrib primitives, adding work/distant to provision a remote engine on demand from Cadence.CI , instead of pointing at a melodium dist node started by hand. An HTTP server accepts POST /chat with a plain-text prompt and streams generated tokens back as they’re produced. Cadence.CI provisions a worker that downloads a Mistral checkpoint from the HuggingFace Hub and loads it into memory, once; the ml package, the downloaded weights, and the compute needed to run them only have to be available on the worker itself, never on the front-end process.

Note

Requires an API token from Cadence.CI . Set MELODIUM_API_TOKEN in the environment and run with --api-report to follow execution there. No LLM provider API key is needed: inference runs on the provisioned worker itself, against a Mistral checkpoint downloaded from the HuggingFace Hub, not against a hosted third-party API.

Running


cd showcase/distributed_llm_cluster
export MELODIUM_API_TOKEN="my-cadence-ci-token"
melodium run --api-report Compo.toml --port 8080
 
curl -X POST http://127.0.0.1:8080/chat \
     -d "Explain the Mélodium dataflow model in one sentence."

How it works

A DistantEngine model requests a worker from Cadence.CI, a DistributionEngine model names the remote treatment (inferText) that runs there, and an HttpServer model is the front-end listener:


model runner: DistantEngine(api_token=_, api_url=_)
model distributor: DistributionEngine(
    treatment = "distributed_llm_cluster/main::inferText",
    version   = "0.1.0"
)
model httpServer: HttpServer(host=|from_ipv4(|localhost_ipv4()), port=port)

distant requests a worker sized for real local inference (16GB memory, 4 CPU, 32GB storage, enough for the Mistral-7B fp16 weights plus engine and Hub cache overhead), then distrib::start connects to it, passing the HuggingFace repository ID once as a params entry rather than with every request:


provisionRunner: distant[distant_engine=runner](
    max_duration        = 3600,
    memory              = 16384,
    cpu                 = 4000,
    storage             = 32768,
    edition             = _,
    arch                = _,
    volumes             = [],
    containers          = [],
    service_containers  = [],
    tags                = []
)
startup.trigger -> provisionRunner.trigger,access -> connectDistributor.access
 
connectDistributor: distribStart[distributor=distributor](params=|dataMap([|dataEntry<string>("repo_id", hf_repo_id)]))

A live distrib connection only means the worker process is reachable, not that its multi-gigabyte model weights have finished downloading and loading, which can take several minutes the first time. The front-end sends one real warm-up request right after connecting, and only starts the HTTP server once that request has actually been answered:


warmupPrompt: emit<string>(value=" ")
connectDistributor.ready -> warmupPrompt.trigger
warmupStream: stream<string>()
warmupPrompt.emit -> warmupStream.block
warmupBytes: encode()
warmupStream.stream -> warmupBytes.text
warmupCall: dispatchInfer[distributor=distributor]()
warmupBytes.data -> warmupCall.prompt
warmupDone: trigger<byte>()
warmupCall.response -> warmupDone.stream
 
startHttp: start[http_server=httpServer]()
warmupDone.start -> startHttp.trigger

On the worker, release holds every incoming prompt, including the warm-up one, behind the model’s loaded signal before it reaches generate: a prompt that arrives during load simply waits there instead of being dropped. That is what turns the warm-up response into a trustworthy readiness signal rather than a race against the model still loading.

Dispatching a request to the worker

dispatchInfer follows the same three-step handshake as the distributed computation tutorial step: allocate a distribution ID, send input, receive output, tagged by name. It carries both the warm-up request above and every real chat request:


treatment dispatchInfer[distributor: DistributionEngine]()
  input  prompt:   Stream<byte>
  output response: Stream<byte>
{
    trig: trigger<byte>()
    dist: distribute[distributor=distributor]()
    Self.prompt -> trig.stream,start -> dist.trigger
 
    sendPrompt:   sendStream<byte>[distributor=distributor](name="prompt")
    recvResponse: recvStream<byte>[distributor=distributor](name="response")
    dist.distribution_id -> sendPrompt.distribution_id
    dist.distribution_id -> recvResponse.distribution_id
 
    Self.prompt       -> sendPrompt.data
    recvResponse.data -> Self.response
}

Fetching and loading the model once

inferText runs on the worker, launched once when distribStart connects to it there rather than once per request: its own startup() fires at that point. It fetches the model weights from the HuggingFace Hub with an HfHub model and fetch, then loads them into a Mistral model with load a single time, and serves every subsequent chat request against that same loaded model with generate:


treatment inferText(const repo_id: string)
  model hub:     HfHub(repo_id=repo_id)
  model mistral: Mistral(max_new_tokens=256)
  input  prompt:   Stream<byte>
  output response: Stream<byte>
{
    startup()
    fetchWeights: fetch[hub=hub]()
    loadModel:    load[mistral=mistral]()
    startup.trigger          -> fetchWeights.trigger
    fetchWeights.safetensors -> loadModel.safetensors
    fetchWeights.tokenizer   -> loadModel.tokenizer
 
    waitForModel: release<byte>()
    loadModel.loaded -> waitForModel.leverage
    Self.prompt      -> waitForModel.data
 
    decodePrompt: decode()
    waitForModel.released -> decodePrompt.data,text -> generateReply.prompt
 
    generateReply: generate[mistral=mistral]()
 
    encodeResponse: encode()
    generateReply.generated -> encodeResponse.text,data -> Self.response
}

Dependencies


[dependencies]
std      = "0.10.3"  # core flows, logging, data structures
http     = "0.10.3"  # HTTP client and server
net      = "0.10.3"  # IP address helpers
encoding = "0.10.3"  # UTF-8 encode / decode
work     = "0.10.3"  # cloud runner provisioning
distrib  = "0.10.3"  # stream distribution across runners
ml       = "0.10.3"  # LLM, STT, TTS and local model inference