DEV Community

shashank ms
shashank ms

Posted on

Building Distributed Systems with LLMs

#engineering #oxlo #ai

We are building a distributed sentiment analysis pipeline that shards a large document across concurrent worker threads, analyzes each chunk with an LLM, and reduces the results into a single summary. This pattern works for any MapReduce-style workload where you want to parallelize inference across nodes without ballooning costs. Because Oxlo.ai uses flat per-request pricing, you can feed long shards to each worker and still pay the same rate per call, which makes it a strong fit for this design. See https://oxlo.ai/pricing for details.

What you'll need

Step 1: Scaffold the project and configure the Oxlo.ai client

We start by importing the OpenAI SDK and pointing it at Oxlo.ai. We also define a synthetic long-form document that simulates server logs or user feedback.

from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")

DOCUMENT = """
The database migration last night caused severe latency across all API endpoints.
Customers reported timeouts and failed checkout flows.
Our on-call engineer rolled back the change within thirty minutes.
Post-incident review highlighted missing integration tests for the new index.
Support tickets spiked but returned to baseline by morning.
""" * 20

Step 2: Define the worker system prompt

Each worker needs a strict system prompt so the pipeline can parse its output as JSON without brittle regex.

WORKER_PROMPT = """You are a sentiment analysis worker node in a distributed pipeline.
Analyze the provided text shard and return a single JSON object with exactly these keys:
- sentiment: one of [positive, negative, neutral, mixed]
- topics: array of up to 3 topics mentioned
- intensity: integer from 1 to 10

Return only the JSON object. Do not wrap it in markdown code fences."""

Step 3: Build the worker node

The worker function accepts a text shard, calls Oxlo.ai with the Llama 3.3 70B model, and returns a Python dictionary.

import json

def analyze_shard(shard):
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": WORKER_PROMPT},
            {"role": "user", "content": f"Analyze this text shard:\n\n{shard}"},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

Step 4: Build the coordinator

The coordinator splits the document into word chunks and fans them out across a thread pool so multiple shards run in parallel.

from concurrent.futures import ThreadPoolExecutor, as_completed

def shard_document(doc, chunk_size=200):
    words = doc.split()
    return [" ".join(words[i:i + chunk_size]) for i in range(0, len(words), chunk_size)]

def run_coordinator(doc, max_workers=4):
    shards = shard_document(doc)
    indexed_results = {}

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_map = {
            executor.submit(analyze_shard, shard): idx
            for idx, shard in enumerate(shards)
        }

        for future in as_completed(future_map):
            idx = future_map[future]
            try:
                indexed_results[idx] = future.result()
            except Exception as exc:
                indexed_results[idx] = {"error": str(exc), "sentiment": "unknown"}

    return [indexed_results[i] for i in range(len(shards))]

Step 5: Reduce results into a final report

After all workers finish, the reducer sends the collected JSON to Qwen 3 32B to synthesize an executive summary.

REDUCE_PROMPT = """You are a reducer node in a distributed pipeline.
You will receive a JSON array of per-shard sentiment analyses.
Produce a single JSON object with:
- overall_sentiment: the dominant sentiment across shards
- top_themes: array of the most common topics
- average_intensity: float
- anomaly_count: number of shards with intensity >= 8

Return only the JSON object, no markdown fences."""

def reduce_results(shard_results):
    payload = json.dumps(shard_results, indent=2)
    response = client.chat.completions.create(
        model="qwen-3-32b",
        messages=[
            {"role": "system", "content": REDUCE_PROMPT},
            {"role": "user", "content": payload},
        ],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

Run it

Wire everything together in a main block and print the intermediate and final output.

if __name__ == "__main__":
    per_shard = run_coordinator(DOCUMENT, max_workers=3)
    print("=== Worker outputs ===")
    print(json.dumps(per_shard, indent=2))

    summary = reduce_results(per_shard)
    print("\n=== Final reduction ===")
    print(json.dumps(summary, indent=2))

Example output:

=== Worker outputs ===
[
  {"sentiment": "negative", "topics": ["latency", "migration"], "intensity": 8},
  {"sentiment": "negative", "topics": ["timeouts", "checkout"], "intensity": 7},
  {"sentiment": "neutral", "topics": ["rollback", "on-call"], "intensity": 4},
  {"sentiment": "mixed", "topics": ["review", "tickets"], "intensity": 5}
]

=== Final reduction ===
{
  "overall_sentiment": "negative",
  "top_themes": ["latency", "timeouts", "checkout"],
  "average_intensity": 6.0,
  "anomaly_count": 2
}

Next steps

Swap ThreadPoolExecutor for a real message queue like Redis or RabbitMQ so workers can run on separate machines. You should also add retry logic with exponential backoff and a dead-letter queue for any shards that fail after repeated attempts.

Top comments (0)