「AI APIを触ってみたが、サンプルコードを動かした先が分からない」——これは実務でAIを組み込もうとする方から最もよく聞く声です。公式ドキュメントのHello Worldは動く。しかし業務で必要なのは、決まった形式のJSONを確実に返させること、コストを予算内に収めること、そして深夜のバッチが429エラーで止まらないことです。この記事では、Claude APIとOpenAI APIの構造の違いから、スクレイピング結果の構造化、コスト削減、本番運用のエラーハンドリングまで、実際に動くPythonコードで解説します。

Claude APIとOpenAI APIは「どこが違う」のか

両者はどちらもチャット形式のAPIですが、リクエストの組み立て方に明確な違いがあります。ここを理解しておくと、片方のコードをもう片方へ移植する作業が一気に楽になります。

まずClaude APIの最小構成です。Python SDKは pip install anthropic で導入し、APIキーは環境変数 ANTHROPIC_API_KEY から自動で読み込まれます。

from anthropic import Anthropic

client = Anthropic()

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    system="あなたは日本語の要約アシスタントです。事実のみを書いてください。",
    messages=[
        {"role": "user", "content": "次の議事録を3行で要約して:\n..."}
    ],
)

print(resp.content[0].text)
print(resp.usage.input_tokens, resp.usage.output_tokens)

対してOpenAI APIでは、システムプロンプトを messages 配列の中にロールとして含め、レスポンスは choices 経由で取り出します。

from openai import OpenAI

client = OpenAI()

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "あなたは日本語の要約アシスタントです。"},
        {"role": "user", "content": "次の議事録を3行で要約して:\n..."},
    ],
)

print(resp.choices[0].message.content)

実装上、押さえるべき差分は3つです。1つ目はsystemの位置。Claudeではトップレベルの独立パラメータで、messages配列にsystemロールを入れることはできません。2つ目はmax_tokensが必須であること。Claudeではこれを省略できず、この値は「出力の上限」を意味します。ここを小さくしたまま長文生成を指示すると、途中で切れた文章が返ってきます。3つ目はレスポンスがブロックの配列である点です。resp.content はテキストブロックやツール使用ブロックが並んだリストなので、常に [0].text で取れると決め打ちすると、後述のTool Useを導入した瞬間に壊れます。

そして最も見落とされがちなのが stop_reason です。"end_turn" なら正常終了、"max_tokens" なら出力が途中で打ち切られたことを意味します。バッチ処理では必ずこの値を検査し、切れた結果をそのままDBへ保存しないようにします。

if resp.stop_reason == "max_tokens":
    raise RuntimeError("出力がmax_tokensで打ち切られました")

実務で一番効く実装：Tool Useで「必ず正しいJSON」を受け取る

スクレイピングや文書処理の案件でAIを使う最大の目的は、非構造なテキストを構造化データに変換することです。ここで多くの実装が「JSON形式で出力してください」とプロンプトで頼み、返ってきた文字列を json.loads() に通し、たまに前後に付く「はい、以下がJSONです」という一文でクラッシュします。

これはプロンプトの工夫で粘る問題ではありません。Tool Use（関数呼び出し）機能を使い、スキーマをAPIレベルで強制するのが正解です。

import json
from anthropic import Anthropic

client = Anthropic()

TOOLS = [{
    "name": "save_product",
    "description": "商品ページから抽出した情報を保存する",
    "input_schema": {
        "type": "object",
        "properties": {
            "name":     {"type": "string", "description": "商品名"},
            "price":    {"type": "integer", "description": "税込価格（円）。記載がなければ -1"},
            "in_stock": {"type": "boolean", "description": "在庫があれば true"},
            "tags":     {"type": "array", "items": {"type": "string"}},
        },
        "required": ["name", "price", "in_stock", "tags"],
    },
}]

def extract(page_text: str) -> dict:
    resp = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1024,
        tools=TOOLS,
        tool_choice={"type": "tool", "name": "save_product"},  # 必ずこのツールを使わせる
        messages=[{"role": "user", "content": page_text}],
    )
    for block in resp.content:
        if block.type == "tool_use":
            return block.input
    raise ValueError("構造化データが返りませんでした")

ポイントは tool_choice に {"type": "tool", "name": ...} を指定している箇所です。これによりモデルは「そのツールを呼ぶ」以外の選択肢を失い、返却値は必ずJSON Schemaに沿ったdictになります。パース処理も正規表現によるJSON抽出も不要です。

さらに実務的なコツが2つあります。1つはdescriptionを日本語で丁寧に書くこと。プロパティのdescriptionは事実上のプロンプトとして機能するため、「記載がなければ -1」のようなフォールバック規則をここに書くと、欠損値の扱いが安定します。もう1つは「不明」を表現できる型にすること。必須フィールドしかないスキーマは、モデルに推測を強要し、ハルシネーションの温床になります。

なお、スクレイピングと組み合わせる場合、HTML全体を丸ごと投げるのはコストの無駄です。BeautifulSoupで本文領域だけを抜き、タグを落としたテキストにしてから渡すと、入力トークンが数分の一になり精度も上がります。「構造が安定している部分は従来のパーサ、崩れる部分だけをAI」——このハイブリッド設計が、サイト改修に強く運用費も安い構成です。

コストとレイテンシを削る3つの実装テクニック

AI APIの費用は「入力トークン×単価＋出力トークン×単価」で決まります。単価はモデルごとに大きく違い、出力側は入力側より高く設定されているのが一般的です。ここを設計で削ります。

