> ## Documentation Index
> Fetch the complete documentation index at: https://docs.radium.cloud/llms.txt
> Use this file to discover all available pages before exploring further.

# 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:

```jsonl reviews.jsonl theme={null}
{"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

```python batch.py theme={null}
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

```bash theme={null}
export RADIUM_API_KEY="YOUR_RADIUM_API_KEY"
python batch.py
```

## Output

```jsonl results.jsonl theme={null}
{"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:

```python batch_with_retries.py theme={null}
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

<CardGroup cols={2}>
  <Card title="Structured extraction" icon="table" href="/examples/structured-extraction">Extract JSON from each document</Card>
  <Card title="Rate limits" icon="gauge-high" href="/core-concepts/rate-limits">Understand limits and headers</Card>
</CardGroup>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.