Claude Sonnet 4.6 — 1Mトークン・Computer Use・Extended Thinking を本番で使いこなす実装パターン

import anthropic
 
client = anthropic.Anthropic()
 
# 複数ファイルを一括でコンテキストに含める
def analyze_codebase(file_paths: list[str]) -> str:
    files_content = ""
    for path in file_paths:
        with open(path, "r") as f:
            files_content += f"\n\n=== {path} ===\n{f.read()}"
 
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=8192,
        messages=[
            {
                "role": "user",
                "content": f"""以下のコードベース全体を分析してください。
アーキテクチャの問題点・改善点・セキュリティリスクを日本語で報告してください。
 
{files_content}
"""
            }
        ]
    )
    return message.content[0].text
 
# 期待する出力: アーキテクチャ分析レポート（問題点・改善提案・セキュリティリスク）

// TypeScript: 長期セッション管理
interface Message {
  role: "user" | "assistant";
  content: string;
}
 
class LongSessionChat {
  private history: Message[] = [];
  private client: Anthropic;
 
  constructor() {
    this.client = new Anthropic();
  }
 
  async chat(userMessage: string): Promise<string> {
    this.history.push({ role: "user", content: userMessage });
 
    const response = await this.client.messages.create({
      model: "claude-sonnet-4-6",
      max_tokens: 4096,
      system: "あなたは専任のコーディングアシスタントです。ユーザーの開発プロジェクトを長期的にサポートしてください。",
      messages: this.history, // 全履歴を送信（1Mトークンで実現可能）
    });
 
    const assistantMessage = response.content[0].text;
    this.history.push({ role: "assistant", content: assistantMessage });
    return assistantMessage;
  }
 
  // トークン使用量の確認
  getHistoryTokenEstimate(): number {
    const totalChars = this.history.reduce((sum, m) => sum + m.content.length, 0);
    return Math.floor(totalChars / 4); // 大まかな推定
  }
}

# 複数の PDF レポートから構造化データを一括抽出
def extract_from_documents(documents: list[str]) -> dict:
    combined = "\n\n".join([f"=== 文書{i+1} ===\n{doc}" for i, doc in enumerate(documents)])
 
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=16384,
        messages=[{
            "role": "user",
            "content": f"""以下の複数のビジネスレポートから、JSON形式で構造化データを抽出してください。
各文書の: 売上高、利益率、主要KPI、リスク要因を抽出してください。
 
{combined}
 
出力形式:
{{
  "documents": [
    {{
      "revenue": "...",
      "profit_margin": "...",
      "kpis": [...],
      "risks": [...]
    }}
  ]
}}"""
        }]
    )
    return response.content[0].text

import anthropic
 
client = anthropic.Anthropic()
 
