Concurrent Model Requests with Asyncio Limits

Good to see you again. Previously, you made Python components dependable at failure boundaries and verified those contracts with pytest: isolated filesystem state, parameterized edge cases, and mocked network behavior.

Now we add a different kind of reliability concern: many requests in flight at once. AI applications routinely batch embeddings, call a hosted LLM for several documents, fan out retrieval work, or serve several users against a resource-constrained local model. Sending every request at once can overwhelm the provider, your connection pool, or your GPU-backed inference server. Sending every request one at a time wastes time while waiting on I/O.

In this lesson, you will use asyncio tasks, gather, and Semaphore to run model-style requests concurrently while ensuring that no more than a chosen number are active at once.

---

Concurrency is useful when the work spends time waiting

A network model request is usually I/O-bound. Your program sends a request, then waits for DNS, a connection, bytes across the network, queueing at the model provider, and a response. During much of that period, Python does not need the CPU to advance that particular request.

Synchronous code waits for each request before beginning the next:

async def run_sequentially(prompts, call_model):
    results = []

    for prompt in prompts:
        result = await call_model(prompt)
        results.append(result)

    return results

Although this is inside an async def, it is still sequential: the next iteration does not begin until the prior call finishes.

With concurrency, multiple requests can be outstanding. When one task reaches an await and must wait for I/O, the event loop can give another ready task a turn. This is not automatically CPU parallelism; it is cooperative scheduling that is especially effective for non-blocking I/O.

Three terms need to stay distinct:

Term	Meaning in this lesson
Coroutine	The object produced when calling an async def function. It describes asynchronous work but has not necessarily been scheduled yet.
Task	A coroutine scheduled on the event loop, typically with asyncio.create_task().
Await	A point where the current coroutine waits for a result and lets the event loop run other ready tasks.

A useful batch shape is:

tasks = [
    asyncio.create_task(call_model(prompt))
    for prompt in prompts
]

results = await asyncio.gather(*tasks)

create_task() schedules all the calls. asyncio.gather() waits until they complete and returns results in the same order as the input tasks, even if request 7 finishes before request 2.

That ordering property is useful when each result must remain associated with a source document, user input, or database record. Do not infer completion order from the returned list.

---

Why “all at once” is not a safe default

Suppose an ingestion job must generate summaries for 200 documents. If you create 200 model-call tasks with no limit, you may have 200 requests attempting to use the same external resource.

That can cause several failures:

• A hosted provider may return rate-limit errors such as HTTP 429.
• Your local model server may run out of memory or accumulate an unhealthy queue.
• A connection pool can be exhausted.
• A downstream system that stores or evaluates results may become the bottleneck.
• A retry policy can amplify the overload if every failed request retries simultaneously.

The relevant question is usually not “can Python create 200 tasks?” It can. The question is: how many active model requests can this specific resource safely sustain?

A semaphore is an admission gate with a fixed number of permits. A semaphore initialized with $3$ permits allows three tasks into the guarded section. A fourth task waits until one of the first three leaves that section and returns its permit.

The core pattern is compact:

limit = asyncio.Semaphore(3)

async def limited_call(prompt, call_model):
    async with limit:
        return await call_model(prompt)

Entering async with limit acquires a permit. Leaving the block releases it, including when call_model() raises an exception or the task is cancelled. That cleanup guarantee is why this form should be preferred over hand-written acquire() and release() calls.

The location of the semaphore matters. Put the constrained operation inside its block:

async with limit:
    response = await client.post("/generate", json=payload)

Prompt construction, local validation, and inexpensive result parsing generally do not need to hold a scarce model-request permit. The network call, and any work the model server performs on its behalf, does.

---

A semaphore limits concurrency, not request rate

These concepts are related but different:

• A concurrency limit bounds how many requests are currently active.
• A rate limit bounds how many requests begin during an interval, such as 60 requests per minute.
• A queue limit bounds how much pending work your process will accept before applying backpressure or rejecting work.

A semaphore solves the first problem. It does not promise a specific number of requests per second.

If each request takes $200$ milliseconds and the semaphore allows three active requests, the service could complete roughly three requests every $200$ milliseconds under stable conditions. If requests finish more quickly, more can start quickly. If a provider enforces a strict per-minute quota, you need a rate-limiting policy in addition to the semaphore.

The following short demonstration shows exactly what a semaphore controls:

