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-FP8")
    args = parser.parse_args()

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

import time
import requests

BASE_URL = "https://api.valarhq.ai/v1"
API_KEY = "YOUR_KEY_HERE"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# 1. Submit a batch
batch_requests = [
    {
        "custom_id": f"review-{i}",
        "params": {
            "model": "zai-org/GLM-5.2-FP8",
            "max_output_tokens": 16,
            "input": [
                {
                    "role": "user",
                    "content": (
                        "Classify the sentiment of this review as positive, "
                        f"negative, or neutral. Review {i}: Sample review body {i}."
                    ),
                }
            ],
            "metadata": {"completion_window": "standard"},
        },
    }
    for i in range(100)
]

resp = requests.post(
    f"{BASE_URL}/batches",
    headers=HEADERS,
    json={
        "endpoint": "/v1/responses",
        "label": "review-sentiment",
        "requests": batch_requests,
    },
)
batch = resp.json()
batch_id = batch["id"]
print(f"Created batch {batch_id} with {batch['request_counts']} requests")

# 2. Poll until every request has settled
while True:
    status = requests.get(f"{BASE_URL}/batches/{batch_id}", headers=HEADERS).json()
    request_status = status["request_status"]
    done = sum(
        1 for r in request_status.values()
        if r["status"] in ("COMPLETED", "FAILED", "CANCELLED")
    )
    print(f"Progress: {done}/{len(request_status)}")
    if done == len(request_status):
        break
    time.sleep(5)

# 3. Retrieve results
for custom_id, info in request_status.items():
    if info["status"] != "COMPLETED":
        continue
    result = requests.get(
        f"{BASE_URL}/batches/{batch_id}/{custom_id}", headers=HEADERS
    ).json()
    output_text = "".join(
        c["text"]
        for item in result.get("output", [])
        for c in item.get("content", [])
        if c.get("type") == "output_text"
    )
    print(f"{custom_id}: {output_text.strip()}")