【Python】結局async/await とか yieldとかってなんなのさ

Pythonを学ぶ上で、async/awaitやyieldの扱いは一つの躓きポイントだと考えています。

これらは一見して通常の関数とは大きく異なる制御フローを提供しているかのように思えます。
その結果、理解を後回しにして「なんとなく」使っている、あるいはチームメンバーのソースコードを「なんとなく」通しているということも実は多いのではないでしょうか。
この記事を通して、async/awaitやyieldの機能を十分に理解いただき、例えばそれらが混在したソースコードすらも「何が起きているか」を理解しながら読める状態になっていただければと思います。

なお、この記事は全体を通して「関数・クラスの概念は十分に理解し、これからasync/await, yield機能を学ぼうとする人（それはPython初学者でも熟練者でも起こり得る）」を想定しています。よってasync/await, yieldの全体感をまず把握してもらうために多少大雑把（かつ正確でない）言い回しをする場合があります。

今後皆様がさらに学習を進めるにつれ、「アイツの記事ではここが雑に説明されていたが、こういうことだったんだなぁ」と振り返ることもあるかもしれませんが、あくまで初学者向けの説明ということで何とぞご容赦いただければと思います。

0. ロードマップ

本編に入る前にまず明示しなければならないことは、async/awaitと、yieldは全く別の文脈で用いられる機能 だということです。よく一緒に使われることが多いから混同しがちですが、例えば片方だけを使った実装というのも可能です（それも、往々にしてあります）。

async/await, yieldを理解する上でのロードマップを以下のように定めます：

① async/awaitを単品として理解する
② yieldを単品として理解する
③ async/awaitとyieldの組み合わせを理解する：これは単純な①＋②の足し算では済みません。async/awaitとyieldが合体したときに特有の問題が生まれてしまうため、追加のルールが必要になります。

これらを順番に抑えていくことで、async/await, yieldを自在に操ることができるようになるかと考えています。

1. async/await（非同期処理）

さて、「async/await」について、どこまで理解されていますでしょうか？（完全に理解している人は本節を飛ばしていただいてかまいません）

以下のような理解をされている方が多いかもしれません：
(1) なんとなくだけど、「非同期処理」で使うんでしょ？
(2) asyncとawaitはセットで使われるんでしょ？
(3) asyncioが必要になってくるんでしょ？

(1)(2)についてはおよそその理解で合っています。以下のサンプルが最も単純な例でしょう：

import asyncio

async def f():
    res = await asyncio.sleep(1)

（本稿では関数を使って例を示しますが、クラスに対して定義されたメソッドであっても構いません。単にasync/await, yieldの性質を述べるにあたり関数/メソッドの違いは本質的でないため、簡単のため関数に統一して説明します。実運用上はクラスのメソッドに対してasync/await, yieldを適用することが多いことに注意してください。）

一方、(3)については明確に否定させてください。上のサンプルコードでも「たまたま」使ってしまったし、以降の記述でも何度も登場しますが、asyncioは必ずしも必須のライブラリではありません：
（事実1）async/awaitを実行するには「特殊なライブラリ」が必要
（事実2）その「特殊なライブラリ」の一つがasyncio
（事実3）だが、他にもそれが可能なライブラリは存在する

例えば、他にはtrio, anyioなどの選択肢もあります。
あるいは、非同期処理の実行をフレームワークレベルで担保していることも多いです（FastAPIなど）。この場合、アプリエンジニアはasync関数を実装する一方で、それがどのように非同期処理されるかについて関心を寄せる必要がないといったこともあります。

さて、「async/awaitはセット」はナイーブな理解ですが、より具体的には以下のような順番で考えるのがよいです：

• awaitは、「この処理をしている間、（I/O待ちなどの）暇な時間は他の関数を実行してもいいよ」という宣言
• asyncは、「この関数の内部には、awaitが含まれている可能性があるよ。従って、それを前提として非同期的に実行する必要があるよ」という宣言

ひとつひとつ順番に見ていきましょう：

1-1. awaitは、「この処理をしている間、（I/O待ちなどの）暇な時間は他の関数を実行してもいいよ」という宣言