You may see examples that call limit.locked() and then add await asyncio.sleep(...). locked() can be useful for diagnostics, but it is not a complete rate limiter. Adding an arbitrary sleep while holding a permit also reduces throughput and makes behavior harder to reason about. Keep the base concurrency mechanism simple unless you have a measured rate-limit requirement.

---

Build a reusable concurrency-limited model batch

The following implementation models a reusable client boundary. It receives prompts and an asynchronous callable that performs one model request.

# app/batching.py
import asyncio


async def run_model_batch(prompts, call_model, *, max_concurrency):
    """Run model calls concurrently without exceeding max_concurrency."""

    if max_concurrency < 1:
        raise ValueError("max_concurrency must be at least 1")

    limit = asyncio.Semaphore(max_concurrency)

    async def run_one(prompt):
        async with limit:
            return await call_model(prompt)

    tasks = [
        asyncio.create_task(run_one(prompt), name=f"model-request-{index}")
        for index, prompt in enumerate(prompts)
    ]

    try:
        return await asyncio.gather(*tasks)

    except BaseException:
        # gather propagates the first failure, while unfinished tasks may
        # otherwise continue. Cancel the rest before propagating the failure.
        for task in tasks:
            if not task.done():
                task.cancel()

        await asyncio.gather(*tasks, return_exceptions=True)
        raise

Read this from the inside outward:

1. run_one() defines the work for one prompt.
2. async with limit makes a task wait if all permits are in use.
3. The call to call_model(prompt) runs only after the task holds a permit.
4. create_task() schedules one run_one() task per prompt.
5. gather() waits for the batch and returns the completed strings in prompt order.

The try and except establish an explicit all-or-nothing batch policy. By default, asyncio.gather() propagates the first exception to its caller, but unfinished tasks are not automatically cancelled merely because another task failed. Here, if one request fails, the remaining outstanding tasks are cancelled and awaited before the original exception is re-raised.

Catching BaseException is deliberate and narrow here: it includes cancellation, allowing the cleanup block to cancel and await sibling tasks before cancellation continues upward. In ordinary application code, do not broadly catch BaseException to handle business failures.

A deterministic model simulator

Before connecting this to a real API, use a simulated async model to make the concurrency behavior visible without network variability:

# app/demo_batching.py
import asyncio

from app.batching import run_model_batch


class SimulatedModel:
    def __init__(self):
        self.active_requests = 0
        self.peak_active_requests = 0

    async def __call__(self, prompt):
        self.active_requests += 1
        self.peak_active_requests = max(
            self.peak_active_requests,
            self.active_requests,
        )

        try:
            # Simulates non-blocking network or model-server waiting.
            await asyncio.sleep(0.2)
            return f"Summary: {prompt.upper()}"
        finally:
            self.active_requests -= 1


async def main():
    model = SimulatedModel()

    prompts = [
        "first document",
        "second document",
        "third document",
        "fourth document",
        "fifth document",
        "sixth document",
        "seventh document",
        "eighth document",
    ]

    results = await run_model_batch(
        prompts,
        model,
        max_concurrency=3,
    )

    for result in results:
        print(result)

    print(f"Peak active requests: {model.peak_active_requests}")


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

Run it from the project root:

python -m app.demo_batching

The output order follows the input prompt order. The peak should be 3, never 4, despite having eight tasks. Because eight requests take about $0.2$ seconds each and only three are active at a time, the batch should take roughly three waves of work rather than eight sequential waits.

The simulator uses asyncio.sleep(), not time.sleep(). time.sleep() blocks the event-loop thread and prevents every other coroutine from progressing. In asynchronous code, a blocking call hidden inside a supposedly async model client can eliminate the benefit of concurrency.

---

Connect the batch runner to an async HTTP client

In an AI service, call_model would usually be a closure around a shared asynchronous client. For example, with an HTTPX-compatible model endpoint:

import httpx

from app.batching import run_model_batch


async def summarize_documents(documents):
    async with httpx.AsyncClient(
        base_url="https://model-provider.example",
        timeout=30.0,
    ) as client:

        async def call_model(document):
            response = await client.post(
                "/v1/chat/completions",
                json={
                    "model": "example-model",
                    "messages": [
                        {
                            "role": "user",
                            "content": f"Summarize this document:\n{document}",
                        }
                    ],
                },
            )
            response.raise_for_status()

            payload = response.json()
            return payload["choices"][0]["message"]["content"]

        return await run_model_batch(
            documents,
            call_model,
            max_concurrency=4,
        )