def think_deeply(problem: str, budget_tokens: int = 10000) -> dict:
    """
    Extended Thinking を使って複雑な問題を解決する。
 
    budget_tokens: 思考プロセスに使えるトークン上限（1024〜32000）
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=16000,  # thinking_tokens + 回答トークンの合計
        thinking={
            "type": "enabled",
            "budget_tokens": budget_tokens,  # 思考に使うトークン予算
        },
        messages=[{
            "role": "user",
            "content": problem
        }]
    )
 
    result = {"thinking": None, "answer": None}
 
    for block in response.content:
        if block.type == "thinking":
            result["thinking"] = block.thinking  # 思考プロセス（参照用）
        elif block.type == "text":
            result["answer"] = block.text  # 実際の回答
 
    return result
 
# 使用例: アーキテクチャ設計の深い分析
design_problem = """
マイクロサービスアーキテクチャへの移行を検討しています。
現状: Python Flask モノリス、PostgreSQL 1台、日次ユーザー10万人
目標: 99.99% SLA、水平スケーリング対応、チーム別デプロイ
最適な移行戦略と、最初に切り出すべきサービスを判断してください。
"""
 
result = think_deeply(design_problem, budget_tokens=15000)
print("=== 思考プロセス（内部） ===")
print(result["thinking"][:500], "...")  # 思考の一部を確認
print("\n=== 回答 ===")
print(result["answer"])
# 期待する出力: 詳細な移行戦略、優先順位付きのサービス分割計画、リスク評価

import anthropic
import base64
from PIL import Image
import io
 
client = anthropic.Anthropic()
 
def run_computer_task(task_description: str, screenshot_b64: str) -> dict:
    """
    スクリーンショットを元に Computer Use でタスクを実行する。
 
    返り値: 実行するアクション（クリック・入力・スクロール等）
    """
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        tools=[
            {
                "type": "computer_20250124",
                "name": "computer",
                "display_width_px": 1920,
                "display_height_px": 1080,
            }
        ],
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": screenshot_b64,
                        },
                    },
                    {
                        "type": "text",
                        "text": f"現在の画面を見て、次のタスクを実行してください: {task_description}"
                    }
                ],
            }
        ],
    )
 
    actions = []
    for block in response.content:
        if block.type == "tool_use" and block.name == "computer":
            actions.append({
                "action": block.input.get("action"),
                "coordinate": block.input.get("coordinate"),
                "text": block.input.get("text"),
            })
 
    return {"actions": actions, "reasoning": response.content[0].text if response.content[0].type == "text" else ""}
 
# 期待する出力: クリック座標 [960, 540]、入力テキスト「example@email.com」等のアクション指示

def safe_computer_action(action: dict, risk_level: str) -> bool:
    """高リスクアクションの場合は確認を求める"""
    HIGH_RISK_ACTIONS = ["delete", "submit", "purchase", "send_email"]
 
    if risk_level == "high" or any(h in str(action).lower() for h in HIGH_RISK_ACTIONS):
        # 人間の確認を求める（本番では Slack 通知等と連携）
        print(f"⚠️ 高リスクアクション検出: {action}")
        confirm = input("実行しますか? (yes/no): ")
        return confirm.lower() == "yes"
    return True

import anthropic
 
client = anthropic.Anthropic()
 
# 長いシステムプロンプトをキャッシュ
SYSTEM_PROMPT = """
あなたは○○株式会社の専属AIアシスタントです。
[...ここに長い会社固有の情報・ルール・ナレッジを記述...]
総文字数が多いほどキャッシュの恩恵が大きくなります。
"""
 
def cached_completion(user_message: str) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},  # キャッシュを有効化
            }
        ],
        messages=[{"role": "user", "content": user_message}],
    )
 
    # キャッシュヒット率を確認
    usage = response.usage
    print(f"入力トークン: {usage.input_tokens}")
    print(f"キャッシュ作成: {getattr(usage, 'cache_creation_input_tokens', 0)}")
    print(f"キャッシュ読み取り: {getattr(usage, 'cache_read_input_tokens', 0)}")
 
    return response.content[0].text
 
# 最初の呼び出し: キャッシュ作成（書き込み $3.75/1M tokens）
first = cached_completion("今日の作業予定を教えてください")
 
# 2回目以降: キャッシュヒット（読み取り $0.30/1M tokens → 90%オフ！）
second = cached_completion("先ほどの件について詳しく教えてください")

import anthropic
import time
 
client = anthropic.Anthropic()
 
def batch_process(tasks: list[dict]) -> list[str]:
    """
    複数のタスクを非同期バッチ処理する。
    コスト: 通常の 50% OFF（24時間以内に完了）
    """
    requests = []
    for i, task in enumerate(tasks):
        requests.append({
            "custom_id": f"task-{i}",
            "params": {
                "model": "claude-sonnet-4-6",
                "max_tokens": 8192,
                "messages": [{"role": "user", "content": task["prompt"]}]
            }
        })
 
    # バッチ作成
    batch = client.messages.batches.create(requests=requests)
    print(f"バッチID: {batch.id}、処理件数: {len(requests)}件")
 
    # 完了を待機
    while True:
        batch = client.messages.batches.retrieve(batch.id)
        if batch.processing_status == "ended":
            break
        print(f"処理中... ({batch.request_counts.processing} 件残り)")
        time.sleep(30)
 
    # 結果取得
    results = []
    for result in client.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            results.append(result.result.message.content[0].text)
 
    return results
 
# 使用例: 100件のブログ記事タイトルから SEO説明文を一括生成
tasks = [{"prompt": f"「{title}」の SEO 用 meta description を160文字以内で生成"} for title in title_list]
descriptions = batch_process(tasks)

import anthropic
 
client = anthropic.Anthropic()
 
def stream_response(prompt: str):
    """
    ストリーミングで Sonnet 4.6 のレスポンスを取得する。
    最初のトークンが到達した時点でユーザーに表示を開始できる。
    """
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        full_text = ""
        for text in stream.text_stream:
            print(text, end="", flush=True)  # リアルタイム表示
            full_text += text
 
        # ストリーム完了後の使用量確認
        final_message = stream.get_final_message()
        print(f"\n\n--- 使用量 ---")
        print(f"入力: {final_message.usage.input_tokens} tokens")
        print(f"出力: {final_message.usage.output_tokens} tokens")
 
        return full_text
 
# TypeScript (Next.js API Route) でのストリーミング実装
"""
// app/api/chat/route.ts
import Anthropic from '@anthropic-ai/sdk';
import { AnthropicStream, StreamingTextResponse } from 'ai';
 
const client = new Anthropic();
 
export async function POST(req: Request) {
  const { messages } = await req.json();
 
  const stream = await client.messages.create({
    model: 'claude-sonnet-4-6',
    max_tokens: 4096,
    stream: true,
    messages,
  });
 
  // Vercel AI SDK を使ったストリーミングレスポンス
  return new StreamingTextResponse(AnthropicStream(stream));
}
"""

def truncate_history(history: list, max_tokens: int = 900000) -> list:
    """トークン数が上限に近づいたら古い会話を削除する"""
    total = sum(len(m["content"]) // 4 for m in history)
    while total > max_tokens and len(history) > 2:
        # 最も古いユーザーとアシスタントのターンを削除
        history.pop(0)
        history.pop(0)
        total = sum(len(m["content"]) // 4 for m in history)
    return history

import time
import random
 
def call_with_retry(prompt: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4096,
                messages=[{"role": "user", "content": prompt}]
            )
            return response.content[0].text
        except anthropic.APIStatusError as e:
            if e.status_code == 529 and attempt < max_retries - 1:
                wait = (2 ** attempt) + random.uniform(0, 1)
                print(f"過負荷エラー。{wait:.1f}秒後にリトライ ({attempt+1}/{max_retries})")
                time.sleep(wait)
            else:
                raise

# NG: max_tokens が budget_tokens より小さい
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=5000,      # ← 不足
    thinking={"type": "enabled", "budget_tokens": 10000},  # budget が max を超えている
    ...
)
 
# OK: max_tokens = budget_tokens + 出力分を確保
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=16000,    # budget(10000) + 出力(6000)
    thinking={"type": "enabled", "budget_tokens": 10000},
    ...
)