はじめに：この記事が解決すること

Pythonの非同期処理に関する多くのチュートリアルは、async def、await、Task、Futureをいきなり紹介します。しかし、それらが互いにどのような関係にあるのかを先に説明しないことも少なくありません。コードをまねることはできても、次の疑問に答えられない読者は多いでしょう。

  • async defで定義した関数を呼び出しても、なぜすぐ実行されないのか？
  • awaitの後ろには何を置けるのか？
  • await time.sleep(5)はなぜ非同期にならず、最後にはエラーになるのか？
  • await fetch_a()の待機中に、イベントループはなぜ次の行のfetch_b()を直接実行できないのか？
  • コルーチン、Task、Futureの違いは何か？
  • async forとasync withは、それぞれどのプロトコルに支えられているのか？
  • 非同期プログラムにもスレッドプールが必要なのはなぜか？
  • asyncio.to_thread()でCPUバウンドな計算を解決できるのか？
  • FastAPI、LLMのストリーミング、AI Agentのツール呼び出しでは、これらをどのように使うべきか？

この記事では同期イテレーションから始め、完全なメンタルモデルを段階的に構築したうえで、コルーチン、タスクスケジューリング、実務上の設計へ進みます。

まず覚えておきたい五つのこと

今すぐ全文を読む時間がない場合は、まず次の五つを覚えてください。

  1. 通常の関数は呼び出すとすぐ実行されます。コルーチン関数は呼び出してもコルーチンオブジェクトが作られるだけで、自動的には実行されません。
  2. イベントループがスケジュールするのは、すでに存在するTaskとコールバックです。現在のコルーチンでまだ実行されていない次の行へ飛ぶことはありません。
  3. awaitで待機できるのはawaitableだけです。代表的なawaitableは、コルーチンオブジェクト、Task、Futureです。
  4. 直接awaitを続けて書くと通常は逐次実行になります。複数のコルーチンを並行実行するには、先にすべてをTaskとしてスケジュールするか、TaskGroupまたはgather()へ渡す必要があります。
  5. async defを書いても、同期ブロッキングコードが自動的にノンブロッキングになるわけではありません。

第1部：正しい並行処理のメンタルモデル

1. レストランの厨房というたとえ

料理人が一人しかいないレストランを想像してください。

料理人が同じ瞬間に本当にできることは一つだけです。

食材を切る
鍋をかき混ぜる
盛り付ける

しかし、スープを火にかけて沸騰を待っている間、料理人は立ち止まる必要がありません。別の料理を進め、コンロから合図が出たら戻ればよいのです。

これはasyncioの典型的な動作に対応します。

一つのイベントループスレッド
+
一時停止と再開が可能な複数のTask
+
I/O完了通知

ここで実現されるのは 並行性（concurrency） です。

複数の仕事が同じ時間帯に交互に進行すること。

必ずしも 並列性（parallelism） ではありません。

複数の仕事が、複数のCPUコアによって同じ瞬間に本当に実行されること。

2. 料理人のたとえが通用しない部分

このたとえは便利ですが、二つの重要な修正が必要です。

イベントループはソースコードを読む賢い料理人ではない

イベントループは次のコードを見て、

await fetch_a()
await fetch_b()

自分で次のように判断するわけではありません。

「Aが待っているので、次の行へ飛んでBを実行しよう。」

イベントループが実行できるのは、すでに作成・スケジュールされたTask、登録済みのコールバック、実行可能になったI/Oイベントだけです。

fetch_b()がまだ呼ばれていないなら、

  • fetch_bのコルーチンオブジェクトは存在せず、
  • 対応するTaskも存在せず、
  • イベントループはBの存在すら知りません。

協調的スケジューリングはプリエンプティブスケジューリングではない

asyncioは主に 協調的スケジューリング を採用します。あるTaskが実行中でも、イベントループが任意の時点で強制的に処理を奪うことはありません。通常、Taskが次のいずれかに到達する必要があります。

  • まだ完了していないawaitに到達する。
  • 自発的に実行権を手放す。
  • 実行を終える。

その時点で、イベントループは別のTaskを実行できます。

Python公式ドキュメントでは、イベントループは一つのスレッドですべてのコールバックとTaskを実行し、実行中のTaskがawaitに到達して一時停止すると、次のTaskを実行できると説明されています。1

3. 混同しやすい四組の概念

並行性と並列性

並行性：
複数のタスクが交互に進行する

並列性：
複数のタスクが複数の実行単位で同時に動く

asyncioが主に扱うのはI/O並行性です。CPU並列性には通常、複数プロセス、複数インタープリター、またはGILを解放するネイティブ拡張が必要です。

ブロッキングとノンブロッキング

ブロッキング：
処理が完了するまで呼び出し側がその場で停止する

ノンブロッキング：
処理が完了していなくても呼び出し側が別の仕事を進められる

