skip to content
The same pipeline moving from a single Ractor process to a distributed Ray cluster

Part 2: Building with Ray

This is Part 2 of a 2-part series on building production-ready, multi-layer RAG systems with Ragforge.

Scaling beyond one box: where Ray and Anyscale actually fit

Everything in Part 1 is a single binary on a single machine. I want to be precise about what comes next: I have not built or tested a Ray version of this pipeline.

What follows is the migration path I would take if this had to become a 100M-patient production job, or if clinical note generation moved from templates to an LLM. It’s a design, not a completed port.

Think of it as a spectrum:

for-loop [simple, slow]
  -> Ractor on Tokio [concurrent, deterministic, laptop-scale]
  -> Ray on Anyscale [distributed, elastic, cluster-scale]

I picked Ractor first on purpose to prove correctness and byte-identical determinism on a laptop before paying the distributed systems tax.

The architecture, side-by-side

Current: Deterministic single binary

Tokio Runtime
OrchestratorActor -> Profile -> Condition -> Medication -> Reaction -> Note -> GuardrailActor -> ChunkingActor -> Writer
                   all in-process, zero-copy message passing via Ractor

Next: Same logic, different runtime

Driver [local laptop OR Anyscale head node]
  | ray.init[address="auto"]
  v
+----------------------------------------------------+
| Anyscale Cluster - autoscaler + Ray Dashboard +    |
| spot instance recovery + per-task tracing          |
+----------------------------------------------------+
  | futures = [process_batch.remote[seed, batch_id] for batch_id in range[N]]
  v
+----------------------------------------------------------------+
| process_batch.remote[] - ONE fused Ray task per batch          |
|   Why fused? Mirrors src/generation.rs. 5 separate tasks =     |
|   5 hops + 5 cloudpickle serializations. 1 task = 0 hops.      |
|   profile -> condition -> medication -> reaction -> note       |
+----------------------------------------------------------------+
  |  batches into Plasma object store [zero-copy between workers on same node]
  v
+----------------------------------------------------------------+
| Guardrail [two options]:                                       |
| 1. @ray.remote class GuardrailActor - stateful counters        |
| 2. ray.data.Dataset.map_batches[guardrail_filter] - stateless  |
+----------------------------------------------------------------+
  | ray.data gives you backpressure here for free
    [current Ractor version dispatches all batches at startup]
  v
ray.data.write_parquet["s3://bucket/chunks/"] // sharded writers

Component-level mapping - nothing in logic changes, only the wrapper

Current - Ractor on TokioRay on AnyscaleWhy
OrchestratorActor dispatching GeneratePatientBatchDriver: batch_ids = range[total_batches] submitted as futuresSame fan-out, now across nodes
Profile -> Condition -> Medication -> Reaction -> NoteOne fused @ray.remote def process_batchAvoids serialization cost
GuardrailActor with 5 checks@ray.remote class GuardrailActor OR Dataset.map_batches[]Stateful stats vs. stateless filter
ChunkingActor -> PatientWriterActorray.data.write_json/parquet[]S3 sharded writers, no manual chunking
EvalOrchestrator reloads patients.jsonl from diskray.data.read_json[].map_batches[generate_evals]No reload, shuffle is native
rng.rs: ChaCha8Rng[seed ^ batch_id]seed ^ batch_id passed explicitly into each taskSame contract, different library. This is how you keep byte-identical

A more concrete sketch

The key contract from src/rng.rs survives: determinism comes from seeding, not from scheduler order.

import ray


ray.init(address="auto")  # locally: local cluster. on Anyscale: autoscaling cluster


@ray.remote
def process_batch(seed: int, batch_id: int, start_id: int, size: int, cfg: dict):
    # mirrors src/rng.rs - explicit seed derivation, no global RNG
    base_rng = seed ^ batch_id
    records = [make_patient(base_rng ^ i, start_id + i, cfg) for i in range(size)]


    # mirrors src/actors/guardrail.rs - inline, so failure loses 1 batch not the job
    return [r for r in records if not has_pii_or_dup(r)]


# Driver - 100M / 1000 = 100k tasks. Anyscale adds nodes as this queue grows.
futures = [process_batch.remote(42, b, b * 1000, 1000, cfg) for b in range(100_000)]
ds = ray.data.from_pandas([ray.get(f) for f in futures]) # Ray handles object store spilling
ds.write_parquet("s3://bucket/chunks/")

For LLM notes, only the last step changes:

# Ray Serve handles GPU scheduling + continuous batching
# Rust binary would need a Python sidecar or FFI for this
serve.run(llm_note_generator.bind(), route_prefix="/notes")

Where this wins, where it loses

Worth it when:

  1. You OOM on one box. 100k patients fits in RAM. 100M doesn’t. Ray shards the work with no code change - range[100] becomes range[100_000].
  2. Notes become an LLM. ray.serve gives you GPU autoscaling and batching. In Rust you’d be building it yourself.
  3. You need ops for free. Anyscale gives you spot-instance retries, lineage-based reconstruction, dashboard tracing, S3 checkpointing.

Not worth it yet:

For corpora < 500k, this cluster machinery is slower than the Rust binary.

  1. Determinism gets harder. Ray’s scheduler is intentionally non-deterministic for throughput. If you don’t thread seed ^ batch_id through every boundary, you lose the byte-identical guarantee I proved in the last section.
  2. Serialization tax. Cloudpickle + Plasma still costs. The Rust version is passing references. Expect 2-3x slower per-record at 100k scale.
  3. Ops tax. cargo build --release && ./generator vs ray up, anyscale job submit, cluster YAML, object store tuning.

My honest decision rule: I started with Ractor to ground determinism on a laptop. If I need 100M patients or LLM notes, the next step is porting process_batch to Ray and letting Anyscale handle the cluster. The logic in src/generation.rs doesn’t change - only the runtime wrapper does.