# 用 Python 的 yield 為執行中的 AI 智能體串流輸入 — 與 return 的差異，以及 Node.js 的實作

> Python 的 yield 與 return 有何不同，以及如何把非同步產生器當作提示詞傳入，為執行中的長時間 AI 智能體串流輸入，並附上 Node.js 的對應寫法。

- Source: https://oharu121.com/zh-tw/blog/python-yield-async-generator-agent-streaming-input/
- Published: 2026-07-12T12:00:00+09:00
- Tags: Python, Node.js, AI 智能體

---
## 引言

長時間執行的 AI 智能體有時需要在執行過程中注入新的資訊（人類的核准、外部事件、後續指示），而不必重新建立工作階段並重送整段歷史紀錄。Python 的 `yield` 正是這個模式背後的機制：一個簡單到容易被忽略的關鍵字，卻直接連結到智能體的架構設計。**這個模式是把一個非同步產生器當作輸入串流、保持開啟：`yield` 會暫停函式並保留其狀態，讓智能體的工作階段能維持在「熱」的狀態，同時一次接收一則新訊息。** 本文從 `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` |
|------|----------|---------|
| 函式結束 | 立即結束 | 暫停（可恢復） |
| 狀態保留 | 捨棄 | 保留 |
| 呼叫模式 | 每次都從頭執行 | 從上次暫停處恢復 |
| 類比 | 寄一封信（一次性） | 打一通電話（保持線路開啟） |

## 為智能體串流輸入 — 非同步迭代器模式

`yield` 那種「一邊保留狀態一邊逐一回傳值」的特性，直接連結到長時間執行的 AI 智能體的輸入模式。

### 問題：單次請求的限制

在一般的 API 呼叫中，所有輸入都必須在請求時就準備好。

```python
# 傳統做法：一次把所有輸入都傳入
response = await client.query(prompt="Analyse this data")
```

但在長時間執行的智能體中，有些情況需要在執行過程中注入新資訊。每次重新建立工作階段都得重送整段過去的對話歷史，效率不佳。

### 解法：用非同步產生器串流輸入

**把 Python 的 `async` 產生器當作提示詞傳入，能讓工作階段保持「開啟」的狀態，** 動態餵入之後才抵達的訊息。

```python
import asyncio

async def message_streamer():
    # 傳送初始指示
    yield "Starting the agent. Please analyse the S3 logs."

    # 等待外部佇列（例如 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 進入點只是示意。接收訊息非同步可迭代物件的方法名稱，各 SDK 不盡相同（`query`、`run`、`chat` 等），而且這些 API 演進得很快，需要查閱所使用 SDK 目前的簽章。各家 SDK 之間穩定不變的是這個模式本身：把一個非同步產生器當作輸入串流保持開啟。

*Figure — Dataflow: 輸入流向執行中的智能體；輸出則從中流回。*

### 注意與 streaming=True 的差異

許多 SDK 中常見的 `streaming=True`，是用來串流**輸出（回應）**的設定。相對地，把非同步迭代器當作提示詞傳入，則是串流**輸入**的機制。方向相反，需留意不要混淆。

| 設定 | 方向 | 用途 |
|------|------|------|
| `streaming=True` | 輸出串流 | 逐一 token 接收回應 |
| 把非同步迭代器當作提示詞傳入 | 輸入串流 | 動態把訊息注入執行中的智能體 |

## 使用情境 — 何時需要串流輸入

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

當智能體詢問確認，例如「可以刪除這個檔案嗎？」，工作階段能在保持存活的同時等待人類的回應。

```python
async def hitl_streamer():
    yield "Please start cleaning up the production database."

    # 智能體要求確認時，從 UI 接收人類的回答
    approval = await wait_for_human_approval()
    yield f"Approval: {approval}"
```

好處在於能一邊讓智能體保持「熱」的狀態，一邊插入人類的判斷。

### 2. 即時注入外部事件

支援機器人正在分析同一位顧客的問題時，注入「顧客升級了方案」這類系統事件的情境。

```python
async def event_injector():
    yield "Please analyse the support ticket for customer #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 與 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` 就是一通保持線路開啟、可以交換資訊的「電話」。
- **它直接連結到長時間執行智能體的串流輸入。** 把非同步迭代器當作提示詞傳入，能把訊息動態注入執行中的智能體。
- **它實現了實務上常見的模式，包括 HITL、外部事件注入、逐步指示。** 每一種都能避免重新建立工作階段的額外開銷。
- **同樣的概念在 Python 與 Node.js 中都適用。** 語法有差異，但設計理念是共通的。

`yield` 不只是一項語言特性，更是 AI 智能體設計模式的基礎，而在長時間執行的智能體需要它的場合，這個串流輸入模式值得採用。