time.sleep(5)は現在のスレッドをブロックします。await asyncio.sleep(5)は現在のTaskを一時停止し、イベントループに別のTaskを実行させます。

同期と非同期

同期インターフェース：
通常、現在の呼び出しが戻る時点で結果を得る

非同期インターフェース：
まず将来の結果を表すオブジェクトを受け取り、
後でその完了を待つ

I/OバウンドとCPUバウンド

I/Oバウンド：
ネットワーク、データベース、ディスク、外部サービスの
待ち時間が大部分を占める

CPUバウンド：
Pythonによる計算時間が大部分を占める

非同期処理が特に適しているのは、

  • LLM APIリクエスト、
  • ベクトルデータベース検索、
  • HTTPツール呼び出し、
  • データベースリクエスト、
  • メッセージキュー、
  • ストリーミングレスポンスです。

一方、次の処理が自動的に高速化されるわけではありません。

  • 大規模な行列計算、
  • テキスト圧縮、
  • 画像処理、
  • pure Pythonで書かれた複雑なループ。

第2部：同期イテレーション体系

async forを理解する前に、通常のforを支えるプロトコルを理解しましょう。

1. Iterable：イテラブル

Iterableとは、

iter(obj)によってイテレーターを取得できるオブジェクトです。

代表的なIterableは次のとおりです。

list
tuple
dict
set
str
range

例：

numbers = [10, 20, 30]
iterator = iter(numbers)

より正確には、Iterableはイテレーターの供給元です。繰り返し走査できる多くのコンテナは、iter()を呼ぶたびに新しい独立したイテレーターを返します。ただし、すべてのIterableが反復可能でなければならないとプロトコルが要求しているわけではありません。

2. Iterator：イテレーター

Iteratorは、実際に走査状態を保持するオブジェクトです。

次のメソッドをサポートする必要があります。

__iter__()
__next__()
  • __iter__()はイテレーター自身を返します。
  • __next__()は次の要素を返します。
  • 要素がなくなるとStopIterationを送出します。

例：

numbers = [10, 20, 30]

it1 = iter(numbers)
it2 = iter(numbers)

next(it1)  # 10
next(it1)  # 20

next(it2)  # 10

it1とit2は、それぞれ独自の現在位置を保持します。

Iteratorにiter()を呼ぶと自分自身を返す理由

it = iter([1, 2, 3])

iter(it) is it  # True

すでにイテレーターなので、さらにラッパーを作る必要がないためです。

ただし、本来の意味での「ファクトリー」と呼ぶべきではありません。イテレーションプロトコルを満たすために自分自身を返すだけで、リストのように新しい独立したカーソルを作るわけではないからです。

Iteratorが通常一度しか使えない理由

it = iter([1, 2, 3])

list(it)  # [1, 2, 3]
list(it)  # []

最初の呼び出しでカーソルが末尾まで進んでいるため、二回目には何も残っていません。

3. Generator：ジェネレーター

ジェネレーターは、yieldを使って記述する特殊なIteratorです。

def countdown(n: int):
    while n > 0:
        yield n
        n -= 1

ジェネレーター関数を呼び出しても、

generator = countdown(3)

関数全体が一度に実行されるわけではありません。ジェネレーターオブジェクトが返ります。

次を呼ぶたびに、

next(generator)

関数は前回yieldで停止した場所の続きから再開します。

ジェネレーターの主な価値は、

  • 遅延計算、
  • 逐次的な出力、
  • ピークメモリの削減、
  • データストリームの表現です。

関係は次のように覚えられます。

GeneratorはIteratorの一種
Iterator自身もIterable

4. forループの正体

次のコードは、

for item in source:
    consume(item)

概念的には次に近い処理です。

iterator = iter(source)

while True:
    try:
        item = next(iterator)
    except StopIteration:
        break

    consume(item)

forはイテレーションプロトコルの構文糖衣です。

第3部：四つの関数形態

defとasync def、returnとyieldは、互いに独立した二つの軸です。

定義形式呼び出した結果代表的な消費方法
def + return通常の戻り値直接使用
def + yieldGeneratorfor
async def + returnCoroutine objectawait
async def + yieldAsync generatorasync for

1. 通常の関数

def add(a: int, b: int) -> int:
    return a + b

呼び出すとすぐに実行されます。

result = add(1, 2)

2. 同期ジェネレーター

def generate_numbers():
    yield 1
    yield 2

呼び出すとジェネレーターが返ります。

generator = generate_numbers()

3. コルーチン関数

async def fetch_data() -> str:
    return "done"

呼び出すとコルーチンオブジェクトが返ります。

coroutine = fetch_data()

この時点では通常、関数本体はまだ実行されていません。

4. 非同期ジェネレーター

async def stream_tokens():
    yield "A"
    yield "B"

