Claude Sonnet 4.6 を本番で回しながら、Sonnet 5 への移行を下見する

import anthropic
 
client = anthropic.Anthropic()
 
 
def first_text(message) -> str:
    """応答から最初の text ブロックを返す。
 
    content[0] が text であるとは限りません。thinking を有効にした応答では
    thinking ブロックが先に並び、ツール利用の応答では tool_use が先に来ます。
    位置ではなく type で選ぶ、というだけの関数ですが、
    これを挟んでおくと Sonnet 5 へ差し替えたときに壊れません。
    """
    for block in message.content:
        if block.type == "text":
            return block.text
    return ""
 
 
# 複数ファイルを一括でコンテキストに含める
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 first_text(message)
 
# 期待する出力: アーキテクチャ分析レポート（問題点・改善提案・セキュリティリスク）

// 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トークンで実現可能）
    });
 
    // 位置ではなく type で選びます（thinking / tool_use が先頭に来ることがあります）
    const textBlock = response.content.find((b) => b.type === "text");
    const assistantMessage = textBlock ? textBlock.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 first_text(response)

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"])
# 期待する出力: 詳細な移行戦略、優先順位付きのサービス分割計画、リスク評価

def think_deeply_v5(problem: str, effort: str = "high") -> dict:
    """Sonnet 5 版。thinking の深さは budget ではなく effort で決めます。"""
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=16000,
        thinking={"type": "adaptive", "display": "summarized"},
        output_config={"effort": effort},  # high / 低くするほど思考は浅く速く
        messages=[{"role": "user", "content": problem}],
        # temperature / top_p / top_k は非既定値だと 400 になるため渡しません
    )
 
    out = {"thinking": None, "answer": None}
    for block in response.content:
        if block.type == "thinking":
            out["thinking"] = block.thinking
        elif block.type == "text":
            out["answer"] = block.text
    return out
 
# 期待する出力: think_deeply と同じ形の dict（thinking は要約テキスト）

import anthropic
 
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_20251124",
                "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": first_text(response)}
 
# 期待する出力: クリック座標 [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 first_text(response)
 
# 最初の呼び出し: キャッシュ作成（書き込み $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(first_text(result.result.message))
 
    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
 

// app/api/chat/route.ts
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
export async function POST(req: Request) {
  const { messages } = await req.json();
 
  const readable = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder();
      const stream = client.messages.stream({
        model: "claude-sonnet-4-6",
        max_tokens: 4096,
        messages,
      });
 
      // text ブロックの差分だけを拾います（thinking の差分はここへ流れません）
      stream.on("text", (delta) => controller.enqueue(encoder.encode(delta)));
      stream.on("error", (err) => controller.error(err));
 
      await stream.finalMessage();
      controller.close();
    },
  });
 
  return new Response(readable, {
    headers: { "Content-Type": "text/plain; charset=utf-8" },
  });
}

import anthropic
 
client = anthropic.Anthropic()
 
PRICES = {  # $/MTok（入力, 出力）
    "claude-sonnet-4-6": (3.0, 15.0),
    "claude-sonnet-5": (2.0, 10.0),
}
 
 
def compare_input_cost(system_prompt: str, user_message: str) -> None:
    """同じプロンプトを両モデルでトークンカウントし、入力側の実費を並べます。"""
    for model, (in_price, _out_price) in PRICES.items():
        counted = client.messages.count_tokens(
            model=model,
            system=system_prompt,
            messages=[{"role": "user", "content": user_message}],
        )
        tokens = counted.input_tokens
        cost = tokens * in_price / 1_000_000
        print(f"{model:22} {tokens:>8,} tokens  ${cost:.6f}")
 
 
# 期待する出力（例）:
# claude-sonnet-4-6         12,480 tokens  $0.037440
# claude-sonnet-5           16,120 tokens  $0.032240

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 first_text(response)
        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},
    ...
)