A Bounded Batch Queue in Your Terminal
A Bounded Batch Queue in Your Terminal
A batch script that launches 2,000 requests at once is not fast; it is an incident waiting for a rate limit. My default is less exciting: a bounded queue, four workers, and one durable result per input. Here is the version I would run from an editor terminal before scaling anything up.
Fluxo gives us one key and one base URL, https://api.fluxorelay.com, across providers. It works with Claude Code, Codex, Gemini CLI, and any OpenAI SDK. For this batch, we will use the Python SDK and select Claude Haiku 4.5 through its exact model ID from the dashboard.
1. Prepare the SDK and twenty test inputs
If you need a key, you can create a Fluxo account and follow along with a small batch first. Run each command below separately. Replace both placeholder values before running the job; the model setting needs an API identifier, not necessarily its display name.
python3 -m venv .venv — . .venv/bin/activate
python -m pip install openai — export OPENAI_API_KEY='YOUR_FLUXO_KEY'
export OPENAI_BASE_URL='https://api.fluxorelay.com' — export FLUXO_MODEL='EXACT_MODEL_ID_FROM_YOUR_DASHBOARD' — export CONCURRENCY=4 — python -c 'import json; print("\n".join(json.dumps({"id": i, "prompt": f"Write one sentence explaining batch processing to developer {i}."}) for i in range(20)))' > inputs.jsonl
Keep credentials out of committed files. These inputs carry stable IDs because completion order will differ from input order.
2. Create four workers behind a bounded queue
The following is one shell command. It writes batch.py using printf, so no heredoc formatting or editor setup is required.
printf '%s\n' 'import asyncio, json, os, sys' 'from openai import AsyncOpenAI' '' 'async def main():' ' n = int(os.environ.get("CONCURRENCY", "4"))' ' if n < 1:' ' raise ValueError("CONCURRENCY must be positive")' ' queue = asyncio.Queue(maxsize=2 * n)' ' failures = 0' ' async with AsyncOpenAI(timeout=60.0, max_retries=3) as client:' ' with open("results.jsonl", "a", encoding="utf-8") as output:' ' async def worker():' ' nonlocal failures' ' while True:' ' item = await queue.get()' ' try:' ' if item is None:' ' return' ' try:' ' response = await client.chat.completions.create(' ' model=os.environ["FLUXO_MODEL"],' ' messages=[{"role": "user", "content": item["prompt"]}],' ' max_tokens=120,' ' )' ' result = {"id": item["id"], "ok": True, "text": response.choices[0].message.content}' ' except Exception as exc:' ' failures += 1' ' result = {"id": item["id"], "ok": False, "error": str(exc)}' ' output.write(json.dumps(result) + "\n")' ' output.flush()' ' finally:' ' queue.task_done()' ' workers = [asyncio.create_task(worker()) for _ in range(n)]' ' with open("inputs.jsonl", encoding="utf-8") as source:' ' for line in source:' ' if line.strip():' ' await queue.put(json.loads(line))' ' for _ in workers:' ' await queue.put(None)' ' await queue.join()' ' await asyncio.gather(*workers)' ' return int(failures > 0)' '' 'sys.exit(asyncio.run(main()))' > batch.py
Four workers mean at most four SDK calls are active. Eight queued items provide a small buffer. When that buffer fills, queue.put waits: this is backpressure, and it prevents the producer from loading the entire input into memory.
Retries stay inside each worker, so they cannot create additional concurrent requests. The SDK handles supported transient failures, including rate-limit responses, with backoff and retry-header handling. Three retries allow up to four attempts; they do not guarantee eventual success.
3. Run the batch and inspect failures
python batch.py — python -c 'import json; rows=[json.loads(line) for line in open("results.jsonl")]; print({"completed": len(rows), "failed": sum(not row["ok"] for row in rows)})'
The runner appends results and exits nonzero if any request exhausts its retries. Flushing each line reduces buffered-output loss, but is not a guarantee against power failure. Rerunning the same input repeats requests and appends duplicate IDs; filter completed IDs before a production restart.
4. Tune concurrency without confusing it with rate
Start at four workers, measure latency and failures, then try eight on a fresh test batch. Concurrency limits in-flight work, not requests per minute or tokens per minute. If short requests still trigger throttling, lower concurrency or add a separate rate limiter; a larger queue will not increase upstream capacity.
Fluxo bills real tokens, including cache reads and writes, with no monthly fee. Budget for retries: a timeout can leave the outcome uncertain, and repeating a request can incur additional usage. Fluxo also shares referral revenue when invited users pay, independently of these queue settings.
For help with unexpected responses, leave a note on the contact page after signing in. Include timestamps and sanitized errors, never your API key.