Google AI Pro と Ultra の損益分岐を実測する — 並列エージェント運用14日間のクォータ消費データから

#!/usr/bin/env python3
"""agent_quota_tracker.py — エージェント実行ログの日次集計"""
import json
import sys
import re
from pathlib import Path
from collections import defaultdict
from datetime import datetime
 
# レート制限を示すメッセージのパターン（観測に基づき随時追加）
RATE_LIMIT_PATTERNS = [
    re.compile(r"rate.?limit", re.IGNORECASE),
    re.compile(r"quota.+exceeded", re.IGNORECASE),
    re.compile(r"resource.?exhausted", re.IGNORECASE),
]
 
def is_rate_limited(text: str) -> bool:
    return any(p.search(text) for p in RATE_LIMIT_PATTERNS)
 
def collect(log_dir: Path) -> dict:
    daily = defaultdict(lambda: {
        "sessions": 0,
        "agent_runs": 0,
        "max_parallel": 0,
        "rate_limit_hits": [],
    })
    for jsonl in sorted(log_dir.rglob("*.jsonl")):
        active = []  # (start, end) のリストで並列度を概算
        for line in jsonl.read_text(encoding="utf-8").splitlines():
            try:
                ev = json.loads(line)
            except json.JSONDecodeError:
                continue
            ts = ev.get("timestamp", "")
            day = ts[:10] if ts else "unknown"
            kind = ev.get("type", "")
            if kind == "session_start":
                daily[day]["sessions"] += 1
            elif kind == "agent_run":
                daily[day]["agent_runs"] += 1
                start = ev.get("started_at", ts)
                end = ev.get("ended_at", ts)
                active.append((start, end))
            elif kind == "error" and is_rate_limited(str(ev.get("message", ""))):
                daily[day]["rate_limit_hits"].append(ts[11:16])
        # 同一時刻に重なる実行数の最大値を並列度とみなす
        for day, stats in daily.items():
            overlaps = [
                sum(1 for s2, e2 in active if s2 <= s1 < e2)
                for s1, _ in active
            ]
            if overlaps:
                stats["max_parallel"] = max(stats["max_parallel"], max(overlaps))
    return daily
 
def main():
    if len(sys.argv) < 2:
        print("usage: agent_quota_tracker.py <log_dir>")
        sys.exit(1)
    daily = collect(Path(sys.argv[1]))
    for day in sorted(daily):
        s = daily[day]
        hits = ",".join(s["rate_limit_hits"]) or "-"
        print(f"{day}  runs={s['agent_runs']:3d}  "
              f"parallel_max={s['max_parallel']}  limited_at={hits}")
 
if __name__ == "__main__":
    main()