I recently shipped a distributed code review cluster that fans out security, performance, and style checks to three independent LLM workers, then aggregates their findings through a coordinator node. It cuts review time by running specialist checks in parallel instead of waiting for one monolithic model to reason about everything. In this post, I will walk you through building the same system on Oxlo.ai.
What you'll need
You will need Python 3.10 or newer so you can use concurrent.futures without extra backports. Install the OpenAI SDK with pip install openai. You will also need an Oxlo.ai API key from https://portal.oxlo.ai. Oxlo.ai fits this workload well because its flat per-request pricing does not scale with the size of the code diffs we send to each worker. That means you can pass a 500-line file to three specialist nodes and pay the same as a 10-line file, which makes horizontal scaling predictable.
Step 1: Scaffold the workers
In a production cluster these workers would be separate containers or services. Each one specializes because a narrow prompt on a focused model is cheaper and faster than asking one generalist to do everything. I map the security worker to DeepSeek V3.2 for its coding focus, the performance worker to Qwen 3 32B for reasoning about complexity, and the style worker to Llama 3.3 70B as a reliable generalist. The shared client factory points every request to Oxlo.ai's OpenAI-compatible endpoint, so the code stays portable if you later move workers to dedicated endpoints. I use the synchronous OpenAI client inside a thread pool so the fan-out stays simple and readable.
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SECURITY_PROMPT = "You are a security reviewer. Flag only security issues. Be concise."
PERFORMANCE_PROMPT = "You are a performance reviewer. Flag only performance issues. Be concise."
STYLE_PROMPT = "You are a style reviewer. Flag only style and readability issues. Be concise."
def review_security(code: str):
resp = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SECURITY_PROMPT},
{"role": "user", "content": code},
],
)
return {"node": "security", "result": resp.choices[0].message.content}
def review_performance(code: str):
resp = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": PERFORMANCE_PROMPT},
{"role": "user", "content": code},
],
)
return {"node": "performance", "result": resp.choices[0].message.content}
def review_style(code: str):
resp = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": STYLE_PROMPT},
{"role": "user", "content": code},
],
)
return {"node": "style", "result": resp.choices[0].message.content}
Step 2: Build the dispatcher
A real distributed system needs a message router. In production you might use Celery, NATS, or a Kubernetes service mesh. Here, a ThreadPoolExecutor simulates the fan-out without infrastructure boilerplate. I submit all three reviews at once so they hit Oxlo.ai in parallel. The key idea is that workers are stateless and idempotent, which means you can safely retry or relocate them without side effects.
def fan_out(code: str):
with ThreadPoolExecutor(max_workers=3) as executor:
futures = [
executor.submit(review_security, code),
executor.submit(review_performance, code),
executor.submit(review_style, code),
]
return [f.result() for f in futures]
Step 3: Add the coordinator
Raw worker outputs often conflict or overlap. The coordinator acts as the reduce phase in a map-reduce pipeline. It reads every review and produces a single prioritized report. I run it on Kimi K2.6 because its reasoning capabilities handle contradictory inputs cleanly. You might be tempted to use a smaller model for the coordinator to save money, but synthesis is where errors hide. A confused coordinator can downgrade a critical security warning to a minor note. With Oxlo.ai's flat request pricing, the coordinator costs the same whether it processes three short reviews or three long ones, so I prioritize accuracy over token counting. I also pass the original code back into the prompt so the coordinator can verify context instead of trusting paraphrased snippets.
def coordinate(reviews, code: str):
valid = [r for r in reviews if not isinstance(r, Exception)]
context = "\n\n".join(
[f"--- {r['node']} ---\n{r['result']}" for r in valid]
)
user_prompt = f"Original code:\n{code}\n\nWorker reviews:\n{context}"
resp = client.chat.completions.create(
model="kimi-k2.6",
messages=[
{"role": "system", "content": COORDINATOR_PROMPT},
{"role": "user", "content": user_prompt},
],
)
return resp.choices[0].message.content
Step 4: Handle failures
Networks partition and pods restart. One slow worker should not hang the entire review. I wrap each call in a 10-second timeout and let the coordinator know when a node dropped out. The coordinator prompt explicitly forbids hallucinating missing sections, which keeps the final report honest. This gives us two layers of resilience: the executor isolates crashes, and the timeout wrapper catches slow nodes. If you add a circuit breaker in front of the cluster, you can drop a worker entirely after repeated timeouts and let a human reviewer take its place.
def safe_review(func, code: str, timeout: float = 10.0):
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(func, code)
try:
return future.result(timeout=timeout)
except Exception:
return {"node": "unknown", "result": "[NODE FAILURE: review unavailable]"}
def resilient_fan_out(code: str):
return [
safe_review(review_security, code),
safe_review(review_performance, code),
safe_review(review_style, code),
]
System prompt
The coordinator is the brain of the cluster. Its prompt must be rigid about using only the evidence supplied by workers, or it will invent security issues to fill quiet sections.
COORDINATOR_PROMPT = """You are a senior staff engineer synthesizing code reviews from distributed specialist nodes.
You have received reviews from three workers: security, performance, and style.
Produce a single unified report with sections for Security, Performance, and Style.
If a section is missing or marked as failed, note it as unavailable.
Do not invent findings that did not appear in the worker reviews."""
Run it
Here is the end-to-end test. The sample function contains both a SQL injection risk and a performance issue, so you should see the coordinator surface both even if the style review is clean. Save the file as review_cluster.py and run it.
def main():
code_snippet = '''
def get_user(user_id):
query = f"SELECT * FROM users WHERE id = {user_id}"
cursor.execute(query)
return cursor.fetchall()
'''
reviews = resilient_fan_out(code_snippet)
report = coordinate(reviews, code_snippet)
print(report)
if __name__ == "__main__":
main()
Example output:
Security: Critical SQL injection vulnerability. User input is interpolated directly into the query string without parameterization.
Performance: Using SELECT * without pagination or column limits can cause memory pressure. Consider adding an index on id and fetching only required fields.
Style: Function lacks type hints and a docstring. Rename cursor to a more descriptive variable.
Unavailable sections: None.
Wrap up
To harden this into production, swap the in-memory thread pool for Redis Streams or RabbitMQ so workers can live on separate hosts. Add a SQLite or PostgreSQL log to record every review for audit trails, and expose Prometheus metrics on worker latency and timeout rates. You can also switch the coordinator to JSON mode and parse its output with Pydantic, which makes it easy to feed the report into Jira or GitHub Actions. Oxlo.ai supports JSON mode on all chat models, so the change is just one extra parameter. Because Oxlo.ai bills per request rather than per token, scaling out to ten or twenty specialist workers stays predictable even when your diffs grow. You can explore pricing at https://oxlo.ai/pricing.
