Claude API のモバイル統合で個人開発者が直面する設計判断 — コスト・UX・オフライン対応の実践パターン

# Python（サーバーサイド処理の例）
import anthropic
 
client = anthropic.Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")
 
def select_model_for_task(task_type: str) -> str:
    """
    タスクの種類に応じてモデルを選択する
    個人開発では、コスト最適化のためにモデルを適切に使い分けることが重要
    """
    model_map = {
        "classification": "claude-haiku-4-5-20251001",   # 分類・ラベリング
        "chat": "claude-sonnet-4-6",                     # ユーザーチャット
        "generation": "claude-sonnet-4-6",               # コンテンツ生成
        "analysis": "claude-opus-4-6",                   # 深い分析（月1回程度）
    }
    return model_map.get(task_type, "claude-haiku-4-5-20251001")  # デフォルトはHaiku
 
def process_user_request(user_input: str, task_type: str):
    model = select_model_for_task(task_type)
    
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": user_input}]
    )
    
    # どのモデルを使ったかをログに残す（コスト分析のため）
    print(f"Model used: {model}, Input tokens: {response.usage.input_tokens}, Output tokens: {response.usage.output_tokens}")
    
    return response.content[0].text

// iOS Swift - ストリーミングレスポンスの実装例
import Foundation
 
class ClaudeStreamingClient {
    private let apiKey = "YOUR_ANTHROPIC_API_KEY"
    private let baseURL = "https://api.anthropic.com/v1/messages"
    
    func streamResponse(userMessage: String, onChunk: @escaping (String) -> Void, onComplete: @escaping () -> Void) async {
        var request = URLRequest(url: URL(string: baseURL)!)
        request.httpMethod = "POST"
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")
        request.setValue(apiKey, forHTTPHeaderField: "x-api-key")
        request.setValue("2023-06-01", forHTTPHeaderField: "anthropic-version")
        
        let body: [String: Any] = [
            "model": "claude-sonnet-4-6",
            "max_tokens": 1024,
            "stream": true,
            "messages": [["role": "user", "content": userMessage]]
        ]
        request.httpBody = try? JSONSerialization.data(withJSONObject: body)
        
        // URLSessionのdelegateを使ってSSEストリームを受信
        let (asyncBytes, _) = try! await URLSession.shared.bytes(for: request)
        
        for try await line in asyncBytes.lines {
            guard line.hasPrefix("data: ") else { continue }
            let jsonString = String(line.dropFirst(6))
            guard jsonString != "[DONE]" else { break }
            
            if let data = jsonString.data(using: .utf8),
               let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
               let delta = json["delta"] as? [String: Any],
               let text = delta["text"] as? String {
                await MainActor.run {
                    onChunk(text)  // メインスレッドでUIを更新
                }
            }
        }
        await MainActor.run { onComplete() }
    }
}

// iOS Swift - スライディングウィンドウによる会話履歴管理
struct ConversationManager {
    private var history: [(role: String, content: String)] = []
    private let maxHistoryItems = 10   // 最大10往復分を保持
    private let maxTokensPerMessage = 500  // 1メッセージあたりの最大トークン数
    
    mutating func addMessage(role: String, content: String) {
        // コンテンツが長すぎる場合は切り詰める（安全策）
        let truncated = content.count > 2000 ? String(content.prefix(2000)) + "..." : content
        history.append((role: role, content: truncated))
        
        // 最大件数を超えた場合は古いものから削除（ただし最初のシステムメッセージは保持）
        if history.count > maxHistoryItems {
            history.removeFirst()
        }
    }
    
    func buildAPIMessages() -> [[String: String]] {
        return history.map { ["role": $0.role, "content": $0.content] }
    }
    
    // 現在の推定トークン数（目安）
    func estimatedTokenCount() -> Int {
        return history.reduce(0) { $0 + ($1.content.count / 4) }  // 大まかな推定
    }
}

// リクエストに適切なタイムアウトを設定する
var request = URLRequest(url: url)
request.timeoutInterval = 30  // 30秒でタイムアウト（デフォルトは60秒）

// 指数バックオフによるリトライ
func sendWithRetry(request: URLRequest, maxRetries: Int = 3) async throws -> Data {
    var lastError: Error?
    
    for attempt in 0..<maxRetries {
        do {
            let (data, response) = try await URLSession.shared.data(for: request)
            
            if let httpResponse = response as? HTTPURLResponse {
                // レート制限（429）の場合は待機してリトライ
                if httpResponse.statusCode == 429 {
                    let waitTime = Double(2 << attempt)  // 2, 4, 8秒と倍増
                    try await Task.sleep(nanoseconds: UInt64(waitTime * 1_000_000_000))
                    continue
                }
                // サーバーエラー（5xx）もリトライ対象
                if httpResponse.statusCode >= 500 {
                    lastError = APIError.serverError(httpResponse.statusCode)
                    try await Task.sleep(nanoseconds: UInt64(Double(2 << attempt) * 1_000_000_000))
                    continue
                }
            }
            return data
        } catch {
            lastError = error
            // ネットワークエラーはリトライ
            if attempt < maxRetries - 1 {
                try await Task.sleep(nanoseconds: UInt64(Double(2 << attempt) * 1_000_000_000))
            }
        }
    }
    throw lastError ?? APIError.maxRetriesExceeded
}
 
enum APIError: Error {
    case serverError(Int)
    case maxRetriesExceeded
}

# Python（サーバーサイド）- ユーザーごとのレート制限
import time
from collections import defaultdict
 
class UserRateLimiter:
    def __init__(self):
        # {user_id: [timestamp, timestamp, ...]}
        self.request_history = defaultdict(list)
        self.max_requests_per_hour = 20   # 1時間あたり20リクエスト
        self.max_requests_per_day = 100   # 1日あたり100リクエスト
    
    def is_allowed(self, user_id: str) -> tuple[bool, str]:
        now = time.time()
        history = self.request_history[user_id]
        
        # 古いリクエストを削除
        history = [ts for ts in history if now - ts < 86400]  # 24時間以内のみ保持
        
        # 日次チェック
        if len(history) >= self.max_requests_per_day:
            return False, "1日の利用上限に達しました"
        
        # 時間次チェック
        recent = [ts for ts in history if now - ts < 3600]  # 1時間以内
        if len(recent) >= self.max_requests_per_hour:
            return False, "1時間の利用上限に達しました。しばらくお待ちください"
        
        # 許可してリクエストを記録
        history.append(now)
        self.request_history[user_id] = history
        return True, ""
 
rate_limiter = UserRateLimiter()
 
def handle_user_request(user_id: str, user_message: str):
    allowed, message = rate_limiter.is_allowed(user_id)
    if not allowed:
        return {"error": message}
    
    # APIリクエストを実行
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    )
    return {"content": response.content[0].text}

# プロンプトキャッシングを使ったシステムプロンプト
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": """あなたは日本の観光スポットに詳しいガイドです。
ユーザーの質問に対して、具体的で役立つ情報を提供してください。
安全で楽しい旅行をサポートすることが目標です。""",
            "cache_control": {"type": "ephemeral"}  # キャッシュを有効化
        }
    ],
    messages=[{"role": "user", "content": user_message}]
)