> Discover all available pages from the documentation index: https://mastra.zisheng.pro/zh-HK/llms.txt

# 串流

Mastra 支援 agents 及 workflows 的即時增量回應，讓使用者在輸出產生時即可查看，而毋須等待完成。這適合聊天、長篇內容、多步驟 workflows，或任何重視即時回饋的情況。

## 開始使用

Mastra 的串流 API 會按模型版本調整：

- **`.stream()`**：適用於 V2 模型，支援 **AI SDK v5** 及更新版本（`LanguageModelV2`）。
- **`.streamLegacy()`**：適用於 V1 模型，支援 **AI SDK v4**（`LanguageModelV1`）。

## Agent 串流

基本 prompts 可傳入單一字串；提供多段上下文時可傳入字串陣列；如要精確控制 roles 及對話流程，則可傳入具有 `role` 及 `content` 的訊息物件陣列。

### 使用 `Agent.stream()`

`textStream` 會在回應產生時將其分成多個 chunks，讓輸出逐步串流，而非一次過抵達。使用 `for await` loop 疊代 `textStream`，即可檢查每個 stream chunk。

```typescript
const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}
```

詳情請參閱 [Agent.stream()](https://mastra.zisheng.pro/zh-HK/reference/streaming/agents/stream)。

> **提示:** 如 agent 會分派[背景任務](https://mastra.zisheng.pro/zh-HK/docs/long-running-agents/background-tasks)，請使用 [`Agent.streamUntilIdle()`](https://mastra.zisheng.pro/zh-HK/reference/streaming/agents/streamUntilIdle) 保持 stream 開啟，直至這些任務完成，而且 agent 已有機會回應其結果。

### `Agent.stream()` 的輸出

輸出會串流 agent 產生的回應。

```text
Of course!
To help you organize your day effectively, I need a bit more information.
Here are some questions to consider:
...
```

### Agent stream 屬性

Agent stream 可存取以下回應屬性：

- **`stream.textStream`**：發出文字 chunks 的可讀 stream。
- **`stream.text`**：resolve 為完整文字回應的 Promise。
- **`stream.finishReason`**：agent 停止串流的原因。
- **`stream.usage`**：Token 使用量資料。

### AI SDK v5+ 相容性

AI SDK v5（及更新版本）的模型 Providers 使用 `LanguageModelV2`。如錯誤訊息指出你正在使用 AI SDK v4 模型，便需要將模型套件升級至下一個 major version。

如要與 AI SDK v5+ 整合，請使用 `@mastra/ai-sdk` 的 `toAISdkV5Stream()` utility，將 Mastra streams 轉換成 AI SDK 相容格式：

```typescript
import { toAISdkV5Stream } from '@mastra/ai-sdk'

const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

// Convert to AI SDK v5+ compatible stream
const aiSDKStream = toAISdkV5Stream(stream, { from: 'agent' })
```

如要將訊息轉換成 AI SDK v5+ 格式，請使用 `@mastra/ai-sdk/ui` 的 `toAISdkV5Messages()` utility：

```typescript
import { toAISdkV5Messages } from '@mastra/ai-sdk/ui'

const messages = [{ role: 'user', content: 'Hello' }]
const aiSDKMessages = toAISdkV5Messages(messages)
```

## Workflow 串流

Workflow 串流會傳回描述執行生命週期的一連串結構化 events，而不是增量文字 chunks。使用 `.createRun()` 建立執行後，這種 event-based 格式可讓你即時追蹤及回應 workflow 進度。

### 使用 `Run.stream()`

`stream()` 方法會直接傳回 events 的 `ReadableStream`。

```typescript
const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}
```

詳情請參閱 [Run.stream()](https://mastra.zisheng.pro/zh-HK/reference/streaming/workflows/stream)。

### `Run.stream()` 的輸出

Event 結構在頂層包含 `runId` 及 `from`，無需深入 payload，亦可更容易識別及追蹤 workflow 執行。

```typescript
{
  type: 'workflow-start',
  runId: '1eeaf01a-d2bf-4e3f-8d1b-027795ccd3df',
  from: 'WORKFLOW',
  payload: {
    stepName: 'step-1',
    args: { value: 'initial data' },
    stepCallId: '8e15e618-be0e-4215-a5d6-08e58c152068',
    startedAt: 1755121710066,
    status: 'running'
  }
}
```

### Workflow stream 屬性

Workflow stream 可存取以下回應屬性：

- **`stream.status`**：workflow 執行狀態。
- **`stream.result`**：workflow 執行結果。
- **`stream.usage`**：workflow 執行的 token 總使用量。

Agents 或 workflows 的串流可即時顯示 LLM 輸出或 workflow 執行狀態。你可將此回饋直接傳給使用者，或在應用程式中於 workflow 狀態變更時顯示最新狀態。

Agents 或 workflows 發出的 events 代表產生及執行的不同階段，例如執行開始、產生文字或呼叫 Tool。

## Event 類型

以下是 `.stream()` 發出 events 的完整清單。 實際出現的 events 只會是其中一部分，取決於你串流的是 **agent** 還是 **workflow**：

- **start**：標示 agent 或 workflow 開始執行。
- **step-start**：表示 workflow step 已開始執行。
- **text-delta**：LLM 產生的增量文字 chunks。
- **tool-call**：Agent 決定使用 Tool 時發出，包含 Tool 名稱及引數。
- **tool-result**：Tool 執行傳回的結果。
- **step-finish**：確認特定 step 已完全結束，並可能包含該 step 結束原因等 metadata。
- **finish**：Agent 或 workflow 完成時發出，包含使用量統計資料。

## 檢查 agent streams

使用 `for await` loop 疊代 `stream`，檢查所有發出的 event chunks。

```typescript
const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

for await (const chunk of stream) {
  console.log(chunk)
}
```

詳情請參閱 [Agent.stream()](https://mastra.zisheng.pro/zh-HK/reference/streaming/agents/stream)。

### Agent 輸出範例

以下是可能發出的 events 範例。每個 event 必定包含 `type`，亦可包含 `from` 及 `payload` 等其他 fields。

```typescript
{
  type: 'start',
  from: 'AGENT',
  // ..
}
{
  type: 'step-start',
  from: 'AGENT',
  payload: {
    messageId: 'msg-cdUrkirvXw8A6oE4t5lzDuxi',
    // ...
  }
}
{
  type: 'tool-call',
  from: 'AGENT',
  payload: {
    toolCallId: 'call_jbhi3s1qvR6Aqt9axCfTBMsA',
    toolName: 'testTool'
    // ..
  }
}
```

## Writer API

Tools 及 workflow steps 共用 `writer` API。特定功能的範例請參閱 Tools 及 Workflows 文件。

## Agent 使用 Tool

Agent 串流可與 tool calls 結合，讓 Tool 輸出直接寫入 agent 的串流回應，並將 Tool 活動呈現為互動的一部分。

```typescript
import { Agent } from '@mastra/core/agent'
import { testTool } from '../tools/test-tool'

export const testAgent = new Agent({
  id: 'test-agent',
  name: 'Test Agent',
  instructions: 'You are a weather agent.',
  model: 'openai/gpt-5.6-sol',
  tools: { testTool },
})
```

### 使用 `context.writer`

`context.writer` 物件可在 Tool 的 `execute()` 函數中使用，並可向目前 stream 發出自訂 events、資料或值。Tools 使用這些 events，在執行期間提供中途結果或狀態更新。

> **注意:** 你必須對 `writer.write()` 的呼叫使用 `await`，否則會鎖定 stream，並出現 `WritableStream is locked` 錯誤。

```typescript
import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.write({
      type: 'custom-event',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.write({
      type: 'custom-event',
      status: 'success',
    })

    return {
      value: '',
    }
  },
})
```

你亦可使用 `writer.custom()` 發出頂層 stream chunks。與 UI 框架整合時，這個方法很有用。

```typescript
import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'success',
    })

    return {
      value: '',
    }
  },
})
```

### 暫時性資料 chunks

使用 `writer.custom()` 發出的 `data-*` chunks 預設會作為訊息記錄的一部分保存至儲存空間。如 chunks 只在即時串流期間需要，例如進度更新或詳細 log 輸出，請設定 `transient: true`，略過儲存持久化。暫時性 chunks 仍會即時串流至用戶端，但不會儲存至資料庫。

```typescript
await context?.writer?.custom({
  type: 'data-build-log',
  data: { line: 'Compiling module 3 of 12...' },
  transient: true,
})
```

如資料量大、頻率高，而且只與即時 session 有關，請使用暫時性 chunks。重新整理頁面後，暫時性 chunks 將不再可用。從儲存空間載入的只有 Tool 傳回值及所有非暫時性 chunks。

## 使用 `writer` 引數

`writer` 引數會傳入 workflow step 的 `execute` 函數，並可向目前 stream 發出自訂 events、資料或值。Workflow steps 使用這些 events，在執行期間提供中途結果或狀態更新。

> **注意:** 你必須對 `writer.write(...)` 的呼叫使用 `await`，否則會鎖定 stream，並出現 `WritableStream is locked` 錯誤。

```typescript
import { createStep } from "@mastra/core/workflows";

export const testStep = createStep({
  execute: async ({ inputData, writer }) => {
    const { value } = inputData;

    await writer?.write({
      type: "custom-event",
      status: "pending"
    });

    const response = await fetch(...);

    await writer?.write({
      type: "custom-event",
      status: "success"
    });

    return {
      value: ""
    };
  },
});
```