呼び出すと非同期ジェネレーターが返ります。

stream = stream_tokens()

次のように消費します。

async for token in stream:
    print(token)

非同期ジェネレーターは値を伴わないreturnで終了できますが、通常のコルーチンのようにreturn some_valueを使うことはできません。

第4部：コルーチン関数とコルーチンオブジェクト

これは非同期処理を理解するうえで最も重要な区別の一つです。

1. Coroutine function：コルーチン関数

async def fetch_data() -> str:
    return "done"

fetch_dataはコルーチン関数です。

次のように確認できます。

import inspect

inspect.iscoroutinefunction(fetch_data)  # True

2. Coroutine object：コルーチンオブジェクト

次の呼び出しは、

coroutine = fetch_data()

コルーチンオブジェクトを返します。

inspect.iscoroutine(coroutine)  # True
inspect.isawaitable(coroutine)  # True

inspectは、実行時オブジェクトを調べるPython標準ライブラリです。inspect.isawaitable(obj)は、そのオブジェクトをawait式で使用できるかどうかを判定します。2

3. コルーチン関数を呼び出してもすぐ実行されない理由

通常の関数：

def normal():
    print("running")

normal()
# 立即打印

コルーチン関数：

async def async_function():
    print("running")

coroutine = async_function()
# 通常还没有打印

コルーチン関数の呼び出しが主に行うのは、

コルーチンオブジェクトを作る
ローカルな実行状態を保存する
後でawaitまたはスケジュールされるのを待つ

という処理です。

awaitもスケジュールも行わなければ、

async_function()

通常は次の警告が出ます。

RuntimeWarning: coroutine was never awaited

Python公式ドキュメントにも、コルーチン関数を呼び出すだけでは実行がスケジュールされないと明記されています。3

第5部：awaitableとは何か

1. 定義

Awaitableとは、

awaitの後ろに置くことができ、最終的に結果、例外、またはキャンセル状態を生成するオブジェクトです。

result = await awaitable

Pythonで代表的なawaitableは次の三種類です。

Coroutine
Task
Future

Pythonデータモデルでは、一般的なカスタムawaitableは通常__await__()プロトコルで実装され、__await__()はイテレーターを返す必要があると定められています。4

2. Coroutine：非同期実行の記述

async def load_user() -> dict:
    ...

呼び出して、

coroutine = load_user()

得られるオブジェクトは、

この非同期コードをどのように実行するか

を記述します。

コルーチンを直接awaitしても、通常は別の独立したTaskが作られるわけではありません。現在のTaskがそのコルーチンへ入り、完了するか待機が必要な処理に到達するまで実行を進めます。

3. Task：イベントループにスケジュールされたコルーチン

Taskはコルーチンをスケジュールするラッパーです。

task = asyncio.create_task(load_user())

これは、

このコルーチンを現在のイベントループへ登録し、独立したTaskとしてできるだけ早く実行できるようにする

という意味です。

Taskは次の操作も提供します。

task.cancel()
task.done()
task.result()
task.exception()

Task自身もawaitableです。

user = await task

4. Future：将来の結果を表すプレースホルダー

Futureは、

非同期処理は後で完了するが、現時点では結果がまだ用意されていない

という状態を表します。

PENDING
→ FINISHED / CANCELLED

Futureには最終的に、

  • 戻り値、
  • 例外、
  • キャンセル状態

が格納されます。

低レベルライブラリでは、コールバック式I/Oをasync / awaitモデルへ接続するためにFutureがよく使われます。

アプリケーション層でFutureを自分で作る必要は通常ありません。Python公式ドキュメントでも、Futureは低レベルのawaitableと位置付けられています。3

TaskとFutureの関係

次のように考えられます。

Future：
「将来、結果が得られる」ことを表す

Task：
「コルーチンを実行し、その最終結果を
Future-likeなオブジェクトとして表す」

次の二つを混同しないでください。

asyncio.Future
concurrent.futures.Future

これらは異なる並行処理体系に属します。5

第6部：awaitは何をしているのか

1. まず右辺の式を評価する

次のコードでは、

result = await operation()

最初に「イベントループへ実行権を返す」わけではありません。まず、

operation()

を呼び出し、その戻り値がawaitableかどうかを確認します。

これがawait time.sleep(5)を理解する鍵です。

2. awaitableがまだ完了していない場合

概念的な流れは次のとおりです。

現在のTaskがawaitを実行
→ awaitableが未完了
→ 現在のTaskが一時停止
→ イベントループが別の実行可能なTaskを実行
→ awaitableが完了
→ 現在のTaskが再開
→ await式が結果を生成

3. awaitableがすでに完了している場合

待機対象がすでに完了していれば、awaitは結果を直ちに取得して処理を続けることがあり、目に見えるタスク切り替えが起きるとは限りません。

したがって、より正確には、

