壁紙アプリのストア説明文を、対応している言語ぶんまとめて点検するエージェントを組んでいた夜のことでした。言語ごとの説明文を取ってくるツール、直近のレビューを取るツール、配信中の壁紙のメタデータを引くツール——Claude はこれらを一つのターンで並べて呼び、応答は目に見えて速くなりました。
ところが翌朝のログを開くと、ある一言語の取得がタイムアウトした瞬間に、残りの言語の結果まで丸ごと捨てられておりました。速くなった分だけ、壊れ方も一度に来るのだと気づいたのは、その朝でした。
最初にお伝えしたいのは、Parallel Tool Use は「並列度を上げる機能」ではなく、失敗の組み合わせを増やす機能でもあるということです。制御・部分失敗の設計・観測の三つを、私が手を動かして辿った順に書き残します。
なぜ並列が効くのか — レイテンシの足し算と max
在庫確認・価格取得・配送料計算を三つの API で行うエージェントを考えます。直列なら 200ms + 180ms + 150ms = 530ms、並列なら max(200, 180, 150) = 200ms です。Claude 自身の生成時間を足しても、体感は半分近くになります。
公式ドキュメントでは、tool_choice が auto（既定）のとき、Claude は独立していると判断したツールを同じターンで複数呼ぶことがある、と説明されています。ただし「独立している」と判断するかどうかはプロンプトに左右されますし、判断が直列側へ倒れると、並列の土台を用意していても速くなりません。
並列が効く場面
- 複数の外部 API を同時に叩く検索・比較（商品検索、フライト比較、ニュース取得）
- 同じ処理を違う入力で一括実行するバッチ（複数 URL の取得、複数言語の説明文の点検）
- 独立した評価を並べる用途（複数の評価軸でのスコアリング、複数モデルへの同時問い合わせ）
並列が効きにくい・危ない場面
- 直前の結果を見て次のツールを決める連鎖（Search → Read → Analyze）
- DB 書き込みやファイル操作で、同じリソースに触る可能性があるもの
- 外部サービスのレート制限が厳しく、並列度をむしろ絞りたいもの
制御の軸は二つある — Claude の判断と、こちらの実行
Parallel Tool Use には「Claude が並列で tool_use ブロックを出すか」という軸と、「こちらのコードがそれを同時に実行するか」という軸があります。この二つを混ぜてしまうと、「並列にしたはずなのに速くならない」という調査に半日を使うことになるのです。
軸1: Claude の判断 — tool_choice と disable_parallel_tool_use
# 並列を明示的に禁止する例
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
tools=[
{"name": "get_inventory", "description": "在庫を取得", "input_schema": {"type": "object", "properties": {"sku": {"type": "string"}}}},
{"name": "get_price", "description": "価格を取得", "input_schema": {"type": "object", "properties": {"sku": {"type": "string"}}}},
],
tool_choice={
"type": "auto",
"disable_parallel_tool_use": True # 1ターンに1ツールだけ呼ぶ
},
messages=[{"role": "user", "content": "SKU-A123 の在庫と価格を教えて"}]
)disable_parallel_tool_use: True を指定すると、Claude は1ターンにつき1つの tool_use ブロックしか出しません。書き込みの順序を保証したいとき、副作用のあるツールを直列化したいときに使います。
既定の False では、Claude は独立していると判断したツールを同じターンで複数呼びます。プロンプトに「可能な限り並列に取得してから回答してください」と書くと、並列で呼ぶ割合が目に見えて上がります。tool_choice の四つのモードの使い分けは、Claude API の tool_choice で無駄な推論を減らす — 4モードの使い分けとコスト最適化パターンに書きました。
軸2: こちらの実行 — asyncio.gather と Semaphore
Claude が複数の tool_use ブロックを返しても、それをどう実行するかはこちら次第です。応答に tool_use が五つ並んでいても、for ループで一つずつ await していれば直列のままです。私も最初はここで止まっておりました。
import anthropic
import asyncio
import httpx
from typing import Any
client = anthropic.AsyncAnthropic()
# 各ツールの実装（async で書くのが前提）
async def get_inventory(sku: str) -> dict:
async with httpx.AsyncClient(timeout=5.0) as http:
r = await http.get(f"https://api.example.com/inventory/{sku}")
r.raise_for_status()
return r.json()
async def get_price(sku: str) -> dict:
async with httpx.AsyncClient(timeout=5.0) as http:
r = await http.get(f"https://api.example.com/price/{sku}")
r.raise_for_status()
return r.json()
async def get_shipping(sku: str, zip_code: str) -> dict:
async with httpx.AsyncClient(timeout=5.0) as http:
r = await http.get("https://api.example.com/shipping", params={"sku": sku, "zip": zip_code})
r.raise_for_status()
return r.json()
TOOL_MAP = {
"get_inventory": get_inventory,
"get_price": get_price,
"get_shipping": get_shipping,
}
# 並列度の上限（外部 API への同時接続を制限）
SEMAPHORE = asyncio.Semaphore(5)
async def run_tool(block: Any) -> dict:
"""1つの tool_use ブロックを実行して tool_result を返す"""
async with SEMAPHORE:
try:
fn = TOOL_MAP[block.name]
result = await fn(**block.input)
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": str(result),
"is_error": False,
}
except Exception as e:
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Error: {type(e).__name__}: {e}",
"is_error": True,
}
async def run_parallel_tools(response) -> list[dict]:
"""Claude の応答内のすべての tool_use ブロックを並列実行"""
tool_blocks = [b for b in response.content if b.type == "tool_use"]
results = await asyncio.gather(
*[run_tool(b) for b in tool_blocks],
return_exceptions=False # 例外は run_tool 内でキャッチ済み
)
return results押さえておきたい点は三つです。
一つめは、asyncio.gather で束ねることです。コルーチンを集めて一度に渡さない限り、await が並んでいても並列にはなりません。
二つめは、Semaphore で同時実行数に上限を置くことです。外部 API には必ずレート制限があり、Claude は思ったより多くのツールを一度に呼んできます。上限がないと、賑やかな一ターンが下流の API を落としてしまうのです。
三つめは、各ツールの例外を run_tool の内側で受け止め、is_error: True の tool_result に変えることです。gather(return_exceptions=False) のまま例外を外へ出すと、他のツールの結果まで失われます。冒頭の「一言語のタイムアウトで全言語が消えた」事故は、まさにこれでした。
部分失敗の設計 — いちばん事故が多いところ
五つのツールを並列で呼び、三つ成功・二つ失敗したとき、エージェントは何をすべきでしょうか。答えはプロダクトによって違い、私は三つの方針を使い分けております。
方針 A: 全部そろったときだけ答える（厳格）
決済金額や医療情報のように、欠けたデータで答えること自体が危険な場面です。一つでも失敗したら、ターン全体を失敗にしてやり直します。
class PartialFailureError(Exception):
def __init__(self, failed_tools: list[str]):
super().__init__(f"failed: {failed_tools}")
self.failed_tools = failed_tools
async def strict_multi_tool(messages, tools):
"""全ツール成功を必須とする厳格モード"""
response = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
tools=tools,
messages=messages,
)
if response.stop_reason != "tool_use":
return response
results = await run_parallel_tools(response)
if any(r["is_error"] for r in results):
raise PartialFailureError(
failed_tools=[r["tool_use_id"] for r in results if r["is_error"]]
)
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": results})
return await strict_multi_tool(messages, tools)方針 B: 失敗も含めて Claude に返し、判断を委ねる（柔軟）
検索や情報収集のように、部分的な結果でも答えられる場面です。失敗したツールの tool_result に is_error: True とエラー内容を入れて返すと、Claude は「この情報は取れなかったので、残りで回答します」と自分で判断してくれます。
async def lenient_multi_tool(messages, tools):
"""部分失敗を許容するモード — Claude に判断を委ねる"""
response = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
tools=tools,
messages=messages,
)
if response.stop_reason != "tool_use":
return response
results = await run_parallel_tools(response)
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": results})
return await lenient_multi_tool(messages, tools)ストア説明文の点検エージェントは、この方針 B に切り替えました。一言語が取れなかった朝は「英語と韓国語は確認できました。ドイツ語は取得できなかったので、後でもう一度お試しください」と返ってくるようになり、私は残りの言語の結果を捨てずに済むようになりました。
方針 C: 失敗したツールだけやり直す（折衷）
ネットワークの一時的な不調のように、やり直せば通る見込みが高いエラーに使います。
async def run_tool_with_retry(block, max_retries=3):
"""指数バックオフで個別リトライ"""
for attempt in range(max_retries):
async with SEMAPHORE:
try:
fn = TOOL_MAP[block.name]
result = await fn(**block.input)
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": str(result),
"is_error": False,
}
except (httpx.TimeoutException, httpx.ConnectError) as e:
# 一時的なエラーはリトライ
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt)
continue
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Error after {max_retries} retries: {e}",
"is_error": True,
}
except Exception as e:
# リトライしても無駄なエラーは即失敗
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Non-retryable error: {e}",
"is_error": True,
}やり直す例外の種類を分けておくのが要点です。401 はやり直しても通りませんし、400 はこちらの引数の不具合なので、直すのが先です。
冪等性 — 同じツールが二度走っても壊れないか
並列ツールでいちばん怖いのは、同じツールが同じ引数で二度実行されることです。Claude の計画の揺れ、こちらのリトライ、ターンの再送——読み取りなら無害ですが、書き込み系では取り返しがつきません。
async def create_order(user_id: str, sku: str, quantity: int, idempotency_key: str) -> dict:
"""注文作成 — 冪等性キーで重複防止"""
async with httpx.AsyncClient(timeout=10.0) as http:
r = await http.post(
"https://api.example.com/orders",
json={"user_id": user_id, "sku": sku, "quantity": quantity},
headers={"Idempotency-Key": idempotency_key}
)
r.raise_for_status()
return r.json()tool_use_id は Claude が各 tool_use ブロックに付ける一意の ID なので、これをそのまま冪等性キーにするのがいちばん簡単で安全です。ツールの引数として Claude に生成させると、リトライのたびに違う値になりかねません。
async def run_tool_idempotent(block) -> dict:
"""tool_use_id を冪等性キーに自動注入"""
if block.name == "create_order":
input_with_key = {**block.input, "idempotency_key": block.id}
result = await create_order(**input_with_key)
else:
fn = TOOL_MAP[block.name]
result = await fn(**block.input)
return {"type": "tool_result", "tool_use_id": block.id, "content": str(result)}ここで私が引いた線引きは一つだけです。並列に走らせるのは読み取りまで。書き込みは一本ずつ、鍵を付けて。 書き込みツールを含むエージェントでは disable_parallel_tool_use: True を選び、読み取り専用のエージェントだけ並列を許しております。
ストリーミングと並列ツール
ストリーミング（SSE）では、tool_use の引数は input_json_delta として断片で届き、content_block_stop が来た時点でそのブロックの JSON が確定します。ブロックは index 順に届くので順序が乱れることはありませんが、確定前の断片を実行にかけてはいけない、という点だけは守る必要があります。
import anthropic
import asyncio
client = anthropic.AsyncAnthropic()
async def stream_and_execute_parallel(messages, tools):
async with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=2048,
tools=tools,
messages=messages,
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True) # テキストは届いた順に UI へ
final = await stream.get_final_message()
tool_blocks = [b for b in final.content if b.type == "tool_use"]
if not tool_blocks:
return final
# ストリーム完了後にまとめて並列実行
results = await asyncio.gather(*[run_tool(b) for b in tool_blocks])
messages.append({"role": "assistant", "content": final.content})
messages.append({"role": "user", "content": results})
return await stream_and_execute_parallel(messages, tools)content_block_stop を拾って、確定したブロックから順に実行を始める書き方もできます。ただ、ターンの最後のブロックが確定するまでの差はせいぜい数百ミリ秒で、get_final_message() を待ってから束ねる方が、私の手元では不具合がはるかに少なく済みました。テキストは先に流し、ツールは確定後に束ねる——この分け方で十分なのだと思います。
観測 — 並列度を分布で見る
本番で並列ツールを運用するなら、ツールごとのレイテンシ・成功率・並列度を指標にしておきます。どこが遅いのか見えないと、最適化の判断ができません。
import time
from prometheus_client import Histogram, Counter
TOOL_LATENCY = Histogram(
"claude_tool_duration_seconds",
"Tool execution duration",
["tool_name", "status"]
)
TOOL_CALLS = Counter(
"claude_tool_calls_total",
"Total tool calls",
["tool_name", "status"]
)
PARALLEL_DEGREE = Histogram(
"claude_parallel_tool_degree",
"Number of tools called in parallel per turn",
buckets=[1, 2, 3, 5, 8, 13, 21]
)
async def run_tool_observed(block) -> dict:
start = time.monotonic()
status = "success"
try:
async with SEMAPHORE:
fn = TOOL_MAP[block.name]
result = await fn(**block.input)
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": str(result),
"is_error": False,
}
except Exception as e:
status = "error"
return {
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Error: {e}",
"is_error": True,
}
finally:
duration = time.monotonic() - start
TOOL_LATENCY.labels(tool_name=block.name, status=status).observe(duration)
TOOL_CALLS.labels(tool_name=block.name, status=status).inc()
async def run_parallel_tools_observed(response):
tool_blocks = [b for b in response.content if b.type == "tool_use"]
PARALLEL_DEGREE.observe(len(tool_blocks)) # 並列度を記録
return await asyncio.gather(*[run_tool_observed(b) for b in tool_blocks])並列度を分布で持っておくと、Claude が実際にどれくらい横に広げているかが見えます。プロンプトを一行変えたときに分布がどう動くかを眺めながら調整するのが、いちばん確かな進め方でした。
Prometheus を入れる前でも、会話ログさえ残っていれば並列度は数えられます。保存済みの messages から、assistant ターンごとの tool_use ブロック数を集計する小さな関数です。
from collections import Counter
def parallel_degree_histogram(messages: list[dict]) -> Counter:
"""保存済み messages から、assistant ターンごとの tool_use 数を数える"""
hist: Counter = Counter()
for m in messages:
if m["role"] != "assistant" or not isinstance(m["content"], list):
continue
n = sum(1 for b in m["content"] if getattr(b, "type", None) == "tool_use" or (isinstance(b, dict) and b.get("type") == "tool_use"))
if n:
hist[n] += 1
return hist
# 例: Counter({1: 41, 2: 12, 3: 7, 5: 1}) → 平均が 1 に近ければ並列化の余地が大きいExtended Thinking との組み合わせで気をつけたいこと
Extended Thinking を有効にしたエージェントで並列度を測ると、私の手元では、思考なしのときより一ターンあたりの tool_use 数が少なめに出ることがありました。一つ結果を見てから次を決める、という進め方に寄るのだと思います。
これは深い推論が要る仕事では望ましい挙動ですが、「独立したデータをまとめて取って組み立てる」だけの仕事では、並列の速さが出ません。並列ツールを評価するときは、Thinking の有無を分けて並列度を測っておくと、「並列が効かない」の原因を取り違えずに済みます。
Semaphore の値をどう決めるか
asyncio.Semaphore(n) の n は、私は次の手順で決めております。
まず n = 3 から始めます。ほとんどの下流 API はこれで乱れません。現実的な負荷を 30 分ほど流し、自分側の p95 レイテンシと下流のエラー率を見ます。両方が健全なら n を 1 上げ、同じことを繰り返します。
止めどきは三つです。下流が 429 や 503 を無視できない頻度で返し始めたとき、自分側の p95 が平らになったとき（並列度を上げても速くならない＝ボトルネックに達している）、そしてメモリや CPU が苦しくなったとき。私の場合は 4〜8 のあいだに落ち着くことがほとんどで、10 を超えるときは下流と喧嘩している合図でした。
一ターン内の並列度はせいぜい 2〜5 なので、Semaphore が本当に効くのは、複数のセッションが同時に走っている時間帯です。
並列をあえて切る場面
- 同じレコードに書き込みが集中するワークフロー: 直列化するだけで、ロックを作り込まずに書き込み競合を防げます
- 依存関係の強い推論: 各ツールの結果が次の引数になるなら、並列は元から成立しません。先に Claude へ伝えた方がトークンの無駄が減ります
- 順序付きの監査ログが要る領域: 並列は順序の保証を難しくします
- 下流が毎秒 1 リクエストしか許さないとき: 並列で出しても、ネットワークで直列に戻るだけです
並列を切る代償は、ターン数が増えることです。三つのツールを一ターンで呼ぶ代わりに、一ツールずつ三ターンになり、Claude の生成時間がそのぶん積み上がります。上の場面では、それでも釣り合う取引だと感じております。
並列の動作をテストで確かめる
並列ツールのテストは、ロジックではなくタイミングから不具合が出るので、直列より少し工夫が要ります。私が頼りにしているのは、遅延を指定できる偽のツールです。「並列のはず」なら、合計時間は最も遅いツールの遅延に近づき、足し算にはならないはずです。
import asyncio
import time
from types import SimpleNamespace
def make_fake_tool(delay: float, result: dict):
async def fake(**kwargs):
await asyncio.sleep(delay)
return result
return fake
async def test_tools_actually_run_in_parallel():
TOOL_MAP["get_inventory"] = make_fake_tool(0.2, {"stock": 10})
TOOL_MAP["get_price"] = make_fake_tool(0.2, {"price": 99})
response = SimpleNamespace(content=[
SimpleNamespace(type="tool_use", id="toolu_01", name="get_inventory", input={"sku": "A"}),
SimpleNamespace(type="tool_use", id="toolu_02", name="get_price", input={"sku": "A"}),
])
start = time.monotonic()
results = await run_parallel_tools(response)
elapsed = time.monotonic() - start
assert len(results) == 2
assert elapsed < 0.3, f"直列になっています: {elapsed:.2f}s" # 並列なら約 0.2s
asyncio.run(test_tools_actually_run_in_parallel())この上に、ランダムに一つのツールを落とす故障注入を重ねると、厳格・柔軟・折衷の方針が意図どおりに動くかまで確かめられます。
私が本番で踏んだ落とし穴
直列のまま並列だと思い込んでおりました。 for の中で await していただけでした。プロファイラで見るまで気づけませんでした。
tool_result の対応を順序に頼っておりました。 Claude API は tool_result の順序が tool_use と一致することを要求しませんが、同じ tool_use_id の組が揃っていることは要求します。gather は入力順で返しますが、ログや UI では tool_use_id で照合する設計にしておく方が安全です。tool_result の欠落で起きる送信エラーの回復は、Claude API の tool_result could not be submitted が再発しないための回復ハンドラ設計に分けて書いております。
Semaphore を入れずに走らせておりました。 Claude が八つのツールを一度に呼び、下流のレート制限に触れ、エージェントごと止まりました。
冪等性キーをクライアント側で毎回生成しておりました。 uuid.uuid4() を呼ぶたびに新しいキーになり、リトライの瞬間に同じ注文が二度通ります。tool_use_id か、リクエストの論理的な一意性から決定的に作ります。
並列度をプロンプトで誘導する
Claude がどれだけ横に広げるかは、システムプロンプトの言葉に左右されます。
効果があった書き方は、「回答に必要な情報は、可能な限り 同時に 取得してください。相互に依存しないツールは並列で呼び出してください」のように、「同時に」「並列で」を明示するものでした。「効率的にツールを使ってください」では、ほとんど動きませんでした。私の手元の記録では、この一行で平均の並列度が 1.3 前後から 2.8 前後へ動いております——ツールの構成で値は変わりますので、ご自身の分布で確かめていただければと思います。
本番投入前の確認
asyncio.gatherで実際に並列実行されているか（偽ツールのテストかプロファイラで確認）- Semaphore の上限が下流のレート制限に収まっているか
- 各ツール関数の内側で例外を受け止め、
is_error: Trueに変えているか - 書き込み系ツールに冪等性キーがあり、
tool_use_idを注入しているか - 部分失敗の方針（厳格・柔軟・折衷）がプロダクトに合っているか
- 並列度・レイテンシ・成功率の指標が取れているか
- プロンプトに「並列で取得してほしい」旨が書いてあるか
次の一歩
まず、いまのエージェントの会話ログから、assistant ターンあたりの tool_use 数を数えてみてください。平均が 1.0 に近ければ、並列化の余地はまだ大きく残っています。 そこからプロンプトに一行足し、Semaphore・冪等性・観測を順に重ねていけば、本番に耐える並列エージェントになります。
読み取りは並列に、書き込みは一本ずつ——この線引きだけは、急ぎの夜でも崩さないようにしております。