Stop Hard-Coding Your Agent Concurrency: Let It Tune Itself From What It Observes

Borrowing From TCP: Additive Increase, Multiplicative Decrease

The core borrowing is AIMD — Additive Increase, Multiplicative Decrease.

• Additive increase: while successes keep coming, add to the concurrency slowly, in a fixed step.
• Multiplicative decrease: the moment you see a congestion signal, cut it by a factor (halve it, say).

Cautious going up, decisive coming down. That asymmetry is the whole point. Congestion means you are starting to break something downstream, so easing down linearly arrives too late. Going up, by contrast, is only a probe for headroom — there is no need to hurry.

import time
from dataclasses import dataclass, field
 
 
@dataclass
class AIMDController:
    """Raises or lowers concurrency in response to observations."""
 
    min_limit: int = 1
    max_limit: int = 12
    limit: float = 2.0            # current allowed concurrency (kept as float, rounded to use)
    increase_step: float = 0.5   # added per success (additive increase)
    decrease_factor: float = 0.5 # multiplied on congestion (multiplicative decrease)
    cooldown_s: float = 8.0      # suppress re-increase right after a cut
    _last_decrease: float = field(default=0.0)
 
    def on_success(self) -> None:
        # Do not grow for a while after a cut, to avoid oscillation.
        if time.monotonic() - self._last_decrease < self.cooldown_s:
            return
        self.limit = min(self.max_limit, self.limit + self.increase_step)
 
    def on_throttle(self) -> None:
        self.limit = max(self.min_limit, self.limit * self.decrease_factor)
        self._last_decrease = time.monotonic()
 
    @property
    def concurrency(self) -> int:
        return max(self.min_limit, int(self.limit))

Keeping limit as a float, not an int, lets increase_step be smaller than 1. You can grow by one slot only every two successes — a slow climb that calms the thrashing you otherwise get near the ceiling.

What Counts as a Congestion Signal

What you feed on_throttle() decides the precision of the whole design. Here are the signals, strongest first.

Watching only the top row — the 429 — already works as a first step. Reach for latency only after you add a moving average so noise does not whip you around. Halving concurrency on one slow response tips you back into wasting capacity.

Put a classifier just outside the call to sort each response into success or congestion.

class ThrottleError(Exception):
    """Raised when downstream clearly signals overload."""
 
 
def classify(status_code: int, headers: dict) -> str:
    if status_code == 429 or status_code == 503:
        return "throttled"
    # A remaining-quota warning means "stop growing", not "shrink".
    remaining = headers.get("x-quota-remaining")
    if remaining is not None and int(remaining) < 5:
        return "near_limit"
    return "ok"

Wiring It Into an Adaptive Scheduler

The controller's concurrency changes from moment to moment, so a fixed-size semaphore cannot express it. You need a variable gate that re-checks the current limit every time.

import asyncio
 
 
class AdaptiveScheduler:
    def __init__(self, controller: AIMDController):
        self.ctrl = controller
        self._inflight = 0
        self._cond = asyncio.Condition()
 
    async def _acquire(self) -> None:
        async with self._cond:
            # Re-read the allowed count each pass, so a shrink takes effect at once.
            while self._inflight >= self.ctrl.concurrency:
                await self._cond.wait()
            self._inflight += 1
 
    async def _release(self) -> None:
        async with self._cond:
            self._inflight -= 1
            self._cond.notify_all()
 
    async def run(self, call):
        """call() is an async function returning (status_code, headers, body)."""
        await self._acquire()
        try:
            status, headers, body = await call()
            outcome = classify(status, headers)
            if outcome == "throttled":
                self.ctrl.on_throttle()
            elif outcome == "ok":
                self.ctrl.on_success()
            # on near_limit, neither grow nor shrink
            return status, headers, body
        except (ThrottleError, asyncio.TimeoutError):
            self.ctrl.on_throttle()
            raise
        finally:
            await self._release()
            # the ceiling may have dropped, so wake waiters to re-evaluate
            async with self._cond:
                self._cond.notify_all()

The point is that _acquire() re-reads self.ctrl.concurrency inside the loop. The instant on_throttle() drops the ceiling from 6 to 3, any task that wants a fresh slot waits right there. Work already running is never killed. By squeezing only "how many to start next," you lower the flow without breaking what is in flight.

Lay this over the Managed Agents API launch call and the number of isolated-environment agents you spin up at once adjusts itself from observation. The same holds for local parallel agent calls.

How This Differs From Backpressure, and How to Layer Them

"Limit the concurrency" sits right next to a backpressure design built from a bounded queue and a semaphore. The two do not compete; they play different roles.

In practice, layering both is the sound move. Let AIMD hunt for the daily optimum, and put an "under no circumstances exceed this" absolute cap on top as a fixed semaphore. max_limit already plays that part, but if you want to rule out cost accidents, keep one more hard human-chosen ceiling outside the controller. Adaptation is for optimization; the fixed cap is for accident prevention. Separating those two purposes keeps the design clear.

Tuning Against Over- and Under-Throttling

A few instincts help when picking the numbers.

Start with a decisive decrease and a modest increase. If you clear congestion quickly and surely, you can afford to idle a little and still protect downstream. As you settle in, raise increase_step to reclaim the idle slots.

One trap: combine retries and adaptive control naively and a task that took a 429 retries immediately, which calls on_throttle() again and cuts you twice over. Leave retries to exponential backoff and decide that congestion is reported to the controller only on the first occurrence. That keeps the two from crossing wires.

Field Notes

There was a stretch when, as an indie developer, I ran a pipeline that updates several sites automatically alongside the background processing for a handful of mobile apps, all in the same hours. At first concurrency was fixed, and I repeated both failures daily: a conservative value left obvious headroom unused overnight, then 429s cascaded the moment I started touching things during the day.

Dropping the constant and switching to a plain AIMD with the 429 as its only signal, the cascading evening failures went quiet first. The controller saw congestion and cut on its own, so the chore of manually lowering the value whenever I sat down disappeared. At night the opposite happened: with nobody competing, it climbed slowly to the ceiling and used the slots up without my attention. I did not add any clever prediction. I just handed the machine one thing — look at the result, decide whether to grow or shrink.

You could say it is nothing more than re-applying an idea long settled in control systems to a new target, agent parallelism. The newer the thing, the better the old, well-tested principles seem to fit. If you have been worn down by a fixed value, start with a small controller that watches only the 429.

Thank you for reading.