awaitは一時停止の機会を提供しますが、すべてのawaitが必ずスケジューリング切り替えを起こすわけではありません。

ただしasyncio.sleep()は明確な例外です。公式ドキュメントには、他のTaskを実行できるよう、常に現在のTaskを一時停止すると説明されています。3

4. awaitは自動的にスレッドを作らない

次のコードは、同期関数を自動的にバックグラウンドへ送るものではありません。

await blocking_function()

Pythonはまずblocking_function()を同期的に実行します。

その戻り値がawaitableである場合に限り、awaitが働きます。

第7部：await time.sleep(5)が成立しない理由

誤ったコード：

import time

async def example() -> str:
    await time.sleep(5)
    return "done"

実際の実行順序は次のとおりです。

1. time.sleep(5)を呼ぶ
2. イベントループスレッドを同期的に5秒間ブロックする
3. time.sleep()がNoneを返す
4. Pythonがawait Noneを実行しようとする
5. TypeErrorを送出する

エラーは次のようになります。

TypeError: object NoneType can't be used in 'await' expression

理由は、

time.sleep(5) is None

だからです。

正しい書き方：

import asyncio

async def example() -> str:
    await asyncio.sleep(5)
    return "done"

違いは次のとおりです。

time.sleep(5)
= 現在のスレッドをブロックする

await asyncio.sleep(5)
= 現在のTaskを一時停止し、イベントループは別のTaskを実行できる

time.sleep()のような同期ブロッキング関数をどうしても呼ぶ必要がある場合は、ワーカースレッドへ移せます。

import asyncio
import time

async def example() -> str:
    await asyncio.to_thread(time.sleep, 5)
    return "done"

第8部：イベントループ、Coroutine、Taskの連携

1. イベントループがスケジュールするもの

イベントループが主に処理するのは、

スケジュール済みのTask
登録済みのコールバック
実行可能になったI/Oイベント
タイマー

です。「ソースコード中の関数名」を直接スケジュールするわけではありません。

2. 現在のTaskとは何か

例：

async def main():
    await fetch_a()
    await fetch_b()

asyncio.run(main())が開始すると、main()は一つのTask内で実行されます。

そのTaskの実行経路は次のとおりです。

mainへ入る
→ fetch_a()を呼ぶ
→ fetch_aコルーチンを進める
→ Aを待つ
→ Aが完了する
→ mainの次の行へ戻る
→ fetch_b()を呼ぶ

前の行が完了するまで、fetch_b()は呼ばれてすらいません。

3. 一つのTaskが持つ命令位置は同時に一つだけ

現在のTaskが、

await fetch_a()

で一時停止しながら、自分自身の次の行である、

await fetch_b()

を同時に実行することはできません。

一つのコルーチンオブジェクトが保持する現在の実行位置は一つだけだからです。

AとBを同時に進めるには、二つの独立したTaskへ分ける必要があります。

第9部：直接awaitを続けると逐次実行になる理由

次のコード：

result_a = await fetch_a()
result_b = await fetch_b()

タイムラインは次のとおりです。

main Task
  |
  |-- fetch_a()を呼ぶ
  |-- Aを待つ
  |   main Taskが一時停止
  |
  |   イベントループは、すでに存在する別のTaskを実行できる
  |   しかしBはまだ呼ばれておらず、Taskも存在しない
  |
  |-- Aが完了
  |-- main Taskが再開
  |-- fetch_b()を呼ぶ
  |-- Bを待つ

したがって、順序は必ず、

Aが完了
→ その後でBが開始

となります。

イベントループが次の行へ飛べない理由

次の行も、一時停止中の同じmain Taskに属しているからです。

イベントループが切り替えられるのは、

すでにスケジュールされた別のTask

であり、

現在のTaskでまだ実行されていない次の行

ではありません。

第10部：create_task()で並行実行できる理由

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

result_a = await task_a
result_b = await task_b

最初の二行ですでに、二つの独立したTaskが作られています。

Task A
Task B

main Taskが、

await task_a

を実行して一時停止すると、イベントループは、

Task A
Task B

を実行できます。

したがって、コード上は先にtask_aを待っていても、task_bもバックグラウンドで進行できます。

1. 所要時間の比較

AとBがどちらも2秒待つとします。

逐次実行

await fetch_a()
await fetch_b()

約4秒かかります。

並行実行

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

await task_a
await task_b

約2秒です。

これは通常、シングルスレッド上のI/O並行処理です。二つのPython関数が二つのCPUコアで並列実行されるという意味ではありません。

2. create_task()の戻り値を保持する

公式ドキュメントによれば、イベントループはTaskへの弱い参照しか保持しません。確実に管理する必要があるバックグラウンドTaskには強い参照を保持し、例外も処理してください。3

background_tasks: set[asyncio.Task[object]] = set()

