Python の yield を理解する — return との違いから、AI エージェントへのストリーミング入力パターンまで
はじめに
AI エージェントを構築していると、こんな疑問に出会いませんか？
「エージェントが動いている最中に、新しい情報を差し込みたい。でも、毎回セッションを作り直すのは非効率すぎる……」
この疑問を掘り下げていくと、Python の yield という仕組みに行き着きました。一見シンプルなキーワードですが、理解すると AI エージェントのアーキテクチャ設計に直結する重要な概念でした。この記事では、yield の基本から、エージェントへの「ストリーミング入力」パターンまでを整理します。
return と yield — 制御フローの根本的な違い
return は「即座に退場」
return は関数の最終出力を返し、その場で関数を終了します。関数内のローカル変数はすべて破棄され、次に呼び出すときは最初の行からやり直しです。
def get_greeting():
name = "Alice"
return f"Hello, {name}"
# ここで関数は完全に終了。name も破棄される
result = get_greeting() # 毎回最初から実行
yield は「一時停止して再開」
yield を使うと、関数はジェネレーターに変わります。値を返しつつ、関数の状態（変数、実行位置）をすべて保持したまま一時停止します。次に値を要求されると、停止した行から実行を再開します。
def count_up():
n = 1
while True:
yield n # n を返して一時停止
n += 1 # 再開時、ここから続く
counter = count_up()
print(next(counter)) # 1
print(next(counter)) # 2（n=1 の状態を覚えている）
print(next(counter)) # 3
比較まとめ
| 特性 | return |
yield |
|---|---|---|
| 関数の終了 | 即座に終了 | 一時停止（再開可能） |
| 状態の保持 | 破棄される | 保持される |
| 呼び出しモデル | 毎回最初から実行 | 前回の停止位置から再開 |
| 例えるなら | 手紙を送る（1通完結） | 電話をかける（回線を維持） |
エージェントへのストリーミング入力 — async iterator パターン
yield の「状態を保持したまま値を逐次的に返す」という特性は、長時間動作する AI エージェントへの入力パターンに直結します。
問題: 単発リクエストの限界
通常の API 呼び出しでは、リクエスト時点ですべての入力が揃っている必要があります。
# 従来: すべての入力を最初に渡す
response = await client.query(prompt="このデータを分析して")
しかし、長時間動くエージェントでは、実行途中で新しい情報を差し込みたい場面があります。毎回セッションを作り直すと、過去の会話履歴をすべて再送する必要があり、非効率です。
解決: 非同期ジェネレーターによるストリーミング入力
Python の async ジェネレーターをプロンプトとして渡すと、SDK はセッションを「オープン」な状態に保ち、後から届くメッセージを動的に流し込めます。
import asyncio
async def message_streamer():
# 最初の指示を送信
yield "エージェントを起動します。S3 のログを分析してください。"
# 外部キュー（SQS など）からイベントを待機して随時注入
while True:
event = await queue.get()
yield event.message
if event.is_last:
break
# SDK にイテレーターを渡す — セッションが維持される（疑似コード）
response = await client.query(prompt=message_streamer())
ポイントは、yield のおかげで関数が終了せず、キューにイベントが届くまで「待機状態」になることです。イベントが届けば再開し、新しいメッセージをエージェントに送り込みます。
streaming=True との違いに注意
多くの SDK で見かける streaming=True は、出力（レスポンス） をストリーミングするための設定です。一方、async iterator をプロンプトに渡すのは 入力 をストリーミングする仕組みです。方向が逆なので混同しないよう注意が必要です。
| 設定 | 方向 | 用途 |
|---|---|---|
streaming=True |
出力のストリーミング | レスポンスをトークン単位で逐次受信 |
| async iterator をプロンプトに渡す | 入力のストリーミング | 実行中のエージェントに動的にメッセージを注入 |
ユースケース — いつストリーミング入力が必要か
1. Human-in-the-Loop（HITL）
エージェントが「このファイルを削除してもよいですか？」と確認を求める場面で、セッションを維持したまま人間の応答を待てます。
async def hitl_streamer():
yield "本番データベースのクリーンアップを開始してください。"
# エージェントが確認を求めてきたら、UI から人間の回答を受け取る
approval = await wait_for_human_approval()
yield f"承認: {approval}"
エージェントを「ウォーム」な状態に保ったまま、人間の判断を挟めるのがメリットです。
2. リアルタイム外部イベントの注入
サポートボットが顧客の問題を分析している最中に、「顧客がプランをアップグレードした」というシステムイベントを差し込むケースです。
async def event_injector():
yield "顧客 #12345 のサポートチケットを分析してください。"
async for event in system_event_stream:
yield f"[SYSTEM EVENT: {event.description}]"
エージェントはイベントを受け取り、回答の方向性をリアルタイムに調整できます。
3. 段階的な指示の追加
100ページのドキュメントを分析中に「やっぱりこの観点も追加して」と指示を差し込むケースです。セッションを作り直す必要がないため、それまでの分析コンテキストが維持されます。
Node.js での実装 — async function*
Python だけでなく、Node.js でも同じパターンが使えます。構文は async function*（アスタリスク付き）と yield です。
// Node.js の非同期ジェネレーター
async function* messageStreamer() {
yield "Initializing agent...";
while (true) {
const event = await waitForNextEvent();
yield event.text;
if (event.type === 'close') break;
}
}
// SDK にイテレーターを渡す（疑似コード）
const client = new AgentClient();
await client.query({ prompt: messageStreamer() });
Python vs Node.js 比較
| 項目 | Python | Node.js |
|---|---|---|
| 構文 | async def func(): + yield |
async function* func() + yield |
| 消費方法 | async for item in gen: |
for await (const item of gen) |
| ジェネレーター判定 | yield の有無で自動判定 |
function* のアスタリスクで明示 |
| エコシステム | asyncio ベース | EventEmitter / Promise ベース |
Node.js では関数宣言に * を付ける必要がある点が Python と異なりますが、概念は同一です。どちらの言語でも「関数を終了させず、状態を保持したまま値を逐次的に送り出す」という動作は変わりません。
まとめ
yieldは関数を終了させずに値を返す。returnが「手紙」なら、yieldは「電話」。回線を維持したまま情報をやり取りできる- 長時間エージェントへのストリーミング入力に直結。 async iterator をプロンプトに渡すことで、実行中のエージェントに動的にメッセージを注入できる
- HITL、外部イベント注入、段階的指示追加など、実務で頻出するパターンを実現。 セッション再作成のオーバーヘッドを回避できる
- Python と Node.js で同じ概念が使える。 構文の差異はあるが、設計思想は共通
yield は単なる言語機能ではなく、AI エージェントの設計パターンを支える基盤です。長時間動作するエージェントを構築する際は、ストリーミング入力パターンを検討してみてください。