例えば外部API実行（SQLや、LLM実行など）では外部サーバーのCPUがひたすら頑張っている一方で、手持ちのPythonランタイム自体は暇していることが多いです。
そうした場合、その暇な時間を別の処理に回すべきという考えが生まれます。
そこで、「この処理をしている間は暇になることもあろうから、その場合は他の処理に切り替えてもいいよ。awaitの処理が終わったら戻ってきてね」とする宣言がawait宣言です：

await execute_llm(prompt)  # LLMの実行時間中、別の処理をしていてもよい

ただし、何でもかんでもawaitできるというわけではないことには注意が必要です。
例えば以下のような実装は成立しません（エラーが発生します）

# ダメな例（エラーが発生します）
import time

async def f():
    res = await time.sleep(1)

「awaitできるかどうか」は、用語として（そのままですが）awaitableとして定義されます。
例えば、asyncio.sleep(0)はawaitableな一方、time.sleep(1)はawaitableではありません。

では、どのようなオブジェクトがawaitableになるのでしょうか。
初学者は、ひとまず以下のようなオブジェクトがawaitableだと考えておくと良いです：

• 外部ライブラリよりasync関数として提供されている関数（の実行結果）
• （後述の）asyncによって定義される関数（の実行結果）

実際には 上記以外にもawaitableになるパターンはあります が、まず全体感をつかんでいただくことが目標なので本稿ではこのパターンのみを扱うことにします。

1-2. asyncは「この関数の内部には、awaitが含まれている可能性があるよ。従って、それを前提として非同期的に実行する必要があるよ」という宣言

前半の主張には注意が必要です。「可能性がある」と言っているだけなので、実はawaitが含まれていない場合もあり得ます。
例えば、以下のようなソースコードを実行してもランタイムエラーは発生しません。：

async def f():
    return

ただし、awaitしないのなら非同期的な実行は必要ないということなので、async関数である意味があまりありません（通常の関数でOK。実装途中のコードで後からawaitを追加する予定があるなら別ですが）。
よって、多くの場合async関数の内部には1つ以上のawaitがあります。そのため、「asyncとawaitはセット」だと思って実務上差支えはありません。

一方でasync関数内に2つ以上awaitが存在することもできるし、こちらはよくあるパターンです。
「DBからデータ抽出」→「LLMによる分析」→「DBにデータ格納」というそれぞれのステップをawait化するなどは往々にして考えられます。

さて、ここからは後半の主張「それ（awaitが含まれていること）を前提として非同期的に実行する必要がある」に着目しましょう。
async関数の最も顕著な特徴は、単純に実行した場合内部の処理は実行されず、「コルーチン」が生み出されるだけだということです：

import asyncio

async def f():
    await asyncio.sleep(1)
    return 0

if __name__ == "__main__":
    res = f()
    print(res) 
    # 0ではなく、<coroutine object f at 0x00000...> が出力される

はい、いきなり「コルーチン」とかいう単語が出てきました。ちょっと待ってくださいね。
技術的な詳細をとやかく並べ立てるつもりはないので、ここでは一旦「処理の設計図のようなもの」だと理解してもらえれば十分だと思います。
設計図なので、それ単体では実行できません。実行するためには、大元に実行主体となる特殊なライブラリが必要です。
考えてみれば当然のことともいえます。「await行に到達したら、そこでいったん処理を区切り、（必要なら）他の処理に切り替えつつ、awaitが完了したらまた戻って...」というフクザツな処理を行う以上、そのアーキテクチャに特化した仕組みが必要です。
代表的な方法としては、asyncio.runを使う方法が挙げられます：

import asyncio

async def f():
    await asyncio.sleep(1)
    return 0

if __name__ == "__main__":
    res = asyncio.run(f())
    print(res)
    # 結果、0が出力される

なお、「どう実行するか」は大元（main関数側）の1か所でのみ考えればよく、例えば中間関数において「どのようにasync関数を実行するか」を気にする必要はありません：

import asyncio

async def g():
    await asyncio.sleep(1)
    return 0

async def f():
    # fの実行時点でasyncio.runを使っているため、ここでは通常の呼び出しの記法でOK
    res = await g()
    
    return res