task = asyncio.create_task(do_work())
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)

より一般的で安全な方法は、構造化並行性を使うことです。

第11部：TaskGroup、gather、構造化並行性

1. TaskGroup：関連する並行Taskの第一選択

Python 3.11以降ではasyncio.TaskGroupを利用できます。

import asyncio

async def main() -> None:
    async with asyncio.TaskGroup() as group:
        task_a = group.create_task(fetch_a())
        task_b = group.create_task(fetch_b())

    result_a = task_a.result()
    result_b = task_b.result()

async withブロックを抜けるとき、グループ内のすべてのTaskが完了するまで待機します。

利点は次のとおりです。

  • 子Taskのライフサイクルがブロック内に限定される。
  • 一つの子Taskが失敗すると、関連する他のTaskがキャンセルされる。
  • 例外がExceptionGroupへ集約される。
  • Taskや例外を見落としにくい。

Python公式ドキュメントでは、TaskGroupはcreate_task()を手動管理する方法より新しく、より強い安全性を備えた並行処理管理手法として位置付けられています。3

2. gather：並行処理の結果を集める

results = await asyncio.gather(
    fetch_a(),
    fetch_b(),
)

gather()は、渡されたコルーチンを自動的にTaskとしてスケジュールし、入力順に結果を返します。

次の場面に適しています。

  • 複数のawaitableを手軽に並行実行したい。
  • 固定の順序で結果を集めたい。
  • 例外伝播の方針を明確に制御できる。

ただし、デフォルトでは一つのTaskが失敗しても、他のTaskがすべて自動的にキャンセルされるわけではありません。関連するTaskのライフサイクルをそろえ、一緒に失敗させたい場合はTaskGroupを優先してください。3

第12部：キャンセルとタイムアウト

非同期Taskでは成功だけでなく、次の状況も考える必要があります。

ユーザーが接続を切る
上流リクエストがタイムアウトする
サービスが停止する
並行実行中の子Taskの一つが失敗する

1. Taskのキャンセル

task.cancel()

キャンセル要求を受けたTaskでは、次の適切な機会に、

asyncio.CancelledError

が送出されます。

コルーチンではtry/finallyを使ってリソースを解放します。

async def worker() -> None:
    resource = await acquire_resource()

    try:
        await do_work(resource)
    finally:
        await release_resource(resource)

通常、CancelledErrorを握りつぶしてはいけません。TaskGroupとasyncio.timeout()はいずれもキャンセル機構に依存しています。3

2. タイムアウト

async def call_model() -> str:
    try:
        async with asyncio.timeout(10):
            return await llm_request()
    except TimeoutError:
        return "request timed out"

タイムアウトは内部で現在のTaskをキャンセルし、コンテキストマネージャーがそのキャンセルをTimeoutErrorへ変換します。

第13部：async forと非同期イテレーション

通常のforは、要素を取得するたびに同期的なnext()を呼びます。

「次の要素を取得する」処理自体に待機が必要な場合、たとえば、

  • 次のLLM Tokenを待つ。
  • WebSocketメッセージを待つ。
  • APIの次のページを待つ。
  • データベースカーソルが次の行を返すのを待つ。

といった場合には、非同期イテレーションが必要です。

1. AsyncIterableとAsyncIterator

非同期イテレーションプロトコルには、

__aiter__()
__anext__()

が含まれます。

規則は次のとおりです。

  • __aiter__()は非同期イテレーターを返す。
  • __anext__()はawaitableを返す。
  • イテレーションが終わるとStopAsyncIterationを送出する。

Pythonデータモデルには、このプロトコルが正式に定義されています。4

2. async forの概念的な展開

async for item in source:
    consume(item)

は、概念的には次に近い処理です。

iterator = source.__aiter__()

while True:
    try:
        item = await iterator.__anext__()
    except StopAsyncIteration:
        break

    consume(item)

通常のforとの重要な違いは、

次の要素を取得する処理でawaitできる

ことです。

3. LLMストリーミングの例

async def stream_answer():
    async for token in llm_client.stream("Explain RAG"):
        yield token

呼び出し側：

async for token in stream_answer():
    print(token, end="")

これは、

async def + yield
= 非同期ジェネレーター

です。

4. 非同期ジェネレーターを途中で抜ける場合は明示的に閉じる

途中でbreakし、その非同期ジェネレーターが接続やカーソルなどのリソースを保持している場合は、明示的に閉じることを推奨します。

import contextlib

async with contextlib.aclosing(stream_answer()) as stream:
    async for token in stream:
        if should_stop(token):
            break

Python 3.14のasyncio開発ドキュメントでは、早期終了などの場合に非同期ジェネレーターを明示的に閉じ、予測できないコンテキストで後からクリーンアップ処理が動かないようにすることを推奨しています。1

第14部：async withと非同期リソース管理