(1) プロンプトキャッシュ。抽出ルールや分類定義など、リクエストのたびに繰り返し送る長い固定文がある場合、その部分をキャッシュ対象としてマークできます。

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": LONG_RULES,          # 数千トークンの抽出ルール
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[{"role": "user", "content": page_text}],
)
print(resp.usage)  # cache_creation_input_tokens / cache_read_input_tokens を確認

キャッシュへの書き込みは通常入力よりやや割高になる一方、ヒットした読み取りは大幅に安くなります。数百件を連続処理するバッチでは、この一行の有無で請求額が桁違いになります。ただしキャッシュには最小トークン数の条件と有効期間（既定は短時間）があるため、短いプロンプトに付けても効果はありません。実際の単価と条件は変動するので、必ず公式ドキュメントで最新値を確認してください。

(2) モデルの使い分け。全処理を最上位モデルで回す必要はまずありません。定型的な分類・抽出・整形は高速で安価なモデル（Haiku系）、複雑な判断や長文の要約は中位モデル（Sonnet系）、設計そのものを任せるような難所だけ最上位（Opus系）——という三段構成にすると、品質を保ったままコストが劇的に下がります。model を設定ファイルの変数にしておき、案件ごとに差し替えられる形にしておくのが実装の作法です。

(3) 非同期・並列とバッチ。即時性が不要な夜間バッチなら、Batch APIを使うことで大幅な割引を受けられます（結果は非同期で受け取ります）。一方、数百件をリアルタイムに処理したい場合は、非同期クライアントで並列度を制御しながら投げます。

import asyncio
from anthropic import AsyncAnthropic

client = AsyncAnthropic()
sem = asyncio.Semaphore(5)   # レート制限に合わせて同時実行数を絞る

async def one(text: str) -> str:
    async with sem:
        r = await client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": text}],
        )
        return r.content[0].text

async def main(texts):
    return await asyncio.gather(*(one(t) for t in texts))

セマフォで同時実行数を絞るのが要点です。無制限に gather すると即座にレート制限に到達し、結局リトライで遅くなります。

本番で落ちないためのエラーハンドリングとストリーミング

AI APIは外部サービスであり、失敗するのが前提です。特に長時間のバッチでは、429（レート制限）、529（過負荷）、一時的な5xxが必ず発生します。

まず知っておくべきは、公式SDKが既にリトライ機構を内蔵していることです。自前で書く前に、まずこれを設定します。

client = Anthropic(max_retries=5, timeout=60.0)

その上で、業務ロジック側では「リトライすべきエラー」と「即座に諦めるべきエラー」を分けます。認証エラーやリクエスト不正は何度投げても直りません。

import time, random
import anthropic

def call_with_retry(**kwargs):
    for attempt in range(5):
        try:
            return client.messages.create(**kwargs)
        except (anthropic.RateLimitError, anthropic.APIConnectionError):
            wait = min(2 ** attempt, 30) + random.random()   # 指数バックオフ＋ジッター
            time.sleep(wait)
        except anthropic.APIStatusError as e:
            if e.status_code >= 500:
                time.sleep(min(2 ** attempt, 30))
            else:
                raise            # 400番台は設定ミス。即座に落とす
    raise RuntimeError("リトライ上限に達しました")

ジッター（ランダムな揺らぎ）を加えるのは、並列プロセスが同じタイミングで一斉に再送し合う事態を避けるためです。加えて、バッチ処理では処理済みIDを永続化し、途中で落ちても再実行時に既処理分へ再課金されない設計にしておきます。これを怠ると、1000件目で落ちたバッチの再実行で丸ごと二重に費用が発生します。

最後にUX面。チャットUIや管理画面から呼ぶ場合、生成完了まで無言で待たせると体感品質が大きく下がります。ストリーミングは数行で導入できます。

with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=2048,
    messages=[{"role": "user", "content": "レポートを作成して"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()   # usage や stop_reason はここで取得

APIキーの扱いにも触れておきます。キーは必ず環境変数か秘密管理サービスから読み、リポジトリにコミットしない。フロントエンドから直接APIを叩かない（キーが露出します）。この2点は、AI機能を持つ受託案件で最初に確認すべきセキュリティ要件です。

📚 この記事に関連するおすすめ書籍

さらに深く学びたい方に、参考になる書籍をご紹介します。

※ 本サイトはAmazonアソシエイト・プログラムの参加者です。紹介リンクを経由してご購入いただいた場合、手数料を受け取ることがあります。

まとめ

AI APIの実装で成果を分けるのは、モデルの選択そのものより「周辺の設計」です。第一に、Tool Useとスキーマ強制によって出力を構造として受け取ること。文字列をパースして祈る実装から卒業するだけで、システムの安定性は別物になります。第二に、プロンプトキャッシュとモデルの三段使い分けでコストを設計に織り込むこと。処理単価が読めれば、業務への導入判断がしやすくなります。第三に、指数バックオフと処理済み管理で「落ちても壊れない」バッチにすること。そして、構造が安定した部分は従来のパーサに任せ、崩れる部分だけをAIに委ねるハイブリッド設計が、実運用では最も費用対効果に優れます。まずは既存業務の中で「人が目視で転記している箇所」を1つ選び、Tool Useによる構造化から試してみてください。

AIを活用した業務自動化を導入したい方はnashiまでお問い合わせください。https://nashi-portfolio.netlify.app