Batch processing

Process thousands of documents in parallel with Radium

Batch processing

When you need to process a large number of items — classify support tickets, summarize articles, extract entities — you want concurrency and rate-limit awareness. This example shows how to do it safely and efficiently.

Read a dataset

Assume you have a reviews.jsonl file with one JSON object per line:

reviews.jsonl

{"id": 1, "text": "The battery lasts forever and the screen is gorgeous."}
{"id": 2, "text": "Shipping was late but the product itself is solid."}
{"id": 3, "text": "Complete waste of money. Broke after two days."}

Process with async and semaphores

batch.py

import asyncio
import json
import os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["RADIUM_API_KEY"],
    base_url="https://api.radium.cloud/v1",
)

# Cap concurrent requests to protect downstream dependencies
SEMAPHORE = asyncio.Semaphore(10)  # adjust to your own safe concurrency level


async def classify_review(review: dict) -> dict:
    async with SEMAPHORE:
        response = await client.chat.completions.create(
            model="tycho-1.0",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Classify the sentiment of a product review. "
                        "Respond with exactly one word: Positive, Neutral, or Negative."
                    ),
                },
                {
                    "role": "user",
                    "content": review["text"],
                },
            ],
            max_tokens=10,
            temperature=0.0,
        )

        sentiment = response.choices[0].message.content.strip()
        usage = response.usage.model_dump()

        return {
            "id": review["id"],
            "sentiment": sentiment,
            "usage": usage,
        }


async def main():
    # Load reviews
    with open("reviews.jsonl") as f:
        reviews = [json.loads(line) for line in f]

    print(f"Processing {len(reviews)} reviews...")

    # Run all tasks concurrently (respecting the semaphore)
    results = await asyncio.gather(
        *[classify_review(r) for r in reviews]
    )

    # Save results
    with open("results.jsonl", "w") as f:
        for r in results:
            f.write(json.dumps(r) + "\n")

    # Summary
    total_tokens = sum(r["usage"]["total_tokens"] for r in results)
    print(f"Done. {len(results)} reviews processed, {total_tokens} total tokens.")


if __name__ == "__main__":
    asyncio.run(main())

Run it

export RADIUM_API_KEY="YOUR_RADIUM_API_KEY"
python batch.py

Output

results.jsonl

{"id": 1, "sentiment": "Positive", "usage": {"prompt_tokens": 42, "completion_tokens": 2, "total_tokens": 44}}
{"id": 2, "sentiment": "Neutral", "usage": {"prompt_tokens": 42, "completion_tokens": 2, "total_tokens": 44}}
{"id": 3, "sentiment": "Negative", "usage": {"prompt_tokens": 42, "completion_tokens": 2, "total_tokens": 44}}

Adding retries and backoff

Production scripts should handle transient failures:

batch_with_retries.py

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ["RADIUM_API_KEY"],
    base_url="https://api.radium.cloud/v1",
)


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60),
)
async def classify_with_retry(review: dict) -> dict:
    response = await client.chat.completions.create(
        model="tycho-1.0",
        messages=[
            {"role": "system", "content": "Classify sentiment as Positive, Neutral, or Negative."},
            {"role": "user", "content": review["text"]},
        ],
        max_tokens=10,
        temperature=0.0,
    )
    return {
        "id": review["id"],
        "sentiment": response.choices[0].message.content.strip(),
    }

Choosing a model

Task	Recommended model	Why
Simple classification / tagging	tycho-1.0	Fastest, cheapest, great at following short instructions
Summarization / moderate reasoning	clarke-1.0	Better comprehension, still fast
Complex analysis per item	hal-1.0	Best reasoning quality, use when accuracy is worth the cost

Next steps

Structured extraction

Extract JSON from each document

Rate limits

Understand limits and headers