通常のコンテキストマネージャー：

with resource:
    ...

は、

__enter__()
__exit__()

に依存します。

非同期コンテキストマネージャー：

async with resource:
    ...

は、

__aenter__()
__aexit__()

に依存します。

後者の二つのメソッドはawaitableを返す必要があります。4

概念的には次のように展開できます。

manager = create_manager()
resource = await manager.__aenter__()

try:
    await use(resource)
finally:
    await manager.__aexit__(...)

次の用途に適しています。

  • 非同期HTTP接続、
  • データベーストランザクション、
  • WebSocket、
  • 非同期ロック、
  • タイムアウトスコープ、
  • TaskGroup、
  • Traceのライフサイクル。

第15部：非同期プログラムにもスレッドプールが必要な理由

1. asyncはサードパーティ製同期ライブラリの性質を変えない

あるライブラリが同期インターフェースしか提供していないとします。

response = requests.get(url)

これをasync def内に書いても、

async def handler():
    response = requests.get(url)

イベントループスレッドをブロックすることに変わりはありません。

2. asyncio.to_thread()

置き換えが難しい同期ブロッキングI/Oには、次のように書けます。

response = await asyncio.to_thread(requests.get, url)

実行関係は、

同期関数
→ ワーカースレッドで実行

イベントループスレッド
→ 結果をawaitしている間も別のTaskを実行

となります。

asyncio.to_thread()はコルーチンオブジェクトを返すため、awaitできます。

3. スレッドプールは関数を「本当の非同期」にするものではない

スレッドプールが行うのは、

ブロッキング処理を別のスレッドへ移し、イベントループスレッドを塞がないようにすること

です。

ブロッキング自体は依然として存在します。唯一のイベントループ料理人を塞がなくなっただけです。

4. 待機のキャンセルはスレッドの停止ではない

to_thread()を待っているTaskがキャンセルされても、通常、Pythonは基盤となるスレッドを強制停止できません。結果を待つ側がいなくなっても、そのスレッド内の同期関数は実行を続けることがあります。

したがって、

スレッドによる橋渡し

を、

自由にキャンセルできるネイティブ非同期処理

と誤解しないでください。

第16部：CPUバウンドな処理はどうするか

asyncio.to_thread()は主にブロッキングI/O向けです。

大量のpure PythonによるCPU計算では、通常、スレッドから理想的なマルチコア並列性は得られません。

代表的な選択肢は、

ProcessPoolExecutor
InterpreterPoolExecutor
専用タスクキュー
外部計算サービス
GILを解放するネイティブ拡張

です。

Python公式ドキュメントでは、ブロッキング処理やCPU負荷の高いコードをイベントループスレッド上で直接実行しないことを推奨しています。スレッド、独立したインタープリター、プロセスExecutorによって隔離できます。1 concurrent.futuresは、スレッドプール、インタープリタープール、プロセスプールに共通のインターフェースを提供します。5

一般的なプロセスプールの例：

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(value: int) -> int:
    return sum(i * i for i in range(value))

async def main() -> int:
    loop = asyncio.get_running_loop()

    with ProcessPoolExecutor() as executor:
        return await loop.run_in_executor(
            executor,
            cpu_heavy,
            10_000_000,
        )

第17部：Starlette / FastAPIがスレッドプールを使う理由

Starletteは、同期コードによるイベントループのブロックを避けるため、次のような場面でワーカースレッドを使います。

  • defで定義した同期endpoint、
  • 同期BackgroundTask、
  • ファイルレスポンス、
  • ファイルアップロード、
  • 一部の内部同期処理。

Starletteは現在、anyio.to_thread.run_sync()によってこれらの同期コードを実行します。公式ドキュメントによれば、デフォルトのスレッド容量制限は40 tokenであり、FastAPIなどとも共有されます。むやみにスレッド数を増やすと、メモリ消費とコンテキストスイッチのコストが増える可能性があります。6

1. iterate_in_threadpool()の意味

同期IterableをAsyncIteratorとして公開します。

async def iterate_in_threadpool(iterator):
    sync_iterator = iter(iterator)

    while True:
        try:
            item = await anyio.to_thread.run_sync(
                next_item,
                sync_iterator,
            )
        except EndOfIterator:
            break

        yield item

中核となる処理は、

同期next()
→ スレッドプールで実行
→ 結果をawait
→ 非同期呼び出し側へyield

です。

関数全体は、

async def + yield

なので、async forで消費できる非同期ジェネレーターです。

2. StarletteがStopIterationを別の例外へ変換する理由

同期Iteratorは終了時に、

StopIteration

を送出します。

しかし、スレッドプール内で送出されたStopIterationを、そのままFuture / await境界の外へ渡して捕捉することはできません。Starletteはワーカースレッド内でカスタム例外へ変換し、非同期ジェネレーター側でその例外を捕捉してループを終了します。

