前回は Claude API の Tool Use(関数呼び出し)を使って、Claudeに外部ツールを持たせる実装方法を紹介しました。「Claudeがどのツールをいつ使うか判断してくれる」という部分が個人的にはかなり面白くて、ちょっとした自動化であれば割とすんなり書けるんですよね。
で、今回は Season2 の最終回ということで、「実際に本番で使うとき詰まるポイント」を中心にまとめます。エラー処理・コスト管理・よくある落とし穴あたり。ちゃんと動くコードを書いてから、「あ、これ本番では死ぬやつだ」と気づくことが自分は多いので、そのあたりを丁寧にやっていきます。
- 本番で踏みやすいエラーの種類と対処法
- リトライ処理の実装(指数バックオフ)
- トークン使用量の追跡とコスト計算
- プロンプトキャッシュでコストを削る
- 会話履歴の肥大化問題とその対処
本番でよく出るエラーの種類
Anthropic の Python SDK には、エラーの種類ごとにちゃんと例外クラスが用意されています。雑に except Exception で全部受けると、「なぜ失敗したか」がわからなくなるので、最低限これだけは把握しておくといいかなと。
- RateLimitError(429 のことが多いようです):リクエストが多すぎるとき。しばらく待てば回復する。一番よく踏む
- APIStatusError(5xx):Anthropic 側のサーバーエラー。こちらも待てば直ることが多い
- APITimeoutError:レスポンスが返ってくるまでに時間がかかりすぎたとき
- APIConnectionError:ネットワーク接続自体の問題
- AuthenticationError(401):APIキーが間違っているか期限切れ。これはリトライしても意味がない
- (4xx系のリクエスト不正):リクエストの形式が間違っている。こちらもリトライ不要(SDK的には
APIStatusErrorのサブクラスとして扱われ、status_codeとエラーメッセージを見て判定するのが安全です)
ポイントは「リトライしていいエラーかどうか」を区別すること。429 や 5xx はリトライ対象だけど、4xx(400・401 など)はコードかAPIキーを直さないと意味がないので、そのまま投げ直してはいけない。
指数バックオフ付きのリトライ実装
429 や一時的なサーバーエラーに対応するための基本的なリトライ処理です。exponential backoff(指数バックオフ)というやり方で、失敗するたびに待ち時間を倍にしていく。
import time
import anthropic
client = anthropic.Anthropic()
def call_with_retry(messages, system=None, model="claude-haiku-4-5", max_tokens=1024, max_retries=3):
last_exception = None
for attempt in range(max_retries):
try:
kwargs = {
"model": model,
"max_tokens": max_tokens,
"messages": messages,
}
if system:
kwargs["system"] = system
return client.messages.create(**kwargs)
except anthropic.RateLimitError as e:
last_exception = e
wait = 2 ** attempt # 1s → 2s → 4s
print(f"Rate limited. {wait}秒待ってリトライします({attempt + 1}/{max_retries})")
time.sleep(wait)
except anthropic.APIStatusError as e:
if e.status_code >= 500:
last_exception = e
wait = 2 ** attempt
print(f"サーバーエラー({e.status_code})。{wait}秒待ってリトライします")
time.sleep(wait)
else:
raise # 4xx はリトライしない
except (anthropic.APITimeoutError, anthropic.APIConnectionError) as e:
last_exception = e
wait = 2 ** attempt
print(f"接続エラー。{wait}秒待ってリトライします")
time.sleep(wait)
raise last_exception
ちなみに wait = 2 ** attempt だと 1, 2, 4 秒になります。本番ではここに少しランダム性(ジッター)を足すとサーバー負荷の集中を避けられるらしいですが、個人用途ならこのくらいで十分かなと思ってます。
あと max_tokens を必ず設定するのは大事で、これを省略すると出力がかなり長くなってしまう可能性があるので注意です。「念のため省略しない」を習慣にするだけで余計なコストがかなり防げます。
トークン使用量を追跡してコストを把握する
API レスポンスの usage フィールドには input_tokens と output_tokens が入っています。これを積み上げておくだけで、どのくらい使ったかが追えます。
class TokenTracker:
def __init__(self):
self.total_input = 0
self.total_output = 0
self.call_count = 0
def record(self, response):
self.total_input += response.usage.input_tokens
self.total_output += response.usage.output_tokens
self.call_count += 1
def estimate_cost(self, model="claude-haiku-4-5"):
# 2026年5月時点の料金(per 1M tokens)
pricing = {
"claude-haiku-4-5": {"input": 1.0, "output": 5.0},
"claude-sonnet-4-6": {"input": 3.0, "output": 15.0},
"claude-opus-4-7": {"input": 5.0, "output": 25.0},
}
if model not in pricing:
return None
p = pricing[model]
cost = (self.total_input / 1_000_000 * p["input"]
+ self.total_output / 1_000_000 * p["output"])
return cost
def summary(self, model="claude-haiku-4-5"):
cost = self.estimate_cost(model)
print(f"呼び出し回数: {self.call_count}")
print(f"入力トークン合計: {self.total_input:,}")
print(f"出力トークン合計: {self.total_output:,}")
if cost is not None:
print(f"推定コスト: ${cost:.4f}")
tracker = TokenTracker()
response = call_with_retry([{"role": "user", "content": "Pythonの良いところを3つ教えて"}])
tracker.record(response)
tracker.summary()
Haiku 4.5 が $1/$5、Sonnet 4.6 が $3/$15、Opus 4.7 が $5/$25(いずれも per 1M tokens の入力/出力)という感じです。ちょっとした処理なら Haiku で十分なことが多いので、モデルの使い分けだけでコストがかなり変わります。
プロンプトキャッシュで同じ内容を使い回す
同じシステムプロンプトを何度も送る場合、プロンプトキャッシュを使うと入力トークンのコストを大幅に削れます。キャッシュを有効にするには、「ここから先(system / messages / tools の一部)をキャッシュ対象にしたい」というコンテンツブロックに cache_control: {"type": "ephemeral"} を付けるイメージです。するとシステムが自動的にキャッシュの管理をしてくれます。
import anthropic
client = anthropic.Anthropic()
# 長いシステムプロンプトを毎回送るケース
LONG_SYSTEM_PROMPT = """
あなたはECサイトのカスタマーサポート担当です。
以下のルールに従って回答してください:
- 返品期限は購入から30日以内
- 送料無料の条件は3,000円以上
- 在庫確認は「在庫確認中」と答え、後ほど連絡する
- クレームには謝罪から入る
... (ここに数千トークン分の規約・FAQ・商品情報などが続く想定)
"""
def chat_with_cache(user_message, conversation_history):
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=512,
system=[
{
"type": "text",
"text": LONG_SYSTEM_PROMPT,
"cache_control": {"type": "ephemeral"}, # ここでキャッシュ指定
}
],
messages=conversation_history + [
{"role": "user", "content": user_message}
],
)
return response
プロンプトキャッシュを使うと、会話の2ターン目以降は入力トークンの多くをキャッシュから読み込めるようになります。ただし、キャッシュされるための最小トークン数(minimum cacheable prompt length)はモデルによって違うので注意が必要です。なのでシステムプロンプトが短い場合は効果がないことがあります。長い FAQ や規約、たくさんのツール定義を毎回送るようなケースで威力を発揮します。
余談ですが、最初にこのキャッシュを試したとき、usage.cache_read_input_tokens が返ってきたときのうれしさが地味にありました。「あ、ちゃんと使われてる」っていう確認ができるので、実装後はレスポンスの usage を眺めてみてください。
会話履歴が肥大化する問題
マルチターンの会話をそのまま全部 messages に積み上げていくと、いつかコンテキストウィンドウの上限に当たります。Claude のモデルは現在 200K トークンまで入るので「そうそう当たらないだろう」と思いがちですが、長い会話 + ツール結果が積み重なると意外と早く膨らみます。
対策は主に2つ:
① 古いメッセージを切り捨てる
def trim_history(messages, max_turns=10):
"""最新のN往復だけ残す(system は別管理なので含まない)"""
if len(messages) <= max_turns * 2:
return messages
return messages[-(max_turns * 2):]
シンプルだけど、古いコンテキストが完全に消えるので、会話の途中で「さっき話した件」を参照できなくなります。用途によってはこれで十分。
② 要約してから圧縮する
def summarize_and_compress(messages, keep_recent=4):
"""古い会話部分を要約して1件にまとめる"""
if len(messages) <= keep_recent * 2:
return messages
old_messages = messages[:-(keep_recent * 2)]
recent_messages = messages[-(keep_recent * 2):]
# 古い部分をClaudeに要約させる
summary_response = client.messages.create(
model="claude-haiku-4-5", # 要約はHaikuで十分
max_tokens=512,
messages=[
{
"role": "user",
"content": f"以下の会話履歴を200字以内で要約してください:\n\n{str(old_messages)}"
}
],
)
summary = summary_response.content[0].text
compressed = [
{"role": "user", "content": f"(これまでの会話の要約){summary}"},
{"role": "assistant", "content": "了解しました。続けましょう。"},
]
return compressed + recent_messages
要約に Haiku を使うのは「要約のためにOpusを使ってたらコスト本末転倒」という理由です。正直このあたりの実装はまだ改善の余地があると思っていて、もっとうまい方法がある気がする。
本番で踏んだ落とし穴メモ
最後に、実際に詰まったポイントをいくつか。ドキュメントを読んでいても「実際に動かすまでわからなかった」系のやつです。
Tool Use のループが終わらない
Tool Use を実装したとき、stop_reason == "tool_use" の間ループし続けるコードを書いたんですが、ループの終了条件をちゃんと設定しないと無限ループになります。最大ループ回数を設定しておくのが安全です。
MAX_TOOL_ITERATIONS = 10
for i in range(MAX_TOOL_ITERATIONS):
response = client.messages.create(...)
if response.stop_reason != "tool_use":
break
# ツール実行 → messages に追加
else:
# MAX_TOOL_ITERATIONS 回ループしても終わらなかった場合
raise RuntimeError("ツールループが上限に達しました")
messages の形式エラーは 400 で返ってくる
Tool Use のレスポンスを messages に追加するとき、フォーマットが少しでもズレると 4xx(たとえば 400)エラーになることがあります。特に tool_result のブロック構造は自分もよく間違えました。エラーメッセージをちゃんと読むと「どのフィールドが足りないか」が書いてあるので、焦らず確認するのが早いです。
レスポンスの中身は必ず型チェックする
# response.content は list のような形で返ってくるので、[0] だけ取ろうとすると落ちることがあります
for block in response.content:
if block.type == "text":
print(block.text)
elif block.type == "tool_use":
print(f"ツール呼び出し: {block.name}")
Tool Use が混在しているとき、response.content[0].text と決め打ちで取るとエラーになることがあります。複数のブロックが返ってくる前提で書くのが安全です。
※この記事にはプロモーションが含まれます
ちなみに、PLAUD NOTE(AIボイスレコーダー。録音から文字起こし・要約まで自動で行える)も気になっています。PLAUD NOTE![]()
おわりに
Season2 は全4回で、基本的な API 呼び出しから始まって、マルチターン会話・Tool Use・そして今回の本番運用ポイントという流れでした。正直「まとめ記事」みたいな位置づけではあるんですが、自分が実際に詰まった部分を中心に書いたので、同じところでハマってる人の参考になれば。

