非ストリーミングの長い応答が10分の壁で黙って二重課金されていた — SDK既定のタイムアウトと自動リトライを設計し直す

from anthropic import Anthropic
 
client = Anthropic()
 
def summarize_streaming(system_prompt: str, user_text: str) -> str:
    """長い出力はストリーミングで受ける。読み取りタイムアウトを回避する。"""
    chunks = []
    with client.messages.stream(
        model="claude-haiku-4-5",
        max_tokens=16000,
        system=system_prompt,
        messages=[{"role": "user", "content": user_text}],
    ) as stream:
        for text in stream.text_stream:
            chunks.append(text)
        final = stream.get_final_message()
 
    # 途切れの検知も stop_reason で必ず行う
    if final.stop_reason == "max_tokens":
        raise RuntimeError("出力が max_tokens で打ち切られました。続きの取得が必要です")
 
    return "".join(chunks)

import time
from anthropic import Anthropic, APITimeoutError, APIConnectionError
 
client = Anthropic()
 
def create_long_safe(system_prompt: str, user_text: str, job_id: str, ledger) -> str:
    """非ストリーミングの長い要求を、二重課金を避けて実行する。"""
 
    # すでに走らせた記録があれば、盲目的な再送を止める
    if ledger.is_inflight_or_done(job_id):
        raise RuntimeError(f"job {job_id} は既に実行済み/実行中。手動確認が必要です")
 
    ledger.mark_inflight(job_id)
 
    # このタスクは長いので timeout を十分に広げ、自動リトライは切る
    scoped = client.with_options(timeout=1800.0, max_retries=0)
 
    try:
        msg = scoped.messages.create(
            model="claude-haiku-4-5",
            max_tokens=16000,
            system=system_prompt,
            messages=[{"role": "user", "content": user_text}],
        )
    except APITimeoutError:
        # タイムアウトは「未確定」。自分では絶対に自動再送しない
        ledger.mark_uncertain(job_id)
        raise
    except APIConnectionError:
        # 接続確立の失敗は未処理が濃厚。ここは再送してよい種類
        ledger.mark_failed(job_id)
        raise
 
    ledger.mark_done(job_id, input_tokens=msg.usage.input_tokens,
                     output_tokens=msg.usage.output_tokens)
    return msg.content[0].text

import json
import time
from pathlib import Path
 
class JobLedger:
    """要求の一生を追う軽量台帳。二重投入の最後の砦。"""
 
    def __init__(self, path: str):
        self.path = Path(path)
        self._state = {}
        if self.path.exists():
            for line in self.path.read_text().splitlines():
                rec = json.loads(line)
                self._state[rec["job_id"]] = rec["status"]
 
    def _append(self, job_id: str, status: str, **extra):
        rec = {"job_id": job_id, "status": status, "ts": time.time(), **extra}
        with self.path.open("a") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        self._state[job_id] = status
 
    def is_inflight_or_done(self, job_id: str) -> bool:
        # uncertain と done と inflight は、機械が勝手に再送してはいけない状態
        return self._state.get(job_id) in {"inflight", "uncertain", "done"}
 
    def mark_inflight(self, job_id): self._append(job_id, "inflight")
    def mark_uncertain(self, job_id): self._append(job_id, "uncertain")
    def mark_failed(self, job_id): self._append(job_id, "failed")
    def mark_done(self, job_id, **usage): self._append(job_id, "done", **usage)