この詳細から、

同期イテレーションプロトコルと非同期待機プロトコルは橋渡しできるものの、同じプロトコルではない

ことが分かります。

第18部：AIアプリケーションでの代表的なパターン

1. 独立した複数ツールを並行呼び出しする

import asyncio

async def collect_context(query: str) -> tuple[dict, dict]:
    async with asyncio.TaskGroup() as group:
        web_task = group.create_task(search_web(query))
        db_task = group.create_task(search_database(query))

    return web_task.result(), db_task.result()

前提は、二つの処理が互いに独立していることです。

2. 逐次実行が必要なAgentステップ

plan = await create_plan(user_message)
tool_result = await execute_tool(plan)
answer = await generate_answer(tool_result)

「並行実行を追求する」ためだけに三つのTaskへ変えてはいけません。なぜなら、

execute_toolはplanに依存する
generate_answerはtool_resultに依存する

からです。

並行実行が適しているのは、依存関係がなく、同時に開始できる処理だけです。

3. LLM Tokenをストリーミングする

async def stream_response(prompt: str):
    async for event in llm.stream(prompt):
        if event.type == "token":
            yield event.text

4. 同期ドキュメントパーサーを非同期サービスへ組み込む

async def parse_document(path: str) -> list[str]:
    return await asyncio.to_thread(sync_parser.parse, path)

解析がCPU負荷の高い処理なら、スレッドプールを長時間占有するのではなく、プロセスプールや専用Workerを検討してください。

5. 並行数の制限

数千件の外部API呼び出しに対して、無制限にTaskを作ってはいけません。

Semaphoreを利用できます。

import asyncio

limit = asyncio.Semaphore(10)

async def limited_call(item: str) -> str:
    async with limit:
        return await call_external_api(item)

これにより、

  • LLMの同時実行数、
  • ツールの同時実行数、
  • データベース接続への負荷、
  • APIレート制限のリスク

を抑えられます。

第19部：非同期プログラムのデバッグ

1. asyncio Debug Modeを有効にする

asyncio.run(main(), debug=True)

次の環境変数も使えます。

PYTHONASYNCIODEBUG=1

Debug Modeは、

  • 遅いコールバック、
  • 誤ったスレッドからの呼び出し、
  • awaitされていないコルーチン、
  • リソース問題

の検出に役立ちます。

1

2. オブジェクトの種類を確認する

import inspect

inspect.iscoroutinefunction(func)
inspect.iscoroutine(obj)
inspect.isawaitable(obj)
inspect.isasyncgenfunction(func)
inspect.isasyncgen(obj)

これらの確認は、学習、フレームワーク開発、サードパーティSDKの戻り値型を調べるときに役立ちます。通常のアプリケーションコードでは、あらゆる場所で動的に確認するのではなく、型注釈とドキュメントを優先してください。

3. よくある二つの警告

Coroutine was never awaited

原因：

fetch_data()

修正：

await fetch_data()

または、

task = asyncio.create_task(fetch_data())
await task

Task exception was never retrieved

通常の原因は、バックグラウンドTaskを作ったものの、結果を待たず、例外も取得していないことです。

TaskGroupを優先してください。本当にバックグラウンドTaskが必要なら、参照を保持して例外を処理します。

第20部：よくある誤解

誤解1：async defを書けば自動的に非同期になる

誤りです。

async defで定義した関数を呼ぶと、コルーチンオブジェクトが作られるだけです。関数本体が同期ブロッキングコードを呼べば、イベントループは依然としてブロックされます。

誤解2：awaitを書けば必ずノンブロッキングになる

誤りです。

await sync_function()

では、まずsync_function()が同期的に実行されます。

誤解3：すべてのawaitで必ずTaskが切り替わる

正確ではありません。

awaitableがすでに完了していれば、処理がすぐ続くことがあります。awaitが提供するのは、一時停止できる能力です。

誤解4：二つの関数を直接awaitすれば自動的に並行実行される

誤りです。

await a()
await b()

Aが完了するまでBは呼ばれてすらいません。

誤解5：イベントループが現在のコルーチンの次の行を実行する

誤りです。

現在のTaskがawaitで一時停止すると、そのTaskの残りのコードも一時停止します。イベントループが実行できるのは、すでにスケジュールされた別のTaskだけです。

誤解6：Taskはスレッドである

誤りです。

Taskはイベントループ内でコルーチンをスケジュールする単位であり、通常は同じスレッド上で実行されます。

誤解7：FutureとTaskは同じである

完全には正しくありません。

Futureは将来の結果を表す低レベルのプレースホルダーです。Taskはコルーチンを実行し、その最終結果をFuture-likeな形で表します。

誤解8：スレッドプールを使えばCPU計算が自動的にマルチコア並列になる

通常は成立しません。