if __name__ == "__main__":
    res = asyncio.run(f())
    print(res)
    # 結果、0が出力される

実際にはasyncio.run(f())は非同期処理一本を実行しているだけなので、並行実行の恩恵は得られません。そこで、通常はasyncio.gatherなどで複数の非同期処理を集めて同時並行で実行することが多いです。
あるいは、async関数の実行をフレームワークレベルで担保しているため、個別の機能を実装する上では実行方法をいちいち気にしなくてよいという場合もあります：例えばFastAPI（APIサーバ構築のためのフレームワークライブラリ）は内部でanyioを用いているため、個々のAPI機能の実装をasync関数として実装することができる一方、それをどう並行実行しているかについて通常気にしなくて大丈夫です。

1-3.（補遺）async with

with句はオブジェクトの開始・終了処理を強制するための仕組みです。
特に、なにかしらのリソースの取得・解放を強制する際によく用いられます：

with open("path/to/file") as f:
    f.read()

上記の例だと、with句を開始する時点でf.__enter__メソッドが実行され、with句内部の処理が終了した時点で（処理の成功・失敗に関わらず）f.__exit__メソッドが実行されます。
ここで、__enter__, __exit__処理に時間がかかり、非同期的に実行したい場合が存在します。
そうした場合、__aenter__, __aexit__をそれぞれasync関数として定義するのが一般的です。
そして、

async with DBConnection() as conn:
    ...

のようにwithの代わりにasync withとすることで__aenter__, __aexit__の実行を指示することができます。

...個人的にはasync関数である__aenter__, __aexit__を実行しているのだからawait withの方が自然な気もしますが、pythonがasync withと決めたのでそれに従いましょう。
まぁいろいろあったんでしょうね。

最後に、本節で学んだことをまとめます：

• awaitは「処理を一時中断してもよい地点」を表す
• async関数は、内部に（通常は）1つ以上のawait文を含むことを意味する
• async関数の実行には、何かしらの仕組み（asyncio等）が必要
• async withによって非同期的なリソースの取得・解放を実現できる

2. yield（ジェネレーター）

本節では、一旦 async/awaitを忘れていただいて 、「通常の関数にyieldを付け加えたときに何が起きるか」について説明します。
実際、本節で語る内容はasync/awaitとは一切関係なく、yieldが単体として持つ仕様です。

yieldが持つ性質は、次のように表現できます：
yieldを内部に含む関数は、戻り値としてジェネレーターを返す。ジェネレーターは、for文などでループさせることができる。

またいきなりジェネレーターとか言い出してすみません、とりあえず例を示します：

import time

def timer():
    time.sleep(1)
    yield "1秒後"
    time.sleep(1)
    yield "2秒後"
    time.sleep(1)
    yield "3秒後"

timerを実行すると、（遅延なく即座に）ジェネレーターなるものが出力されます

print(timer())
# <generator object timer at 0x00000...>

上述の通り、ジェネレーターはfor文でループすることができます：

def main():
    generator = timer()
    for msg in generator:
       print(msg)

if __name__ == "__main__":
    main()

このサンプルコードを実行すると、

• 実行から1秒後、「1秒後」と出力されます
• それからさらに1秒後、「2秒後」と出力されます
• それからさらに1秒後、「3秒後」と出力されます

つまり、ジェネレーターが定義された段階（generator = timer()）ではtimer内部の処理は実行されません。for文などでループさせた場合に初めて実行され、yieldした値をループの要素として逐次取り出すことができます。

ここで、timer内の処理がyield文によって分割されることに注意してください。
timer内部の処理としては

1. yield文に到達するまで処理実行
2. yield文に到達したとき、yieldで指定された値を返却（※）
3. 次回処理ではyield文の次の行を開始地点とする（※※）
   のループが繰り返されています。

※ 補足：本稿では yield を「for で値を取り出す」用途として説明しますが、send()/throw()/close() により外部から値や例外を注入する使い方もあります（高度な話題のため本稿では扱いません）。
※※ 値を出力するという用途に限定しても、yieldは「その箇所から処理を再開する」という意味でreturn文とは明確に異なります。yieldを一時停止、returnを停止ボタンのようにイメージいただくのが良いかと思います

