1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@yureki_lab

Claude API を Python の asyncio で並列に呼ぶ実装手順(AsyncAnthropic) — gather で1件の失敗が全件の結果を道連れ・Semaphore を関数内で作って同時実行数が絞れない・クライアントを毎回作って接続を使い捨てる、3つのハマりどころ【2026】

1
Posted at

はじめに / 対象と前提

数百件のテキストを Claude API で分類・要約したいとき、1件ずつ for で回すと遅すぎる。かといって asyncio.gather に全部投げると、1件のエラーで全部の結果が消えたり、同時実行数が思ったように絞れなかったりする。この記事は、自分が実際に詰まった3点を、手元で動かした出力付きでまとめたもの。

  • 想定読者:Python で Claude API を叩いたことがあり、async / await は一応書ける人
  • 前提環境
    • Python 3.14.3(asyncio.TaskGroup を使う部分は 3.11 以上が必要)
    • anthropic Python SDK 1.14.0
    • macOS(Linux でも同じ挙動)

Batch API(結果が非同期で返ってきて半額になる仕組み)を使えるなら、そちらの方が安い。この記事は「数十〜数百件を、今この場で数分以内に処理したい」ケースが対象。

TL;DR

  • 並列化は AsyncAnthropic を1つだけ作って使い回し、asyncio.Semaphore を1つだけ作って全タスクで共有する
  • asyncio.gather は return_exceptions=True を付ける。付けないと1件の失敗で成功分の結果まで受け取れなくなる
  • 「1件でも失敗したら残りを止めたい」なら gather ではなく asyncio.TaskGroup を使う

手順 / 動かし方

1. 最小構成

import asyncio
import anthropic

MODEL = "claude-sonnet-5-5"
MAX_CONCURRENCY = 5  # 同時に飛ばすリクエスト数の上限

async def classify(client: anthropic.AsyncAnthropic,
                   sem: asyncio.Semaphore,
                   text: str) -> str:
    async with sem:  # ここで同時実行数を絞る
        msg = await client.messages.create(
            model=MODEL,
            max_tokens=64,
            messages=[{
                "role": "user",
                "content": f"次の文の感情を positive/negative/neutral の1語で答えて:\n{text}",
            }],
        )
        return msg.content[0].text.strip()

async def main(texts: list[str]) -> list[str | BaseException]:
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    async with anthropic.AsyncAnthropic(max_retries=4) as client:
        return await asyncio.gather(
            *(classify(client, sem, t) for t in texts),
            return_exceptions=True,
        )

if __name__ == "__main__":
    texts = ["最高の一日だった", "電車が止まって最悪", "明日は火曜日"]
    for t, r in zip(texts, asyncio.run(main(texts))):
        print(t, "->", r)

ポイントは3つだけ。

  • AsyncAnthropic は async with で1回だけ作る
  • Semaphore は main で1回だけ作り、引数で全タスクに渡す
  • gather に return_exceptions=True を付け、結果リストに例外が混ざる前提で後処理する

2. 失敗分だけ拾い直す

return_exceptions=True にすると、戻り値は「成功した文字列」と「例外オブジェクト」が入力と同じ順番で並んだリストになる。順番は入力どおりに保証されるので、zip で元データと突き合わせられる。

results = asyncio.run(main(texts))
failed = [t for t, r in zip(texts, results) if isinstance(r, BaseException)]
ok = {t: r for t, r in zip(texts, results) if not isinstance(r, BaseException)}
print(f"成功 {len(ok)} 件 / 失敗 {len(failed)} 件")
# failed だけ時間をおいて再実行する

SDK 自体も 429 や 5xx を自動でリトライする(既定は2回、上の例では max_retries=4 に増やしている)。それでも落ちたものだけを、ここで拾い直す。

ハマりどころ

ここからは、API キーなしで再現できるように asyncio.sleep で API 呼び出しを模した関数で挙動を確かめた。2件目だけ必ず失敗するようにしてある。

async def fake_call(i):
    await asyncio.sleep(0.2)
    if i == 2:
        raise RuntimeError(f"overloaded_error (req {i})")
    return f"ok {i}"

ハマり1:gather で1件の失敗が全件の結果を道連れにする

症状:100件中1件だけ 529(overloaded)で落ちたら、残り99件分の結果も1つも手に入らない。

try:
    r = await asyncio.gather(*(fake_call(i) for i in range(5)))
except Exception as e:
    print("gather raised:", e)

実行結果:

gather raised: overloaded_error (req 2)

r は代入されないので、成功した4件の結果も受け取れない。さらに厄介なのは、gather は最初の例外を投げた時点で他のタスクをキャンセルしないこと(公式ドキュメントにも明記されている)。つまり残りのリクエストは裏で走り続けてトークンを消費するのに、その結果はどこにも返ってこない。お金だけ払って結果を捨てている状態になる。

回避策:return_exceptions=True を付ける。

return_exceptions: ['ok 0', 'ok 1', RuntimeError('overloaded_error (req 2)'), 'ok 3', 'ok 4']

逆に「1件でも失敗したら残りは即中止したい」(例:API キーが間違っていて全部 401 になるのが分かっている)場合は、Python 3.11 以降の asyncio.TaskGroup を使う。こちらは1件が例外を投げた時点で残りのタスクをキャンセルし、例外を ExceptionGroup にまとめて投げてくれる。