There are two important resource-lifecycle decisions here:

• One httpx.AsyncClient is reused for the batch, allowing connection reuse rather than creating a new client and TCP connection per request.
• The client is closed by its async with block only after run_model_batch() finishes.

At this stage, a non-success HTTP response or network timeout causes call_model() to raise. The batch runner then cancels remaining work and propagates the failure. That is appropriate for a job where every result is required, such as generating artifacts for a release.

Some workflows instead allow partial success: perhaps 98 of 100 embeddings are useful and the two failures should be recorded for retry. Do not silently mix strings and exception objects in a result list without a clear contract. A better design is to return a typed success-or-failure result per input, with the error represented explicitly. You have already used dataclasses and Pydantic models earlier in this module; those are the right tools for defining that result contract.

---

Verify the limit without a real provider

The previous lesson emphasized that unit tests should avoid live upstream dependencies. The SimulatedModel gives you a deterministic test double that records its own peak concurrency.

# tests/unit/test_batching.py
import asyncio

from app.batching import run_model_batch
from app.demo_batching import SimulatedModel


def test_batch_never_exceeds_configured_concurrency():
    model = SimulatedModel()

    results = asyncio.run(
        run_model_batch(
            ["a", "b", "c", "d", "e", "f", "g"],
            model,
            max_concurrency=3,
        )
    )

    assert results == [
        "Summary: A",
        "Summary: B",
        "Summary: C",
        "Summary: D",
        "Summary: E",
        "Summary: F",
        "Summary: G",
    ]
    assert model.peak_active_requests <= 3
    assert model.peak_active_requests == 3

The first concurrency assertion checks the safety contract. The second confirms that the test exercised actual overlap rather than accidentally running sequentially.

For application code already running inside an event loop, such as an async FastAPI endpoint, call await run_model_batch(...) directly. Reserve asyncio.run(...) for your top-level script entry point or a synchronous test like the one above. Do not call asyncio.run() from inside a running event loop.

---

Scope the limit to the resource you are protecting

The batch runner above creates a new semaphore for each batch. That means it limits concurrency within one call to run_model_batch().

If a web service handles five user requests simultaneously and each creates a batch with a limit of four, the actual service could send up to 20 model requests at once. That may be correct for a provider with ample capacity, but it does not protect one shared local model server with a safe limit of four.

For a shared constrained resource, create one semaphore during application startup and pass it into the model-client layer used by all requests. The limit should be scoped to the resource:

Resource being protected	Appropriate semaphore scope
One offline document batch	Local to that batch
One shared hosted-provider API key	Shared by the application’s model client
One local inference server or GPU	Shared by every request targeting that server
Per-tenant capacity allocation	One semaphore per tenant, plus possibly a global limit

Choose the initial limit from evidence rather than intuition: provider documentation, local-model benchmarks, observed latency, error rates, and memory usage. A higher limit is not necessarily faster. Once the server saturates, it can increase queueing delay, trigger errors, or degrade every user’s experience.

A semaphore also limits active work, not the number of task objects created. A batch of a few hundred tasks is often reasonable; a stream containing millions of items needs a bounded worker queue and backpressure. You will return to queues, backpressure, and resource-aware API design in the production architecture module.

---

Key takeaways

asyncio lets an AI application make progress on other I/O-bound requests while one request is waiting, but concurrency needs an explicit resource policy.

• Calling an async def creates a coroutine; asyncio.create_task() schedules it.
• asyncio.gather() waits for a batch and returns results in input-task order.
• asyncio.Semaphore(n) permits at most $n$ tasks inside its guarded block.
• Use async with semaphore so permits are released correctly during errors and cancellation.
• A semaphore limits in-flight concurrency, not requests per minute and not queue size.
• Reuse an asynchronous HTTP client for a batch; avoid blocking calls such as time.sleep() or synchronous requests inside the event loop.
• Decide explicitly whether one failed model call should fail the whole batch or yield a structured partial-result record.
• Test the concurrency contract with a deterministic async fake that records peak active requests.
• Scope a semaphore according to the resource it protects; a per-batch limit does not automatically protect a whole web service.

This completes the Python foundations module. Next, you will move into the numerical and data foundations behind machine learning, beginning with NumPy arrays: shapes, axes, indexing, and broadcasting.