並列エージェントが同じタスクを二度実行して請求が膨らんでいたとき — 重複を計測して止める運用メモ

import time
 
class IdempotencyGuard:
    def __init__(self, store, ttl_seconds: int = 900):
        self.store = store
        self.ttl = ttl_seconds
 
    def claim(self, fp: str, worker_id: str) -> bool:
        """予約を先取りできたら True。既に誰かが実行中/実行済みなら False。"""
        key = f"claim:{fp}"
        # NX: キーが無いときだけ書き込む（原子的）。これが競合の要。
        acquired = self.store.set(key, worker_id, nx=True, ex=self.ttl)
        return bool(acquired)
 
    def save_result(self, fp: str, result: dict) -> None:
        self.store.set(f"result:{fp}", json.dumps(result), ex=self.ttl)
 
    def wait_result(self, fp: str, timeout: float = 30.0) -> dict | None:
        """予約を取れなかった側が、先行実行の結果を待つ。"""
        deadline = time.time() + timeout
        while time.time() < deadline:
            raw = self.store.get(f"result:{fp}")
            if raw is not None:
                return json.loads(raw)
            time.sleep(0.5)
        return None  # タイムアウト時はオーケストレーターへ判断を戻す
 
 
def run_task(worker_id, task_type, payload, guard, execute_fn):
    fp = task_fingerprint(task_type, payload)
    if guard.claim(fp, worker_id):
        result = execute_fn(payload)   # 実際の LLM 実行はここだけ
        guard.save_result(fp, result)
        return {"source": "executed", "result": result}
    # 予約を取れなかった＝別ワーカーが同じタスクを走らせている
    shared = guard.wait_result(fp)
    if shared is not None:
        return {"source": "reused", "result": shared}
    return {"source": "fallback", "result": execute_fn(payload)}

def cumulative_input_tokens(steps: int, avg_per_step: int = 1000) -> int:
    """各ステップが前ステップまでの履歴を含むため、入力は級数的に増える。"""
    # 1000*(1+2+...+steps) = 1000 * steps*(steps+1)/2
    return avg_per_step * steps * (steps + 1) // 2
 
for s in (1, 5, 10, 20):
    print(f"{s:2d}ステップ: 累積入力 {cumulative_input_tokens(s):>7,} tokens")
# 1ステップ: 累積入力   1,000 tokens
# 5ステップ: 累積入力  15,000 tokens
# 10ステップ: 累積入力  55,000 tokens
# 20ステップ: 累積入力 210,000 tokens

def log_cost(store, fp, worker_id, task_type, usage: dict):
    record = {
        "fp": fp,
        "worker": worker_id,
        "task_type": task_type,
        "input_tokens": usage["input_tokens"],
        "output_tokens": usage["output_tokens"],
        "ts": time.time(),
    }
    store.rpush(f"cost:{task_type}", json.dumps(record))
 
def task_type_summary(store, task_type) -> dict:
    rows = [json.loads(r) for r in store.lrange(f"cost:{task_type}", 0, -1)]
    runs = len(rows)
    in_tok = sum(r["input_tokens"] for r in rows)
    out_tok = sum(r["output_tokens"] for r in rows)
    return {
        "task_type": task_type,
        "runs": runs,
        "avg_input": in_tok // runs if runs else 0,
        "total_tokens": in_tok + out_tok,
    }