Anthropic リファレンスアーキテクチャ

概要

Claude API統合のための3つの検証済みアーキテクチャパターン：同期APIゲートウェイ、非同期キューベース処理、マルチモデルルーティング。

アーキテクチャ1: 同期APIゲートウェイ(シンプル)

User → API Gateway → Claude Service → Messages API
                                     ↓
                                   Response → User

# 最適な用途: チャットボット、インタラクティブツール、低ボリューム(<100 RPM)
from fastapi import FastAPI
import anthropic

app = FastAPI()
client = anthropic.Anthropic(max_retries=3, timeout=60.0)

@app.post("/chat")
async def chat(prompt: str):
    msg = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return {"text": msg.content[0].text, "tokens": msg.usage.output_tokens}

アーキテクチャ2: 非同期キューベース(スケーラブル)

User → API → Queue (Redis/SQS) → Worker Pool → Messages API
  ↑                                                ↓
  └──────────── Status/Result ←── Result Store ←───┘

# 最適な用途: バッチ処理、高ボリューム、バックグラウンドタスク
from redis import Redis
from rq import Queue
import anthropic

redis = Redis()
task_queue = Queue("claude-tasks", connection=redis)
result_store = Redis(db=1)

def process_task(task_id: str, prompt: str, model: str):
    client = anthropic.Anthropic()
    msg = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    result_store.setex(f"result:{task_id}", 3600, msg.content[0].text)

# エンキュー
import uuid
task_id = str(uuid.uuid4())
task_queue.enqueue(process_task, task_id, prompt, "claude-sonnet-4-20250514")

アーキテクチャ3: マルチモデルルーター

User → Router → Haiku    (分類/抽出)
              → Sonnet   (一般/コード)
              → Opus     (研究/複雑)
              → Batches  (一括/オフライン)

class ModelRouter:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.classifier = anthropic.Anthropic()  # 同じクライアントでも可

    def route_and_execute(self, prompt: str, context: dict) -> str:
        # ステップ1: Haikuで分類(低コスト、高速)
        classification = self.classifier.messages.create(
            model="claude-haiku-4-20250514",
            max_tokens=32,
            messages=[{
                "role": "user",
                "content": f"Classify this request as: simple|moderate|complex|bulk\n\n{prompt[:200]}"
            }]
        )
        complexity = classification.content[0].text.strip().lower()

        # ステップ2: 適切なモデルにルーティング
        model_map = {
            "simple": "claude-haiku-4-20250514",
            "moderate": "claude-sonnet-4-20250514",
            "complex": "claude-opus-4-20250514",
        }
        model = model_map.get(complexity, "claude-sonnet-4-20250514")

        # ステップ3: 選択したモデルで実行
        msg = self.client.messages.create(
            model=model,
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}]
        )
        return msg.content[0].text

プロジェクトレイアウト

my-claude-app/
├── src/
│   ├── main.py              # FastAPIアプリ
│   ├── claude/
│   │   ├── client.py         # シングルトン + 設定
│   │   ├── router.py         # モデルルーティングロジック
│   │   ├── tools.py          # ツール定義
│   │   └── prompts/          # システムプロンプト(ファイル形式)
│   ├── workers/
│   │   └── claude_worker.py  # キューコンシューマー
│   └── middleware/
│       ├── rate_limiter.py   # アプリレベルのレート制限
│       └── cost_tracker.py   # 支出監視
├── tests/
│   ├── unit/                 # モック済みテスト
│   └── integration/          # ライブAPIテスト
└── config/
    ├── .env.development
    ├── .env.staging
    └── .env.production

エラーハンドリング

アーキテクチャ	障害モード	対策
同期ゲートウェイ	429/5xxがユーザーをブロック	サーキットブレーカー + フォールバック応答
キューベース	ワーカークラッシュ	デッドレターキュー + リトライポリシー
マルチモデル	ルーターの誤分類	Sonnetをデフォルト(最安全な中間値)

リソース

• API概要
• ツール使用
• プロンプトキャッシング

次のステップ

マルチ環境セットアップについては、anth-multi-env-setupを参照してください。