Files API を自動パイプラインに組み込んで二週間ほど経った頃、ふと GET /v1/files を叩いてみたら、同じ参照データが日付違いで何十個も並んでいて少し青ざめました。アップロードのコードは正しく動いていました。問題は「一度上げたファイルをもう一度上げない」仕組みと「使わなくなったファイルを片付ける」仕組みが、どちらも私のコードに無かったことです。
Files API は「一度アップロードして何度でも参照する」ための機能ですが、その"一度"を保証するのは API ではなく呼び出し側の責任です。手作業で時々使う分には溜まりませんが、毎日同じ参照データを上げ直す自動運用では、孤児ファイルとストレージ課金が静かに積み上がります。ここからは、その積み上がりを内容ハッシュ台帳と孤児 GC の二段で止める設計を、私が Dolice Labs の自動投稿パイプラインで実際に組んだ形をもとにたどります。
基礎的なアップロード手順はClaude Files API の基本操作ガイド で扱っているので、ここでは「溜めない運用」に絞ります。
なぜ Files API は黙って溜まるのか
Files API のアップロードは冪等ではありません。同じ内容の PDF を二回 POST /v1/files すると、別々の file_id を持つ二つの実体ができます。API から見れば別物なので、エラーにもならず警告も出ません。
自動運用ではここが効いてきます。たとえば「参照データを毎朝アップロードして、その file_id を使って記事を生成する」というジョブを毎日回すと、中身が前日と一字一句同じでも、新しい file_id が毎回生まれます。一ヶ月で 30 個、四つのサイト分なら 120 個。一つ一つは小さくても、参照されなくなった実体がストレージに残り続けます。
Files API のストレージは保管しているバイト数に対して課金されます。つまり「使っていないのに消していないファイル」は、何の役にも立たないまま毎日課金対象であり続けるわけです。私の場合、気づいたときには本来 4 個で足りるはずの参照データが 80 個近くまで膨らんでいました。
落とし穴は二つに分けて考えると整理できます。ひとつは重複アップロード （同じ内容を二度上げる）、もうひとつは孤児ファイル （もう誰も参照していないのに残っている）です。前者は上げる前に止め、後者は定期的に回収します。
内容ハッシュ台帳という考え方
重複を止める一番素直な方法は、「上げる前に、この内容を前に上げたか確認する」ことです。確認の鍵になるのが内容のハッシュです。
ファイルのバイト列から SHA-256 を計算し、ハッシュ → file_id の対応を自分側の台帳に持ちます。アップロードしたい内容のハッシュが台帳にあれば、既存の file_id をそのまま使い回します。無ければ初めてアップロードして、結果を台帳に書き戻します。これだけで「中身が同じなら絶対に二度上げない」が保証できます。
台帳の置き場所はワークロードによります。私は最初ローカルの JSON ファイルで始め、複数プロセスが同時に走るようになってから KV ストアに移しました。要件はシンプルで、ハッシュをキーに file_id を引ければ何でも構いません。
ハッシュに SHA-256 を選んだのは、衝突の心配を実務上ゼロとみなせるからです。中身の違うファイルが偶然同じハッシュになって取り違える確率は天文学的に小さく、参照データの規模では考慮に値しません。逆に MD5 のような古いハッシュを避けたのは、いざ衝突を疑う場面が来たときに、原因の切り分けで余計な時間を取られたくなかったからです。台帳という長く使う仕組みほど、土台は安全側に倒しておくと後が楽になります。
import hashlib import json from pathlib import Path from anthropic import Anthropic
client = Anthropic() # ANTHROPIC_API_KEY は環境変数から LEDGER_PATH = Path( "file_ledger.json" ) BETA = "files-api-2025-04-14" # Files API はベータヘッダーが必要
def load_ledger () -> dict : if LEDGER_PATH .exists(): return json.loads( LEDGER_PATH .read_text()) return {}
def save_ledger (ledger: dict ) -> None : # 一時ファイルに書いてから置き換え、書き込み途中の破損を防ぐ tmp = LEDGER_PATH .with_suffix( ".tmp" ) tmp.write_text(json.dumps(ledger, ensure_ascii = False , indent = 2 )) tmp.replace( LEDGER_PATH )
def content_hash (data: bytes ) -> str : return hashlib.sha256(data).hexdigest()
def get_or_upload (path: str ) -> str : """同じ内容なら既存の file_id を返し、初見ならアップロードする。""" data = Path(path).read_bytes() digest = content_hash(data)
ledger = load_ledger() if digest in ledger: return ledger[digest][ "file_id" ] # 再利用 — アップロードしない
uploaded = client.beta.files.upload( file = (Path(path).name, data, "application/octet-stream" ), betas = [ BETA ], ) ledger[digest] = { "file_id" : uploaded.id, "name" : Path(path).name} save_ledger(ledger) return uploaded.id このコードのポイントは、ハッシュ計算をファイル名ではなくバイト列 に対して行っていることです。ファイル名で判定すると、同名で中身が変わったケースを取りこぼします。逆に内容で判定すれば、ファイル名が違っても中身が同じなら一つにまとめられます。
台帳に「最終参照日」を持たせて孤児に備える
ここで一手間を加えておくと、後の GC が楽になります。台帳のエントリに「最後にこの file_id を使った日」を記録しておくのです。
import datetime
def touch (digest: str , ledger: dict ) -> None : """参照したことを台帳に刻む。""" if digest in ledger: ledger[digest][ "last_used" ] = datetime.date.today().isoformat() save_ledger(ledger) get_or_upload の中で、再利用したときもアップロードしたときも last_used を更新するようにします。こうしておくと、「直近で一度も参照されていない file_id」を台帳から機械的に拾えるようになり、GC の判断材料になります。
私はこの last_used を持たせていなかった時期に、まだ使うかもしれないファイルを勘で消してしまい、翌朝のジョブで file_id が見つからずに落ちる失敗をしました。消す根拠を台帳の側に持たせておくと、勘で消す判断をしなくて済みます。
孤児 GC — list と delete を慎重に組む
台帳が整ったら、孤児の回収に進みます。Files API の GET /v1/files はアカウント上の全ファイルを返します。これと自分の台帳を突き合わせれば、「API 上には存在するが、自分の台帳が現役だと認めていないファイル」＝孤児候補が見つかります。
ここで最も大事なのは、台帳に載っていないからといって即座に消さない ことです。別のプロセスや別のツールが上げたファイル、あるいはアップロード直後でまだ台帳に書かれていないファイルを巻き込むと、本番運用のジョブを壊します。この取り違えへの対処として、私は次の二重チェックを必ず通します。
ひとつ、削除対象は「台帳が現役と認めた file_id の集合に含まれない」こと。ふたつ、作成からの経過時間が十分に長いこと（アップロード直後の取り違えを避けるため、私は 24 時間を下限にしています）。
import datetime
KEEP_DAYS = 14 # この日数以内に参照したものは現役とみなす MIN_AGE_HOURS = 24 # 作成直後のファイルは触らない
def active_file_ids (ledger: dict ) -> set : cutoff = datetime.date.today() - datetime.timedelta( days = KEEP_DAYS ) ids = set () for entry in ledger.values(): last = entry.get( "last_used" ) if last and datetime.date.fromisoformat(last) >= cutoff: ids.add(entry[ "file_id" ]) return ids
def collect_orphans (dry_run: bool = True ) -> list : ledger = load_ledger() keep = active_file_ids(ledger) now = datetime.datetime.now(datetime.timezone.utc) deleted = []
for f in client.beta.files.list( betas = [ BETA ]): if f.id in keep: continue # 現役なので保護 age = now - f.created_at if age < datetime.timedelta( hours = MIN_AGE_HOURS ): continue # 新しすぎるので保護 if dry_run: deleted.append(f.id) # 候補を返すだけ else : client.beta.files.delete(f.id, betas = [ BETA ]) deleted.append(f.id) return deleted dry_run=True を既定にしているのは意図的です。最初の数回は必ず候補一覧だけを出力し、本当に消えてよいファイルだけが並んでいるかを自分の目で確かめてから、dry_run=False に切り替えます。自動削除を最初から無人で走らせるのは避けることを強く推奨します。私はおすすめしません。一度だけ、検証なしで delete を回して、必要なファイルを巻き込んだことがあります。
台帳と実体がずれたときに直す
運用を続けると、台帳には載っているのに API 側に実体が無い、という逆方向のずれも起きます。誰かが手動でファイルを消した、TTL で失効した、別環境の台帳が混ざった、といった原因です。
このずれを放置すると、get_or_upload が「台帳にあるから再利用」と判断して死んだ file_id を返し、その file_id を使ったメッセージ送信が not found で落ちます。そこで、再利用する前に実体の存在を軽く確認する経路を足しておくと安全側に倒せます。
def resolve (path: str ) -> str : """再利用前に実体の存在を確認し、消えていれば上げ直す。""" data = Path(path).read_bytes() digest = content_hash(data) ledger = load_ledger()
if digest in ledger: fid = ledger[digest][ "file_id" ] try : client.beta.files.retrieve_metadata(fid, betas = [ BETA ]) touch(digest, ledger) return fid except Exception : del ledger[digest] # 死んだエントリを捨てる save_ledger(ledger) return get_or_upload(path) 毎回 retrieve_metadata を挟むと一往復ぶんのレイテンシが乗るので、私は「ジョブの先頭で一度だけ整合性を取り、以降は台帳を信頼する」運用にしています。どこで存在確認のコストを払うかは、ジョブの形に合わせて決めるのがよいと思います。
ストレージを定点観測する小さな習慣
最後に、設計そのものよりも地味に効いた習慣を一つ。週に一度、ファイルの総数と合計バイト数をログに出すだけのジョブを足しました。
def storage_snapshot () -> dict : files = list (client.beta.files.list( betas = [ BETA ])) total_bytes = sum (f.size_bytes for f in files) return { "count" : len (files), "total_mb" : round (total_bytes / 1024 / 1024 , 2 )} 数字が階段状に増え続けていたら、台帳か GC のどこかが効いていないサインです。私自身、この定点観測を入れてから、80 個近くまで膨らんでいたファイルを 4〜6 個で安定させられるようになりました。保管バイト数にして 90% 以上を削った計算です。コストの絶対額は小さくても、「使っていないものに払い続けている」状態が消えるのは、自動運用を長く続けるうえで気持ちの面でも軽くなります。
まず最初の一歩としては、今動いているパイプラインに対して GET /v1/files を一度叩いてみてください。そこに並ぶ数が想定より多ければ、この記事の台帳と GC を組む価値があります。