Sending Requests at Scale

pip install tqdm 'openai[aiohttp]'

import argparse
import asyncio

from tqdm import tqdm
from openai import AsyncOpenAI, DefaultAioHttpClient

REVIEW_TEMPLATE = (
    "Classify the sentiment of this product review as positive, negative, or "
    "neutral. Reply with one word.\n\nReview {i}: {text}"
)


async def main(num_requests: int, max_output_tokens: int, model: str):
    async with AsyncOpenAI(
	    base_url="https://api.valarhq.ai/v1", # or read OPENAI_BASE_URL from the environment
    	api_key="YOUR_VALAR_API_KEY",  # or read OPENAI_API_KEY from the environment
        http_client=DefaultAioHttpClient(),
    ) as client:
        submit_sem = asyncio.Semaphore(200)
        submit_bar = tqdm(total=num_requests, desc="Submitting")

        async def submit(i):
            content = REVIEW_TEMPLATE.format(
                i=i, text=f"Sample review body number {i} goes here."
            )
            async with submit_sem:
                response = await client.responses.create(
                    model=model,
                    input=[{"role": "user", "content": content}],
                    max_output_tokens=max_output_tokens,
                    background=True,
                )
            submit_bar.update(1)
            return response.id

        response_ids = list(await asyncio.gather(*[submit(i) for i in range(num_requests)]))
        submit_bar.close()
        print(f"Submitted {len(response_ids)} requests")

        # Second pass: poll each background response until it completes.
        completed = {}
        poll_sem = asyncio.Semaphore(200)
        poll_bar = tqdm(total=len(response_ids), desc="Polling")

        async def poll(response_id):
            for _ in range(3600):  # up to ~1 hour
                async with poll_sem:
                    response = await client.responses.retrieve(response_id)
					if response.status in {"queued", "in_progress"}:
					    await asyncio.sleep(1)
						continue

                    completed[response_id] = response
					poll_bar.update(1)
					return				
            print(f"Timed out waiting for {response_id}")

        await asyncio.gather(*[poll(rid) for rid in response_ids])
        poll_bar.close()

        for response_id in response_ids:
            resp = completed[response_id]
            text = "".join(
                c.text
                for item in resp.output
                for c in getattr(item, "content", []) or []
                if getattr(c, "type", None) == "output_text"
            )
            print(f"{response_id}: {text.strip()}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--num-requests", type=int, default=20000)
    parser.add_argument("--max-output-tokens", type=int, default=16)
    parser.add_argument("--model", type=str, default="zai-org/GLM-5.2")
    args = parser.parse_args()

    asyncio.run(
        main(
            num_requests=args.num_requests,
            max_output_tokens=args.max_output_tokens,
            model=args.model,
        )
    )

import json
import time

from openai import OpenAI

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

# 1. Write the requests file - one request per line
with open("requests.jsonl", "w") as f:
    for i in range(100):
        f.write(json.dumps({
            "custom_id": f"review-{i}",
            "method": "POST",
            "url": "/v1/chat/completions",
            "body": {
                "model": "zai-org/GLM-5.2",
                "max_tokens": 16,
                "messages": [{
                    "role": "user",
                    "content": (
                        "Classify the sentiment of this review as positive, "
                        f"negative, or neutral. Review {i}: Sample review body {i}."
                    ),
                }],
            },
        }) + "\n")

# 2. Upload it and create the batch
batch_input = client.files.create(file=open("requests.jsonl", "rb"), purpose="batch")
batch = client.batches.create(
    input_file_id=batch_input.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
print(f"Created batch {batch.id}")

# 3. Poll until the batch settles
while batch.status not in ("completed", "failed", "expired"):
    time.sleep(30)
    batch = client.batches.retrieve(batch.id)
    counts = batch.request_counts
    print(f"{batch.status}: {counts.completed + counts.failed}/{counts.total}")

# 4. Download the results - one JSONL line per request
for line in client.files.content(batch.output_file_id).text.splitlines():
    result = json.loads(line)
    body = result["response"]["body"]
    text = body["choices"][0]["message"]["content"]
    print(f"{result['custom_id']}: {text.strip()}")