try:
    async with asyncio.TaskGroup() as tg:
        for i in range(5):
            tg.create_task(fake_call(i))
except* RuntimeError as eg:
    print("TaskGroup:", [str(e) for e in eg.exceptions])

TaskGroup: ['overloaded_error (req 2)']

自分は「全件処理して失敗だけ後で拾う」なら gather(return_exceptions=True)、「前提が崩れたら即停止」なら TaskGroup、と使い分けている。

ハマり2:Semaphore を関数の中で作って同時実行数が絞れない

症状:Semaphore(2) で2並列に絞ったつもりが、全件が一斉に飛んで 429(rate_limit_error)が大量に出る。

原因はこう書いていたこと。

async def bad_call(i):
    sem = asyncio.Semaphore(2)  # 呼ばれるたびに新しい Semaphore を作っている
    async with sem:
        await call_api(i)

Semaphore は「同じ1個を全員で取り合う」から上限が効く。タスクごとに別々の Semaphore を作ると、全員が自分専用の枠を持っているのと同じで、何の制限にもならない。

同時実行中の数を数えて、ピーク値を比べてみた(10タスク・各0.1秒)。

bad peak 10
good peak 2 0.51

関数内で作った版はピークが10、つまり全部同時。外で1つ作って渡した版はピーク2で、所要時間も 0.1秒 × 10件 ÷ 2並列 = 約0.5秒と計算どおりになった。

回避策:Semaphore は main() など イベントループの中で1回だけ 作り、引数で渡す。Python 3.10 以降はモジュールのトップレベルで作っても動くが、テストで asyncio.run を複数回呼ぶ構成だと見通しが悪くなるので、自分は main() 内で作る派。

なお、並列数をいくつにするかは使っているプランのレート制限(1分あたりのリクエスト数・トークン数)次第。いきなり大きくせず、5 前後から始めて 429 の出方を見て調整するのが無難。

ハマり3:クライアントを毎回作って接続を使い捨てる

症状:並列化したのに思ったほど速くならない。長時間回すと ResourceWarning やファイルディスクリプタ不足のエラーが出ることがある。

async def classify(text):
    client = anthropic.AsyncAnthropic()  # 呼ぶたびに新しいクライアント
    return await client.messages.create(...)

AsyncAnthropic はインスタンスごとに自分の接続プール(使い終わった HTTPS 接続を取っておいて再利用する仕組み)を持っている。毎回作ると、毎回 TLS ハンドシェイクからやり直しになり、しかも明示的に閉じていないので、古い接続はガベージコレクションに回収されるまで残る。

SDK 1.14.0 の既定値を確認すると、1クライアントで最大1000接続・keep-alive 100本まで持てる設定になっている。

import anthropic._constants as c
print(c.DEFAULT_CONNECTION_LIMITS, c.DEFAULT_MAX_RETRIES, c.DEFAULT_TIMEOUT)

Limits(max_connections=1000, max_keepalive_connections=100, keepalive_expiry=5.0) 2 Timeout(connect=5.0, read=600, write=600, pool=600)

(_constants は SDK 内部のモジュールなので、確認用に覗くだけにしておく)

数百件程度なら クライアント1つで十分。async with で使えば抜けるときに確実に閉じる。

async with anthropic.AsyncAnthropic(api_key="x") as c2:
    print("ctx ok", c2.is_closed())
print("after", c2.is_closed())

ctx ok False
after True

ついでに、上の出力から 既定のタイムアウトが読み取り600秒 なのも分かる。短い分類タスクで1件が詰まって10分待たされるのは困るので、自分は用途に合わせて短くしている。

client = anthropic.AsyncAnthropic(timeout=60.0, max_retries=4)

背景・補足:SDK のリトライと自前リトライを二重にしない

SDK は 429・5xx・接続エラーで自動リトライする(初回 0.5秒待ち・最大 8秒の指数バックオフ。いずれも上の定数で確認できる)。ここに自前の while True リトライを重ねると、1件あたり「SDK のリトライ × 自前のリトライ」回まで膨らみ、Semaphore の枠を長時間占有して全体が詰まる。

自分の落としどころは次のとおり。

  • 一時的なエラーは SDK の max_retries に任せる
  • それでも落ちたものは return_exceptions=True で結果に残し、全体が終わってからまとめて再実行
  • 401 / 400 のように再実行しても直らないエラーは、anthropic.APIStatusError の status_code を見て再実行対象から外す
def is_retryable(e: BaseException) -> bool:
    if isinstance(e, anthropic.APIConnectionError):
        return True
    if isinstance(e, anthropic.APIStatusError):
        return e.status_code == 429 or e.status_code >= 500
    return False

retry_targets = [t for t, r in zip(texts, results)
                 if isinstance(r, BaseException) and is_retryable(r)]

まとめ

  • AsyncAnthropic は1つだけ作って async with で閉じる。毎回作ると接続を使い捨てる
  • Semaphore は1つだけ作って全タスクで共有する。関数内で作ると制限が効かない(実測でピーク10 vs 2)
  • gather は return_exceptions=True 必須。付けないと成功分の結果も失い、残りのリクエストは裏で走り続ける
  • 「1件失敗で全停止」なら TaskGroup、「全件処理して後で拾う」なら gather と使い分ける
  • リトライは SDK に任せ、自前では「全体が終わった後の再実行」だけにする
1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?