ここで、yield文によって「逐次値が取り出される」というのがミソです。例えば

def timer():
    msgs = []
    time.sleep(1)
    msgs.append("1秒後")
    time.sleep(1)
    msgs.append("2秒後")
    time.sleep(1)
    msgs.append("3秒後")
    return msgs

のようにすべての処理を行ったうえでlist型を返すようにした場合でも結果的には同じ値が出力されます。
しかし、実行してから3秒後に初めて全ての値が出力されるという意味で、リアルタイム性が損なわれてしまいます。

リアルタイム性やメモリ効率など、何かしらのパフォーマンス改善が望める場合に使われるのが、このジェネレーター機能（=yield機能）です。
例えば、LLMの小分けされた出力を逐次ユーザーに返却したり、SQLによりテーブルを数百行ずつに分けてフェッチするなどの際に用いられます。

上記の例が、ファイル読み込みに類似していると感じた方もいらっしゃるかもしれません：

with open("path/to/file") as f:
    for line in f:
        print(line)

ここではf（ファイル）にfor文を適用することでファイルを1行ずつ逐次的にフェッチするという構成になっています。
このような「for文によるループが可能」という性質はiterableと呼ばれます。
iterableなオブジェクトは一般にクラスを用いて定義することができますが、yieldはより短い定義でiterableなオブジェクトの一種としてジェネレーターを作成する記法だと捉えていただくのが良いかと思います。

2-1.（補遺）yield挿入の危険性

ここで、「yieldを含む関数はジェネレーターになる」に一抹の空恐ろしさを感じた方もいるのではないでしょうか。
例えば、 「数百行に及ぶ関数（※）の中に、たった1行でもyieldが入っていたら、関数ではなくジェネレーターになってしまうのか？」 という疑問を抱くかもしれません。
※ このようなソースコード構成がそもそもよくないという方もいらっしゃると思いますが、現実問題こうしたソースコードに出くわす場面は少なくありません。

残念ながら、答えは Yes です。特に、関数ではなくなるため通常の関数呼び出しは機能しなくなってしまいます：

def f():
    return "Hello World"
    yield


if __name__ == "__main__":
    print(f())
    # "Hello World"は出力されない
    # <generator object f at 0x00000...>が出力される

例えばこの例では、yieldがreturnの後に書かれている（ので、ランタイムには何の影響もないように見える）にも関わらず、関数の戻り値がreturn文の値ではなくジェネレーターになってしまいます。
これはPythonがソースコードを読み込む段階でyieldの存在を検査しているからです。ランタイム（実行）が始まる前にどの関数がジェネレーターかを判断してしまうため、returnとyieldの前後は関数/ジェネレーターの判断に影響しません。

では開発者として「関数がジェネレーターか否か」を判別するためには、 関数内のすべての行に目を凝らして、チームメンバーが勝手にyieldを差し込んでいないか確認しなくてはいけないのでしょうか？
...ご安心ください。答えは （一部）No です。
VSCodeに適切な拡張を差し込むことで、どの関数がジェネレーターかを判別できるようになります（とはいえ面倒なことに変わりはないのでチームメンバーにはいたずらを控えてもらいたいものですが）。

個人的にはそもそもpythonが言語として、例えば「def gen fだとか@generatorのようなデコレータをつけないとyieldの使用を認めない（勝手にジェネレーターにしない）」みたいな仕様にしてくれてもよかったのではと思いますが、実際にそうなっていないのだからしょうがない。まぁいろいろあったんでしょうね。

2-2.（補遺）ジェネレーター内のreturn値の行方

ジェネレーターは通常の関数と同様、return文で値を返すことができます

def timer():
    time.sleep(1)
    yield "1秒後"
    time.sleep(1)
    yield "2秒後"
    time.sleep(1)
    yield "3秒後"
    return "Finished!"

ただし、これは単純なfor文では抽出できません：

print([res for res in timer()]) 
# ["1秒後", "2秒後", "3秒後"]が出力される一方、"Finished!"は出力されない

return文で出力している値を取得するためには、ループをwhile文で表現する必要があります。
まず、（ジェネレーターとは無関係な仕様として）for文：for e in objは実は以下のソースコードと等価といえます：