pure PythonのCPUバウンド処理には、プロセスプール、インタープリタープール、専用計算サービスの方が適しています。

誤解9：並行数は多いほど速い

誤りです。

Taskやスレッドが多すぎると、

  • APIレート制限、
  • データベース接続の枯渇、
  • メモリ使用量の増加、
  • コンテキストスイッチの増加、
  • 下流サービスの過負荷

を引き起こします。

並行処理には制限とバックプレッシャーが必要です。

第21部：完全な早見表

概念意味代表的な作成方法消費方法
IterableIteratorを提供できるオブジェクトlist、str、独自の__iter__for
Iterator走査位置を保持するオブジェクトiter(source)next() / for
Generatoryieldを使ったIteratorジェネレーター関数を呼ぶnext() / for
Coroutine functionasync defで定義した関数async def f()呼び出すとCoroutineを得る
Coroutine objectコルーチン関数の呼び出し結果f()awaitまたはスケジュール
Awaitableawaitで使用できるオブジェクトCoroutine / Task / Futureawait
Taskイベントループにスケジュールされたコルーチンcreate_task() / TaskGroupawait
Future将来の結果を表す低レベルオブジェクト通常はライブラリまたはイベントループが作成await
AsyncIterableAsyncIteratorを提供できるオブジェクト__aiter__()async for
AsyncIterator非同期に要素を生成するオブジェクト__anext__()async for
AsyncGeneratorasync def + yield非同期ジェネレーター関数を呼ぶasync for
Async context manager非同期にリソースを取得・解放する__aenter__、__aexit__async with

第22部：十五の実務ルール

  1. コルーチン関数を呼ぶとコルーチンオブジェクトが作られるだけで、自動的には実行されない。
  2. コルーチンオブジェクトを作ったまま、awaitもスケジュールもしない状態にしない。
  3. awaitの後ろの式は、最終的にawaitableを生成しなければならない。
  4. 直接awaitを続ける処理は逐次制御フローである。
  5. 並行TaskにはTaskGroupを優先する。
  6. データ依存関係のあるステップは逐次実行を維持する。
  7. イベントループスレッド上で同期ブロッキングI/Oを呼ばない。
  8. 同期I/Oはto_thread()で一時的に橋渡しできる。
  9. CPU負荷の高い処理には、プロセスプールまたは専用Workerを優先する。
  10. Taskはスレッドではなく、asyncioの並行性はCPU並列性ではない。
  11. Semaphoreなどを使って外部呼び出しの同時実行数を制限する。
  12. Taskがキャンセルされたときはfinallyでリソースを解放する。
  13. CancelledErrorを安易に握りつぶさない。
  14. 非同期ジェネレーターを途中で抜ける場合は、明示的なクローズを検討する。
  15. 開発中はasyncio Debug Modeを有効にする。

まとめ

Pythonの非同期体系には多くの概念がありますが、中心となる流れは明快です。

Iterable / Iterator
「データを一つずつ取得する方法」を解決する

Coroutine
「一時停止と再開が可能な非同期実行」を記述する

Awaitable
「何をawaitできるか」を定義する

Task
Coroutineをイベントループへ渡してスケジュールする

Future
「将来得られる結果」を表す

Event Loop
すでに存在するTask、コールバック、I/Oイベントをスケジュールする

async for
要素を非同期に取得する

async with
リソースを非同期に取得・解放する

Thread / Process Pool
ネイティブに非同期化できないブロッキング処理を
イベントループスレッドの外へ移す

最後に、最も混乱しやすい問題へ戻ります。

await fetch_a()
await fetch_b()

なぜ逐次実行になるのでしょうか？

最初の行で現在のTaskが一時停止するからです。二行目も同じTaskに属しているため、最初の行が完了するまで実行できません。この時点ではfetch_b()はまだ呼ばれておらず、独立したTaskも存在しないため、イベントループにはスケジュールできるBがありません。

一方、

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

では、二つの独立したTaskが明示的に作られています。これによって初めて、AとBがそれぞれI/Oを待つ間に、イベントループが両方を交互に進められます。

ここを理解すれば、コルーチン、Task、イベントループ、並行性の関係が一つにつながります。

参考資料

  1. Python 3.14 Documentation: Developing with asyncio ↩︎ ↩︎ ↩︎ ↩︎

  2. Python 3.14 Documentation: inspect — Inspect Live Objects ↩︎

  3. Python 3.14 Documentation: Coroutines and Tasks ↩︎ ↩︎ ↩︎ ↩︎ ↩︎ ↩︎ ↩︎

  4. Python 3.14 Documentation: Data Model — Coroutines, Async Iterators and Async Context Managers ↩︎ ↩︎ ↩︎

  5. Python 3.14 Documentation: concurrent.futures ↩︎ ↩︎

  6. Starlette Documentation: Thread Pool ↩︎