並列エージェントの同時実行数を固定しない — 観測した詰まりから自動で増減させる設計

import time
from dataclasses import dataclass, field
 
 
@dataclass
class AIMDController:
    """同時実行数を観測に応じて増減させるコントローラ。"""
 
    min_limit: int = 1
    max_limit: int = 12
    limit: float = 2.0           # 現在の許容同時実行数（実数で保持し、丸めて使う）
    increase_step: float = 0.5   # 成功ごとの加算（加法的増加）
    decrease_factor: float = 0.5 # 詰まり検知時の乗算（乗法的減少）
    cooldown_s: float = 8.0      # 減少直後に再増加を抑える猶予
    _last_decrease: float = field(default=0.0)
 
    def on_success(self) -> None:
        # 減らした直後は、しばらく増やさない。揺り戻しを防ぐため。
        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))

class ThrottleError(Exception):
    """下流が明確に過負荷を示したときに送出する。"""
 
 
def classify(status_code: int, headers: dict) -> str:
    if status_code == 429 or status_code == 503:
        return "throttled"
    # 残量警告は「これ以上増やすな」のサイン。減らすほどではない。
    remaining = headers.get("x-quota-remaining")
    if remaining is not None and int(remaining) < 5:
        return "near_limit"
    return "ok"

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:
            # 現在の許容数を毎回読み直すので、上限が縮んだら即座に効く。
            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() は (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()
            # near_limit のときは増やしも減らしもしない
            return status, headers, body
        except (ThrottleError, asyncio.TimeoutError):
            self.ctrl.on_throttle()
            raise
        finally:
            await self._release()
            # 上限が減った可能性があるので、待機者を起こして再評価させる
            async with self._cond:
                self._cond.notify_all()