このノウハウで解決する課題

「AI の出力を待つ時間が気になる」「100 件のが 1 時間以上かかる」「Web UI と比べて自社が遅く感じる」。AI 体感速度の課題です。

結論

ストリーミングと並列実行を実装すれば、待ち時間は 5 分の 1〜10 分の 1 に短縮可能。Web UI は両方既定で有効なので意識しないが、 経由・社内ツール開発では明示的に書く必要があります。

必要な準備

  • ツール: API / API、Python（asyncio） or Node.js
  • 想定環境: 社内ツール開発 / バッチスクリプト
  • 前提知識: API 経由の AI 呼び出し基礎、async/await の理解

手順

1. ストリーミング表示の実装

通常の同期呼び出しは「全文生成完了を待つ → 表示」で、長文ほど待ち時間が長く感じる。 ストリーミングは「生成と同時に 1 ずつ表示」 で、最初の 0.5 秒で文字が出始めます。

Python（Anthropic SDK）

import anthropic

client = anthropic.Anthropic()

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    messages=[{"role": "user", "content": "月次レポートを生成"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

Node.js（fetch + EventSource）

Web フロント側ではサーバーから Server-Sent Events（SSE）で受信し、リアルタイム表示。

2. 並列実行の実装（重要）

12 件の処理を順次 1 件ずつやると、1 件 5 秒なら 60 秒。 並列にすれば 約 5〜6 秒（API 側の同時接続上限による）。

Python（asyncio）

（例：取引先別の月次サマリ生成の場合。自組織の業務内容に合わせて書き換えてください）

import anthropic
import asyncio

client = anthropic.AsyncAnthropic()

async def process_one(prompt: str) -> str:
    msg = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

async def process_all(prompts: list[str]) -> list[str]:
    tasks = [process_one(p) for p in prompts]
    return await asyncio.gather(*tasks)

# 12 件を並列実行
prompts = ["取引先 ABC の月次サマリ", "取引先 DEF の月次サマリ", ...]
results = asyncio.run(process_all(prompts))

3. 同時接続上限と Rate Limit に注意

並列度を上げすぎると 429（Rate Limit） エラー。Anthropic / OpenAI 共に Tier 制度（月利用額に応じて上限が上がる）あり。

実用的な並列度の目安:

  • Tier 1: 5〜10 並列まで
  • Tier 2 以降: 20〜50 並列も可能
# Semaphore で並列度を制御
sem = asyncio.Semaphore(10)
async def process_with_limit(prompt):
    async with sem:
        return await process_one(prompt)

4. バッチ API の活用（最強の最適化）

リアルタイム性が不要な業務は Batch API を使うと:

  • 料金が 50% 引き
  • 24 時間以内に完了保証

夜間に大量処理を回す用途に最適:

batch = client.messages.batches.create(
    requests=[
        {"custom_id": f"job-{i}", "params": {...}} for i in range(1000)
    ]
)
# 数時間後に結果取得

5. ストリーミング + 並列の組合せ

並列実行の各タスクをストリーミングで進捗表示すれば、ユーザーは「どこまで終わったか」をリアルタイムで把握可能。Web UI で複数タスクを並列実行する社内ツールに最適。

効果と限界

効果:

  • ストリーミング: 体感待ち時間が 0 秒に近づく
  • 並列実行: 処理時間 5〜10 分の 1
  • バッチ API: 料金 50% 削減 + 並列度の制約なし

限界:

  • Rate Limit に注意（Tier に応じて並列度を調整）
  • ストリーミングは エラーハンドリングが複雑化（途中で切れた時の扱い）
  • バッチ API は 24 時間 あり、リアルタイム業務には不向き

応用・派生

  • 複数の並列呼び出し: / / に同時に投げて品質比較
  • Hierarchical Tasks: 大タスクを Opus が分解 → サブタスクを Haiku で並列実行 → Opus で統合