ストリーミングを並行で束ねると、ノートPCの CPU が先に音を上げる — 実測値で並行本数を絞る適応スロットルの設計

import psutil
 
class CpuSampler:
    """ホスト CPU 使用率を EWMA でならして返す。"""
 
    def __init__(self, alpha: float = 0.3):
        self.alpha = alpha
        self.value = 0.0
        psutil.cpu_percent(interval=None)  # 初回呼び出しは捨て値なので空読み
 
    def sample(self) -> float:
        raw = psutil.cpu_percent(interval=None)
        self.value = self.alpha * raw + (1 - self.alpha) * self.value
        return self.value

import asyncio
from contextlib import asynccontextmanager
 
class AdaptiveCpuGate:
    """ホスト CPU の実測値に応じて同時実行本数を上下させるゲート。"""
 
    def __init__(self, target_cpu=70.0, min_slots=1, max_slots=16, alpha=0.3):
        self.target_cpu = target_cpu
        self.min_slots = min_slots
        self.max_slots = max_slots
        self._slots = min_slots          # 現在許可している同時本数
        self._in_flight = 0              # 実行中の本数
        self._sampler = CpuSampler(alpha)
        self._cond = asyncio.Condition()
 
    async def governor(self, stop: asyncio.Event, period: float = 2.0):
        """並行数を定期的に見直す常駐ループ。"""
        while not stop.is_set():
            cpu = self._sampler.sample()
            async with self._cond:
                if cpu > self.target_cpu:
                    self._slots = max(self.min_slots, self._slots - 1)
                elif cpu < self.target_cpu * 0.8:
                    self._slots = min(self.max_slots, self._slots + 1)
                self._cond.notify_all()   # 枠が増えた可能性を待機者に通知
            await asyncio.sleep(period)
 
    @asynccontextmanager
    async def slot(self):
        async with self._cond:
            # 実行中本数が許可本数を下回るまで待つ（背圧）
            await self._cond.wait_for(lambda: self._in_flight < self._slots)
            self._in_flight += 1
        try:
            yield
        finally:
            async with self._cond:
                self._in_flight -= 1
                self._cond.notify_all()

async def run_one(site, gate, client):
    async with gate.slot():
        async with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            messages=[{"role": "user", "content": build_prompt(site)}],
        ) as stream:
            async for _ in stream.text_stream:
                pass
        return await stream.get_final_message()
 
async def main(sites, client):
    gate = AdaptiveCpuGate(target_cpu=70.0, min_slots=2, max_slots=12)
    stop = asyncio.Event()
    gov = asyncio.create_task(gate.governor(stop))
    try:
        await asyncio.gather(*(run_one(s, gate, client) for s in sites))
    finally:
        stop.set()
        await gov