# Python の yield で動作中の AI エージェントに入力をストリーミングする — return との違いと Node.js での実装

> Python の yield が return とどう違うか、そして async ジェネレーターをプロンプトとして渡すことで長時間動作する AI エージェントに入力をストリーミングする方法を、Node.js の等価な書き方とともに解説します。

- Source: https://oharu121.com/ja/blog/python-yield-async-generator-agent-streaming-input/
- Published: 2026-07-12T12:00:00+09:00
- Tags: Python, Node.js, AIエージェント

---
## はじめに

長時間動作する AI エージェントでは、セッションを作り直して会話履歴全体を再送することなく、実行の途中で新しい情報（人間の承認、外部イベント、追加の指示）を差し込みたい場面があります。その仕組みの正体は Python の `yield` でした。見過ごしてしまいそうなほどシンプルなキーワードですが、AI エージェントのアーキテクチャに直結しています。**このパターンは、async ジェネレーターを入力ストリームとして開いたままにする、というものです。`yield` は状態を保持したまま関数を一時停止させるので、新しいメッセージが1つずつ届く間もエージェントのセッションを「ウォーム」に保てます。** この記事では、`yield` を `return` との違いから、このストリーミング入力パターン、そして Node.js での同じ実装まで整理します。

## return と yield — 制御フローの根本的な違い

### return は「即座に退場」

**`return` はその場で関数を終了させます。** 関数内のローカル変数はすべて破棄され、次に呼び出すときは最初の行からやり直しです。

```python
def get_greeting():
    name = "Alice"
    return f"Hello, {name}"
    # ここで関数は完全に終了。name も破棄される

result = get_greeting()  # 毎回最初から実行
```

### yield は「一時停止して再開」

`yield` を使うと、関数はジェネレーターに変わります。**値を返しながら一時停止し、関数の状態（変数、実行位置）はすべて保持されます。** 次に値を要求されると、停止した行から実行を再開します。

```python
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
```

*Figure — ReturnVsYield: `return` は毎回ゼロから、`yield` は止まった続きから再開します。*

### 比較まとめ

| 特性 | `return` | `yield` |
|------|----------|---------|
| 関数の終了 | 即座に終了 | 一時停止（再開可能） |
| 状態の保持 | 破棄される | 保持される |
| 呼び出しモデル | 毎回最初から実行 | 前回の停止位置から再開 |
| 例えるなら | 手紙を送る（1通完結） | 電話をかける（回線を維持） |

## エージェントへのストリーミング入力 — async iterator パターン

`yield` の「状態を保持したまま値を逐次的に返す」という特性は、長時間動作する AI エージェントへの入力パターンに直結します。

### 問題: 単発リクエストの限界

通常の API 呼び出しでは、リクエスト時点ですべての入力が揃っている必要があります。

```python
# 従来: すべての入力を最初に渡す
response = await client.query(prompt="このデータを分析して")
```

しかし、長時間動くエージェントでは、実行途中で新しい情報を差し込みたい場面があります。毎回セッションを作り直すと、過去の会話履歴をすべて再送する必要があり、非効率です。

### 解決: 非同期ジェネレーターによるストリーミング入力

**Python の `async` ジェネレーターをプロンプトとして渡すと、セッションが「オープン」な状態に保たれます。** 後から届くメッセージを動的に流し込めます。

```python
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` のおかげで関数が終了せず、キューにイベントが届くまで「待機状態」になることです。イベントが届けば再開し、新しいメッセージをエージェントに送り込みます。

ここで示している SDK の入口は例示です。メッセージの async イテラブルを受け取るメソッド名は SDK ごとに異なり（`query`、`run`、`chat` など）、これらの API は急速に進化しているため、実際に使う SDK の最新のシグネチャを確認してください。どの SDK でも共通して安定しているのはパターンの方です——async ジェネレーターを入力ストリームとしてオープンに保つ、という設計です。

*Figure — Dataflow: 入力は実行中のエージェントへ、出力はそこから戻ってきます。*

### streaming=True との違いに注意

多くの SDK で見かける `streaming=True` は、**出力（レスポンス）** をストリーミングするための設定です。一方、async iterator をプロンプトに渡すのは **入力** をストリーミングする仕組みです。方向が逆なので混同しないよう注意が必要です。

| 設定 | 方向 | 用途 |
|------|------|------|
| `streaming=True` | 出力のストリーミング | レスポンスをトークン単位で逐次受信 |
| async iterator をプロンプトに渡す | 入力のストリーミング | 実行中のエージェントに動的にメッセージを注入 |

## ユースケース — いつストリーミング入力が必要か

### 1. Human-in-the-Loop（HITL）

エージェントが「このファイルを削除してもよいですか？」と確認を求める場面で、セッションを維持したまま人間の応答を待てます。

```python
async def hitl_streamer():
    yield "本番データベースのクリーンアップを開始してください。"

    # エージェントが確認を求めてきたら、UI から人間の回答を受け取る
    approval = await wait_for_human_approval()
    yield f"承認: {approval}"
```

エージェントを「ウォーム」な状態に保ったまま、人間の判断を挟めるのがメリットです。

### 2. リアルタイム外部イベントの注入

サポートボットが顧客の問題を分析している最中に、「顧客がプランをアップグレードした」というシステムイベントを差し込むケースです。

```python
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` です。

```javascript
// 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 エージェントの設計パターンを支える基盤です。長時間動作するエージェントが必要とする場面では、このストリーミング入力パターンを検討する価値があります。