API利用料金が増加し続けて困っていませんか？大規模言語モデルを活用したサービスを提供していると、API呼び出しのコストが急速に膨らむ課題に直面します。特に同じプロンプトやコンテキストを何度も送信する場合、無駄な費用が発生していることに気づく人も多いでしょう。
Anthropic社が提供するClaudeのプロンプトキャッシュ機能は、この問題を劇的に解決します。適切に実装すれば、API利用コストを最大90%削減することも可能です。本記事では、Pythonを使ったプロンプトキャッシュの実装方法を、初心者向けにステップバイステップで解説します。
Claude APIのプロンプトキャッシュとは何か
プロンプトキャッシュは、Claude APIが頻繁に使用されるプロンプトやコンテキストをサーバー側で一時保存し、以降の呼び出しで再利用する機能です。キャッシュされたコンテンツの処理にかかるトークン数が大幅に削減されるため、結果的にAPI料金が安くなります。
具体的には、入力トークン1000個分のキャッシュを初回作成する際には通常料金が発生しますが、2回目以降は同じコンテンツに対してわずか10%の料金で処理されます。頻繁にアクセスされるドキュメント、FAQセクション、システムプロンプトなどの用途に特に効果的です。
プロンプトキャッシュが活躍する実践的なシーン
以下のような場面でプロンプトキャッシュの効果が大きく発揮されます。
- 社内ドキュメント、マニュアル、ナレッジベースの質問応答システム
- 複数ユーザーが同じ企業情報やAPI仕様書を使って質問する場合
- 長文の契約書、利用規約、プライバシーポリシーの分析と要約
- コードレビュー用に大規模なソースコードを何度も処理する場合
- 画像分析で同じ参考画像セットを複数の質問で活用する場合
例えば、100人のユーザーが同じ50ページのドキュメントについて質問する場合、プロンプトキャッシュなしでは毎回全文を処理する必要があります。キャッシュを活用すれば、初回のみ全文処理を行い、以降99回はキャッシュ料金で対応でき、総コストが大幅に削減されます。
Claude APIの料金体系の理解
プロンプトキャッシュの効果を最大化するには、Claude APIの料金構造を理解することが重要です。2024年時点のClause 3.5 Sonnetの料金は以下の通りです。
| 項目 | 料金（1M トークンあたり） |
|---|---|
| キャッシュ作成時の入力 | $3.00 |
| 標準入力トークン | $3.00 |
| キャッシュヒット時の入力 | $0.30 |
| 出力トークン | $15.00 |
注目すべきは、キャッシュヒット時は通常の10%の料金で処理されるという点です。つまり、同じコンテンツに10回以上アクセスされれば、キャッシュ作成の初期コストをすぐに回収できます。
Pythonでのプロンプトキャッシュ実装準備
実装に必要な環境を整えましょう。まずはPythonパッケージをインストールします。
pip install anthropic python-dotenv次に、Claude APIのキーを環境変数に設定します。
export ANTHROPIC_API_KEY="your-api-key-here"または、.envファイルを作成して以下のように記述します。
ANTHROPIC_API_KEY=your-api-key-here基本的なプロンプトキャッシュの実装
最初に、シンプルなテキストコンテンツを使ったプロンプトキャッシュの実装例をみていきます。
from anthropic import Anthropic
import os
from dotenv import load_dotenv
load_dotenv()
client = Anthropic()
# キャッシュを使用した基本的な実装
def cached_document_qa(document: str, question: str) -> str:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[
{
"type": "text",
"text": "あなたは専門的なドキュメント分析アシスタントです。提供されたドキュメントについて、正確かつ詳細に回答してください。"
},
{
"type": "text",
"text": document,
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{"role": "user", "content": question}
]
)
return response.content[0].text
# 使用例
sample_document = """
Python プログラミング言語について
Pythonは1991年にGuido van Rossumによって開発された高水準プログラミング言語です。シンプルで読みやすい文法が特徴であり、初心者から上級者まで幅広い層に利用されています。
主な特徴：
1. 可読性が高い - インデント構造で視認性が優れている
2. 豊富なライブラリ - 数千のオープンソースライブラリが利用可能
3. 多用途 - Web開発、データ分析、機械学習、自動化など幅広い分野で活用
4. クロスプラットフォーム - Windows、Mac、Linuxで実行可能
5. 動的型付け - 柔軟な開発が可能
Pythonは科学計算やデータ分析の分野で特に人気があり、NumPy、Pandas、ScikitLearnなどの強力なライブラリが存在します。
"""
# 最初の質問（キャッシュ作成）
answer1 = cached_document_qa(sample_document, "Pythonの主な特徴を3つ教えてください")
print("質問1の回答:")
print(answer1)
# 2番目の質問（キャッシュヒット）
answer2 = cached_document_qa(sample_document, "PythonはどのようなOSで実行可能ですか？")
print("\n質問2の回答:")
print(answer2)このコードのポイントは、cache_control: {"type": "ephemeral"}の指定です。ephemeralキャッシュは、APIリクエスト間でキャッシュを保持します（通常は5分間有効）。
複数のドキュメントを管理するスケーラブルな実装
実務ではより複雑な要件に対応する必要があります。複数のドキュメント、ユーザー質問を効率的に管理するコードを紹介します。
from anthropic import Anthropic
import json
from datetime import datetime
import os
from dotenv import load_dotenv
load_dotenv()
class DocumentQASystem:
def __init__(self):
self.client = Anthropic()
self.cache_stats = {
"total_requests": 0,
"cache_hits": 0,
"total_input_tokens": 0,
"cache_input_tokens": 0,
"cost_saved": 0.0
}
def load_documents(self, doc_paths: list) -> dict:
"""複数のドキュメントを読み込む"""
documents = {}
for path in doc_paths:
try:
with open(path, 'r', encoding='utf-8') as f:
doc_name = os.path.basename(path)
documents[doc_name] = f.read()
except FileNotFoundError:
print(f"警告: {path} が見つかりません")
return documents
def query_document(self, doc_name: str, document_content: str, question: str) -> dict:
"""ドキュメントに対して質問を実行"""
self.cache_stats["total_requests"] += 1
response = self.client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[
{
"type": "text",
"text": f"あなたは『{doc_name}』の専門家です。提供されたドキュメント内容に基づいて、ユーザーの質問に正確に回答してください。"
},
{
"type": "text",
"text": document_content,
"cache_control": {"type": "ephemeral"}
}
],
messages=[
{"role": "user", "content": question}
]
)
# キャッシュ使用状況の追跡
usage = response.usage
self.cache_stats["total_input_tokens"] += usage.input_tokens
if hasattr(usage, 'cache_read_input_tokens') and usage.cache_read_input_tokens > 0:
self.cache_stats["cache_hits"] += 1
self.cache_stats["cache_input_tokens"] += usage.cache_read_input_tokens
# キャッシュ料金は通常の10%
cost_saved = (usage.cache_read_input_tokens / 1_000_000) * 3.00 * 0.9
self.cache_stats["cost_saved"] += cost_saved
return {
"document": doc_name,
"question": question,
"answer": response.content[0].text,
"input_tokens": usage.input_tokens,
"output_tokens": usage.output_tokens,
"cache_read_tokens": getattr(usage, 'cache_read_input_tokens', 0)
}
def print_statistics(self):
"""キャッシュ使用統計を表示"""
print("\n========== キャ