Claude API のレート制限と向き合う
複数のブログを自動投稿で運用していると、夜間にまとめてリクエストを投げた瞬間だけ処理が詰まる、という現象に何度も出くわしました。原因を追っていくと、たいていはコードのバグではなくレート制限でした。Claude API を個人開発の自動化に組み込むなら、この上限の挙動を先に理解しておくほど、後で慌てずに済みます。
ここでは RPM・TPM・TPD の仕組みから、トークンの節約、リクエストの流量制御、そして上限に当たったときの再試行の作法までを、実際に運用しながら身につけた順序で整理します。
レート制限の仕組み
Claude API のレート制限は、API の安定性を確保し、すべてのユーザーが公平にリソースを利用できるようにするための仕組みです。
制限の種類
Claude API では 3 つの軸でレート制限が適用されます：
| 制限タイプ | 説明 | 単位 |
|---|---|---|
| RPM (Requests Per Minute) | 1 分間あたりのリクエスト数 | リクエスト数 |
| TPM (Tokens Per Minute) | 1 分間あたりの入出力トークン数 | トークン数 |
| TPD (Tokens Per Day) | 1 日あたりの入出力トークン数 | トークン数 |
プランごとの制限
レート制限はプランやモデルによって異なります。Anthropic のドキュメントで最新の制限値を確認してください。一般的に、上位プランではより高い制限が適用されます。
レスポンスヘッダーの活用
API レスポンスには現在のレート制限状況がヘッダーとして含まれます。これらを活用してリクエストの流量を制御できます：
RateLimit-Limit-Requests: 50
RateLimit-Limit-Tokens: 90000
RateLimit-Remaining-Requests: 35
RateLimit-Remaining-Tokens: 72000
RateLimit-Reset-Requests: 2026-03-09T12:00:30Z
RateLimit-Reset-Tokens: 2026-03-09T12:00:30Z
Retry-After: 15
トークン使用量の最適化
API コストとレート制限の両方を最適化するには、トークン使用量を効率的に管理する点が肝心です。
1. プロンプトの最適化
プロンプトを簡潔かつ効果的に設計することで、入力トークンを削減できます：
# 非効率なプロンプト（トークン消費が多い）
inefficient_prompt = """
以下のテキストを読んでください。テキストの内容を分析してください。
分析した結果を元に、テキストの要約を生成してください。
要約は簡潔にまとめてください。できるだけ短くしてください。
長い要約は避けてください。ポイントだけを抽出してください。
"""
# 最適化されたプロンプト（同じ意図でトークン削減）
efficient_prompt = """
以下のテキストを3文以内で要約してください。
"""2. max_tokens の適切な設定
max_tokens を必要最小限に設定することで、出力トークンを制御できます：
import anthropic
client = anthropic.Anthropic()
# 短い応答が期待される場合
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=256, # 短い応答には小さい値を設定
messages=[
{"role": "user", "content": "この単語を英訳してください: 効率化"}
]
)3. システムプロンプトの効率化
システムプロンプトは毎回のリクエストで送信されるため、簡潔さが重要です：
# 冗長なシステムプロンプト
verbose_system = """
あなたは優秀なアシスタントです。ユーザーの質問に対して丁寧に回答してください。
回答は正確で、わかりやすく、適切な長さにしてください。
必要に応じて例を交えて説明してください。不確かな場合はその旨を伝えてください。
ユーザーの意図を正確に汲み取り、最適な回答を提供してください。
"""
# 最適化されたシステムプロンプト
concise_system = "正確かつ簡潔に回答する技術アシスタント。不確かな場合は明示する。"4. トークンカウントの事前確認
リクエスト前にトークン数を見積もることで、レート制限超過を防げます：
import anthropic
client = anthropic.Anthropic()
# トークンカウント API を使用
count_response = client.messages.count_tokens(
model="claude-sonnet-4-6",
messages=[
{"role": "user", "content": "長いテキスト..."}
],
system="システムプロンプト"
)
print(f"入力トークン数: {count_response.input_tokens}")リクエスト効率化のテクニック
1. リクエストキューイング
リクエストをキューに入れて流量を制御することで、レート制限に引っかかるのを防ぎます：
import asyncio
import time
from collections import deque
class RequestQueue:
"""レート制限を考慮したリクエストキュー"""
def __init__(self, max_rpm: int = 50, max_tpm: int = 90000):
self.max_rpm = max_rpm
self.max_tpm = max_tpm
self.request_times: deque = deque()
self.token_usage: deque = deque()
self.lock = asyncio.Lock()
async def wait_if_needed(self, estimated_tokens: int):
"""レート制限に達しそうな場合は待機"""
async with self.lock:
now = time.time()
# 1分以上前のエントリを削除
while self.request_times and now - self.request_times[0] > 60:
self.request_times.popleft()
while self.token_usage and now - self.token_usage[0][0] > 60:
self.token_usage.popleft()
# RPM チェック
if len(self.request_times) >= self.max_rpm:
wait_time = 60 - (now - self.request_times[0])
if wait_time > 0:
print(f"RPM limit approaching. Waiting {wait_time:.1f}s")
await asyncio.sleep(wait_time)
# TPM チェック
current_tokens = sum(t[1] for t in self.token_usage)
if current_tokens + estimated_tokens > self.max_tpm:
wait_time = 60 - (now - self.token_usage[0][0])
if wait_time > 0:
print(f"TPM limit approaching. Waiting {wait_time:.1f}s")
await asyncio.sleep(wait_time)
self.request_times.append(time.time())
self.token_usage.append((time.time(), estimated_tokens))2. バッチ処理の活用
大量のリクエストを処理する場合、Message Batches API を使用するとレート制限の影響を軽減できます：
import anthropic
client = anthropic.Anthropic()
# バッチリクエストの作成
batch = client.messages.batches.create(
requests=[
{
"custom_id": f"request-{i}",
"params": {
"model": "claude-sonnet-4-6",
"max_tokens": 1024,
"messages": [
{"role": "user", "content": f"タスク {i}: {task}"}
]
}
}
for i, task in enumerate(tasks)
]
)
# バッチの状態を確認
result = client.messages.batches.retrieve(batch.id)
print(f"Status: {result.processing_status}")バッチ処理のメリット：
- レート制限が別枠で管理される
- コストが 50% 割引になる
- 大量リクエストを効率的に処理できる
- 24 時間以内に結果が返される
3. プロンプトキャッシングの活用
繰り返し使用するプロンプトにはキャッシングを活用して、トークンの再処理を削減できます：
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=[
{
"type": "text",
"text": "大量のコンテキスト情報...",
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{"role": "user", "content": "質問内容"}
]
)キャッシュヒット時は入力トークンのコストが 90% 削減されます。
上限に当たったら — Retry-After を無視しない
どれだけ流量を整えても、共有環境である以上、上限に当たる瞬間がゼロになることはありません。私自身、自動投稿のスクリプトが深夜に 429 を返したまま止まっているのを、翌朝になって気づいた失敗があります。そこで学んだのは、エラーを避けようとするより、429 を受け取った後の振る舞いを正しく設計するほうが堅いということでした。
Claude API は 429 を返すとき、Retry-After ヘッダで「何秒待てば良いか」を教えてくれることがあります。自前で待ち時間を決め打ちするより、まずこの値を尊重するのが筋です。値が無いときだけ、指数的に待ち時間を伸ばし、さらに少しのゆらぎ（ジッター）を足して、複数プロセスの再試行が同じ瞬間に重ならないようにします。
import time, random
import anthropic
from anthropic import RateLimitError
client = anthropic.Anthropic()
def create_resilient(messages, max_retries=6):
"""429 のときは Retry-After を尊重し、無ければジッター付きで待つ"""
backoff = 2.0
for attempt in range(max_retries):
try:
return client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=messages,
)
except RateLimitError as e:
if attempt == max_retries - 1:
raise
retry_after = None
resp = getattr(e, "response", None)
if resp is not None:
hdr = resp.headers.get("retry-after")
if hdr:
retry_after = float(hdr)
wait = retry_after if retry_after is not None else backoff
wait += random.uniform(0, 0.5) # ジッターで同時再試行を散らす
time.sleep(wait)
backoff *= 2
raise RuntimeError("max retries exceeded")ここで効くのは、待ち時間を「サーバーが教えてくれるなら従い、教えてくれないときだけ自分で決める」という優先順位です。固定値だけで組むと、混雑時には足りず、空いている時間には待ちすぎます。私はこの形に切り替えてから、夜間のバッチがエラーで止まることはほとんどなくなりました。
モニタリングと可視化
使用量のトラッキング
import anthropic
from datetime import datetime
class UsageTracker:
"""API 使用量のトラッキング"""
def __init__(self):
self.requests = []
def track(self, response):
"""レスポンスから使用量を記録"""
self.requests.append({
"timestamp": datetime.now().isoformat(),
"model": response.model,
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
"cache_read_tokens": getattr(
response.usage, "cache_read_input_tokens", 0
),
"cache_creation_tokens": getattr(
response.usage, "cache_creation_input_tokens", 0
),
})
def summary(self):
"""使用量のサマリーを出力"""
total_input = sum(r["input_tokens"] for r in self.requests)
total_output = sum(r["output_tokens"] for r in self.requests)
total_cache = sum(r["cache_read_tokens"] for r in self.requests)
print(f"Total requests: {len(self.requests)}")
print(f"Total input tokens: {total_input:,}")
print(f"Total output tokens: {total_output:,}")
print(f"Cache read tokens: {total_cache:,}")
# 使用例
tracker = UsageTracker()
response = client.messages.create(...)
tracker.track(response)
tracker.summary()スケーリング戦略
段階的なスケールアップ
本番環境へのデプロイでは、段階的にトラフィックを増やしていくアプローチが推奨されます：
- 開発段階: デフォルトのレート制限で十分
- テスト段階: 負荷テストで制限に達する場合はプランのアップグレードを検討
- 本番初期: 実トラフィックを監視しながら調整
- スケール段階: Anthropic に連絡してカスタムレート制限を相談
レート制限の引き上げリクエスト
デフォルトの制限では不十分な場合、Anthropic Console からレート制限の引き上げをリクエストできます。リクエスト時には以下の情報を準備してください：
- 現在の使用パターン（RPM、TPM の実績値）
- 必要な制限値とその根拠
- ユースケースの説明
- 予想されるトラフィックの成長率
まず手をつけるなら
レート制限は厄介な壁に見えますが、正体は「共有資源を公平に保つための交通整理」です。やみくもにプランを上げる前に、できることがあります。
最初の一歩としておすすめなのは、いま動いているコードに使用量のロギングを一つ足すことです。レスポンスの usage を記録するだけで、自分のボトルネックが入力トークンなのか、リクエストの頻度なのかが見えてきます。詰まっている軸が分かれば、プロンプトを削るのか、キューで流量を抑えるのか、バッチに寄せるのか、打つ手は自然と決まります。測ってから動く——遠回りに見えて、これが結局いちばん速い道だと感じています。