it = iter(obj) # ①
while True:
    try:
        e = next(it)  # ②
        ... # for文内部に記載された処理
    except StopIteration: # ③
        break  

ここで、iter, nextはpythonの組み込み関数であり、引数オブジェクトの仕様によって動作を変えるモノだと考えてください（実際には、obj.__iter__, it.__next__を実行しています）。

例えばobjが通常のリストの場合、
① iter(obj)はobj自身（※）と「カウンター」を与える。特に、カウンターは0で初期化される。
※ コピーではなく、objそのもの（参照）を渡す
② next(it)はリスト内のカウンター値に対応する要素を出力し、カウンターをインクリメントする
③ next(it)にてカウンター値がリストの長さに到達したときStopIterationを発生させる（whileループから抜け出す）
という方式でfor文が実行されます。

objがジェネレーターの場合も同様に、for文は以下のように実行されます：
① iter(obj)：obj自身を返す。よって、it = objと読み替えてもよい
② next(it)では、ジェネレーター内の処理を実行する。yieldに到達した場合一時停止し、yield値を出力する。
（次回のnext(it)では一時停止した箇所から処理を再開する）
③ return文に到達するか、処理が全て完了した場合StopIterationを発生させる
（whileループから抜け出す）

ここで、return文に到達した際に発生するStopIteration例外には、return値の情報が含まれています。実際、例外のvalue項目として抽出することができます：

obj = timer()
while True:
    try:
        e = next(obj) 
        ... # ループ内部の処理
    except StopIteration as err:
        res = err.value
        break

（objがジェネレーターの場合にiter(obj)がobj自身になることを利用して、it=iter(obj)を省略し、ループ内のitもobjに置き換えました。）

出力値が例外値として表現されるのは直感に反しますが、ともかくもpythonではこのようにしてジェネレーターの出力値を取得することができます。

補遺2-3. yield from

yield from自体はasync/awaitとyieldの組み合わせを知るうえで重要な機能ではないので、一度この項を飛ばしていただいても構いません。
一方、yield単体の機能という意味では無視できない機能のため説明します。

ジェネレーターを利用すると、しばしばジェネレーターの連鎖が発生する場合があります：
関数SにてジェネレーターAをループするが、Aのyield値は実際にはさらに奥のジェネレーターBのyield値であり...といった具合です。
また、前述の通りジェネレーターのreturn値の取得のためにはtry-exceptを書く必要があり、面倒です。

これらの面倒を一手に引き受けてくれるのがyield fromです：

def generator_b():
    for i in range(3):
        yield i
    return "Finished!"

def generator_a():
    res = yield from generator_b()
    if res == "Finished!":
        print("Success")

def main():
    for msg in generator_a():
        print(msg)

if __name__ == "__main__":
    main()

# 実行結果
# 0
# 1
# 2
# Success

generator_aにてyield fromを利用しています。
これは、「generator_aとして何をyieldするか」を記述していますが、yield fromにより「generator_bを実行したときのyield値」を指定しています。
また、generator_bがreturnで出力した値は、yield fromの左辺（res）に渡されます。
従って、generator_a内の後続処理にてresを利用することができます。

yield fromはgenerator_aが行うべきループ処理をgenerator_bに任せていることから「委譲」とも呼ばれます。

最後に、本節で学んだことをまとめます：

• yieldを含む関数はジェネレーターになる
• ジェネレーターからはfor文を通してyield値を逐次取り出すことができる
• ジェネレーター内でreturnされた値はStopIteration例外のvalueとして取り出すことができる
• ジェネレーターはyield from句によって処理を委譲することができる

3. async/await + yield（非同期ジェネレーター）

さて、いよいよ大詰めです。
ここまで話したasync/awaitとyieldを組み合わせた際に何が起きるかについて説明します。
結論を言ってしまえば、これは第2節で説明した「ジェネレーター」の非同期版 = 非同期ジェネレーター を与えます（※）。
しかし、これらが組み合わさった際に生じる問題から、単純な仕様の足し算では済まなくなってしまいます。本節ではそのあたりを詳しく説明します。
※ 非同期関数のジェネレーター版と捉えることもできなくはないですが、ジェネレーターありきで考えた方がわかりやすいです。

