失敗を握りつぶさずに安く回す — Gemini 3.5 Flash を前提にしたエージェント再試行予算の設計

import time
from dataclasses import dataclass, field
 
@dataclass
class RetryBudget:
    max_attempts: int = 3        # 一過性でも3回まで
    max_cost_usd: float = 0.15   # このタスクが使ってよい上限
    max_seconds: float = 90.0    # 暴走を時間で止める
    spent_usd: float = 0.0
    started: float = field(default_factory=time.monotonic)
    attempts: int = 0
 
    def allow(self) -> bool:
        if self.attempts >= self.max_attempts:
            return False
        if self.spent_usd >= self.max_cost_usd:
            return False
        if time.monotonic() - self.started >= self.max_seconds:
            return False
        return True
 
def run_with_budget(run_once, classify, budget: RetryBudget):
    last_error = None
    while budget.allow():
        budget.attempts += 1
        result = run_once()          # エージェントを1回起動
        budget.spent_usd += result.cost_usd
        if result.ok:
            return result
        kind = classify(result.error) # transient / input / permanent
        if kind == "permanent":
            raise PermanentFailure(result.error)  # 再試行しない
        if kind == "transient":
            time.sleep(min(2 ** budget.attempts, 20))  # 指数バックオフ
        last_error = result.error
    raise BudgetExhausted(last_error)

import json, time
 
def log_attempt(task_id, attempt, kind, cost, ok):
    line = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "task": task_id,
        "attempt": attempt,
        "kind": kind,        # transient / input / permanent
        "cost_usd": round(cost, 4),
        "ok": ok,
    }
    with open("retry_log.jsonl", "a") as f:
        f.write(json.dumps(line, ensure_ascii=False) + "\n")