High-Concurrency Batch Requests with KiosAPI and asyncio

Run thousands of concurrent KiosAPI requests for batch evaluation or dataset generation using Python asyncio and aiohttp for maximum throughput.

When you need to process large volumes of requests — for batch evaluation, dataset generation, or production workloads — use Python’s asyncio with aiohttp to run thousands of concurrent calls against KiosAPI from a single script. This approach gives you connection pooling, fine-grained flow control, and the throughput needed for demanding pipelines.

Installation

Install the required packages:

pip install asyncio aiohttp

httpx is less efficient than aiohttp for high-concurrency workloads. For best performance, use aiohttp with a single shared ClientSession as shown below.

Full Example

The example below demonstrates the recommended pattern: a shared ClientSession for connection pooling, a Semaphore to cap in-flight requests, and a while True loop that continuously drains your prompt source until it is exhausted.

import asyncio
import aiohttp
import json

# ---- Configuration ----
API_KEY = "sk-xxx"
BASE_URL = "https://kiosapi.com/v1/chat/completions"
MODEL = "gpt-4o-mini"
MAX_CONCURRENCY = 2000  # concurrent in-flight requests


async def create_completion(session, prompt):
    """Send a single chat completion request."""
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": MODEL,
        "messages": [
            {"role": "user", "content": prompt},
        ],
    }

    async with session.post(BASE_URL, headers=headers, json=payload) as resp:
        data = await resp.json()
        return data["choices"][0]["message"]["content"]


async def main():
    # Shared session — reuse across all requests for connection pooling
    connector = aiohttp.TCPConnector(limit=MAX_CONCURRENCY)
    timeout = aiohttp.ClientTimeout(total=120)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

        async def bounded_completion(prompt):
            async with semaphore:
                return await create_completion(session, prompt)

        # Example: continuously process a stream of prompts
        while True:
            # Replace this with your actual prompt source
            prompts = generate_prompts()  # e.g. read from a queue or file
            if not prompts:
                break

            tasks = [bounded_completion(p) for p in prompts]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for result in results:
                if isinstance(result, Exception):
                    print(f"Error: {result}")
                else:
                    print(result)


def generate_prompts():
    """Yield the next batch of prompts. Replace with your logic."""
    # Example: read from a file, message queue, or any iterable source
    return ["Summarize this article: ..."]


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

Key Design Points

Three architectural decisions make this pattern reliable at scale:

• Single shared ClientSession — create one aiohttp.ClientSession and reuse it for every request. This enables HTTP connection pooling and dramatically reduces per-request overhead compared to opening a new session each time.
• Semaphore for concurrency control — caps the number of in-flight requests at MAX_CONCURRENCY, preventing resource exhaustion on both the client and the server.
• while True loop — continuously pulls the next batch of prompts until your source is exhausted. Replace generate_prompts() with your own logic: a file reader, message queue consumer, or any other iterable source.

Limits and Best Practices

Do not exceed 5,000 concurrent requests in a single script. Beyond this threshold, Python’s single-threaded event loop and OS socket limits become bottlenecks, leading to timeouts and degraded throughput.

If you need more than 5,000 concurrent requests, run multiple scripts in parallel instead of raising the concurrency limit of a single process. For example, run three scripts each with MAX_CONCURRENCY = 2000 to reach ~6,000 concurrent requests across separate processes.

Approach	Max concurrent (per process)	Recommendation
aiohttp + asyncio	up to ~5,000	✅ Best performance
httpx async	up to ~5,000	⚠️ Less efficient, higher overhead
requests (sequential)	1	❌ Not suitable for batch

Error Handling

At high concurrency, individual request failures are expected. Always pass return_exceptions=True to asyncio.gather and implement retries with exponential backoff for transient errors such as 429, 500, 502, and 503.

async def create_completion_with_retry(session, prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await create_completion(session, prompt)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s backoff