async def print_table():
    for _ in range(5):
        lines = await fetch_table(n=200)
        yield show(lines)

# fetch_table, showは架空の関数。それぞれ、以下のような想定と考えてください：
# fetch_table: あるテーブルをn行ずつフェッチする関数
# show: テーブル情報を表示用に成形する関数

この例では、print_tableはasync関数として定義され（実際awaitも内部に含み）、かつyieldを内部に含んでいるため非同期ジェネレーターといえます。

ジェネレーターからはfor文でyield値を抽出することができました。
非同期ジェネレーターでも同様ですが、次の2つの対応が追加で必要となります：
① 非同期ジェネレーターからは、（for文ではなく）async for文によって値を抽出する
② async forを実行する関数自身は、async関数（or 非同期ジェネレーター）でなければならない

従って、非同期ジェネレーターprint_tableは、例えば以下のように実行されます：

import asyncio

async def main():
    async for msg in print_table():
        print(msg)

if __name__ == "__main__":
    asyncio.run(main())

この例では、main関数内でasync forにより非同期ジェネレーターから値を抽出し、
また、main関数自身をasync関数として定義してasyncio.runで実行しています。

一般に、非同期ジェネレーターは以下の要件に対して実装されることが多いです：
要件1：途中の実行結果を逐次出力する必要がある（yieldが必要）
要件2：要件1の処理か、あるいはそれ以外の箇所で時間のかかる処理があり、非同期化によるパフォーマンス改善が必要（async/awaitが必要）

例えば、LLMの小分けされた出力を逐次ユーザーに返却したり、SQLによりテーブルを数百行ずつに分けてフェッチするなどの際に用いられます。これらは「ジェネレーター」の節で示した例ではありますが、さらに非同期性を加えたい場合に「非同期ジェネレーター」が使用されます。

3-1. 非同期ジェネレーターの制約：returnで値を返せない

非同期ジェネレーターには、returnで値を返すことができないという著しい制約があります：

async def print_table():
    N = 5
    for _ in range(N):
        lines = await fetch_table(n=200)
        yield show(lines)
    return N

# 実行時、エラーが発生する
# SyntaxError: 'return' with value in async generator

returnとして返却する値を指定していない場合は、問題なく処理が実行されます：

async def print_table():
    N = 5
    for _ in range(N):
        lines = await fetch_table(n=200)
        yield show(lines)
    return
# Syntax/Runtimeともにエラーは発生しない

つまり、非同期ジェネレーターにおいてはreturnは処理を中断するための演算子としてのみ利用することができます。
なお、「なぜ非同期ジェネレーター内でreturnによる値返却が禁止されているか」はPython本体の設計・実装の文脈が絡み合うため、本稿では詳しく立ち入らないこととします。

3-2. return値を受け取れないって困るんじゃない？

とはいえ実際問題「逐次yieldしたい一方で、一連の処理による成果物も出力したい（returnしたい）」という場面も起こり得るかと思います。
ここでは、その場合の対処法を述べます：

方法1：非同期ジェネレーターの実装を回避する（推奨）

上述の要件が発生したとき、まず初めに考えるべきは「それって非同期ジェネレーターじゃないとダメ？」ということです：

• 逐次の値出力（yield）は本当に必要？最後に成果物をreturnするときにタイミングを合わせる必要があるなら、それまで値を逐一出力する必要はないのでは？
  （yieldがなくなれば単にasync関数として実装可能）
• 非同期性(async/await)は本当に必要？awaitしている処理が実はあまり時間がかからないなら、async/awaitをやめて単純なジェネレーターにしてもよいのでは？
• 非同期ジェネレーターと呼び出し元の関数の切り分けは適切？過剰な機能を非同期ジェネレーターに押し付けていない？例えば「出力結果を集計する機能」を呼び出し元の関数で実現するなどして「非同期ジェネレーターが最終成果物を返却するべき」という構造から脱却できない？

方法2：最後のyieldで最終成果物を出力する（非推奨）

例えば以下のように最終成果物としてreturnしたい箇所をyieldに置き換える方法が考えられますが、個人的にはあまり推奨できない方法です。

async def print_table():
    N = 5
    for _ in range(N):
        lines = await fetch_table(n=200)
        yield show(lines)
    yield N # 最後のyieldで最終成果物を出力する
    return

呼び出し側の関数で出力値が「ループ中の通常のyield」か「最終成果物」かを判別する必要が出てしまうという問題があります。
実装の工夫次第でこの問題を解消することもできますが、それをするぐらいなら方法1, 3に記載の手法を使う方がよほどきれいな実装になります。

方法3：return, yield以外の方法で最終成果物を出力する（若干非推奨）

例えばmutableな値を入力し、出力値の受け渡しのチャネルとして利用するという方法があります：

async def print_table(context: dict):
    N = 5
    for _ in range(N):
        lines = await fetch_table(n=200)
        yield show(lines)
    context["result"] = N

async def main():
    context = {}
    async for msg in print_table(context):
        print(msg)
    print(context["result"])

この例では辞書型という最もナイーブな方法を用いましたが、他にも様々な方法があります。
それ単体で1記事書けてしまうぐらい方法があるため、本稿ではこれ以上の言及は控えます。

さて、ここまで方法1～3を挙げましたが、結局どれを使うのが良いでしょうか：筆者としては、まずは方法1を検討することを推奨したいです。そもそもPythonは「非同期ジェネレーターを使う場合の多くは、逐次のyieldを出力できれば十分なはず」という設計思想でいるはずですので、可能な限りそれに従うべきです。そして、どうしてもその型に沿わない場合にのみ方法3を検討いただくのが良いと考えています。

3-3.（補遺）yield fromの禁止

第2節で、通常の（非同期でない）ジェネレーターの処理を他のジェネレーターに委譲する記法としてyield from句を導入しました。
一方で、async defで定義された関数の中ではyield fromを使えない（SyntaxErrorになる）という仕様があります。
つまり、非同期ジェネレーターの内部でyield fromを使うことはできません：

async def generator_a2():
    res = yield from generator_b()
    if res == "Finished!":
        print("Success")
# 実行時、SyntaxErrorが発生する：
# SyntaxError: 'yield from' inside async function

（非同期ジェネレーター内部でyieldは使ってよいが、yield fromはダメ）

最後に、本節で学んだことをまとめます：

• async defで定義され、かつyieldを含む関数は非同期ジェネレーターになる。
• 非同期ジェネレーターはasync for文を通してyield値を逐次取り出すことができる。
  また、その際async for文を実行する関数はasync関数（or 非同期ジェネレーター）でなければならない。
• 非同期ジェネレーターにおいては、returnは処理の中断を表す演算子としてしか使えない。
  特に、returnによって値を返却することはできない。

4. Step Up

本編はここまでで終わりですが、最後に本稿で敢えて触れなかったトピックを挙げたいと思います。興味のある方は調べてみてください：

• （asyncioなどのライブラリが）実際にasync関数をどのように実行しているか：awaitで待機するか/しないかの判断や各種非同期処理へのランタイムの配分をどのように行っているか
• （一般的な）iterableなオブジェクトや、非同期ジェネレーターをクラスとして定義する方法：本稿ではasync/await, yield演算子の使い方を主軸に、関数を非同期化・ジェネレーター化する方法を記載しましたが、実際にはクラス定義により同様のことを実現することが可能です。
  特にyieldに関する理解が深まるかと思いますので、興味のある方は当たってみてください
• yieldを使って外部からデータを注入する方法：本稿では yield を「for で値を取り出す」用途として説明しましたが、他にsend()/throw()/close() により外部から値や例外を注入する使い方もあります
• 非同期ジェネレーターからyield以外のチャネルで値を出力する方法：第3節3-2項の「方法3」に該当します。ただし、これは「方法1」がどうしても取れない場合の応急処置に過ぎないことに注意してください
• Pythonの「この仕様なんかヘンじゃない？」と思える箇所について、なぜそのような仕様になったのかの歴史的経緯

最後に

以上を通してasync/await, yieldに関するチュートリアルは終わりです。
長いことお付き合いいただきありがとうございました。
この記事が皆様の開発者ライフの一助になれば幸いです。