smoothStream()

smoothStream() 會建立實驗性的轉換串流，先緩衝文字與推理 delta，再以一致的區塊輸出。當模型輸出的 delta 大小不一時，可用它讓串流回應以更穩定的節奏呈現。

非文字區塊會原封不動地通過。在 Tool、控制或完成區塊之前，所有已緩衝的內容都會先輸出。

使用範例

將 Agent 的 fullStream 導入轉換串流：

src/stream-agent.ts

import { smoothStream } from '@mastra/core/stream'

const result = await agent.stream('Explain how rainbows form')

const stream = result.fullStream.pipeThrough(
  smoothStream({
    delayInMs: 20,
    chunking: 'word',
  }),
)

for await (const chunk of stream) {
  if (chunk.type === 'text-delta') {
    process.stdout.write(chunk.payload.text)
  }
}

此轉換只會變更導入的串流。原始 MastraModelOutput 上的 Promise 屬性與回呼（例如 result.text 和 onChunk）仍會保留模型原本的區塊輸出時序。

AI SDK 路由

從 @mastra/ai-sdk 匯入 smoothStream()，即可在 handleChatStream() 將 Agent 輸出轉換為 AI SDK UI 區塊前，使輸出更平順：

app/api/chat/route.ts

import { handleChatStream, smoothStream } from '@mastra/ai-sdk'
import { createUIMessageStreamResponse } from 'ai'
import { mastra } from '@/src/mastra'

export async function POST(req: Request) {
  const params = await req.json()
  const stream = await handleChatStream({
    mastra,
    agentId: 'weatherAgent',
    params,
    experimentalTransform: smoothStream({
      delayInMs: 20,
      chunking: 'word',
    }),
  })

  return createUIMessageStreamResponse({ stream })
}

@mastra/ai-sdk 匯出的函式會回傳可重複使用的轉換工廠函式，因此路由設定會為每個請求建立新的 TransformStream。@mastra/core/stream 匯出的函式則會回傳 TransformStream，可直接搭配 pipeThrough() 使用。

這個可重複使用的工廠函式也能傳入 Agent.stream()：

import { smoothStream } from '@mastra/ai-sdk'

const result = await agent.stream('Explain how rainbows form', {
  experimentalTransform: smoothStream({ delayInMs: 20 }),
})

for await (const chunk of result.fullStream) {
  // Consume the transformed Mastra chunks.
}

參數

options?:

SmoothStreamOptions

控制轉換後串流的延遲與區塊邊界。

SmoothStreamOptions

delayInMs?:

number | null

每個區塊輸出後的延遲時間，單位為毫秒。將此值設為 null 可停用延遲。

chunking?:

'word' | 'line' | RegExp | SmoothStreamChunkDetector | Intl.Segmenter

控制如何將已緩衝的文字與推理內容分割成區塊。

chunking 選項接受下列值：

• 'word'：輸出完整單字，包括其後的空白字元。
• 'line'：輸出直到每個換行字元為止的內容。
• RegExp：輸出直到第一個相符項目為止的內容。
• Intl.Segmenter：使用可感知地區設定的分段方式，適合單字之間沒有空格的語言。
• SmoothStreamChunkDetector：以目前的緩衝區呼叫函式。函式會回傳要輸出的非空前綴；若要等待更多內容，則回傳 null 或 undefined。

自訂區塊分割方式

使用規則運算式定義區塊邊界：

const stream = result.fullStream.pipeThrough(
  smoothStream({
    chunking: /[^,]*,\s*/,
  }),
)

使用 Intl.Segmenter 進行可感知地區設定的分段：

const stream = result.fullStream.pipeThrough(
  smoothStream({
    chunking: new Intl.Segmenter('ja', { granularity: 'word' }),
  }),
)

當區塊邊界取決於自訂邏輯時，請使用偵測函式。回傳值必須是緩衝區的前綴：

const stream = result.fullStream.pipeThrough(
  smoothStream({
    chunking: buffer => {
      const boundary = buffer.indexOf('. ')
      return boundary === -1 ? null : buffer.slice(0, boundary + 2)
    },
  }),
)

回傳值

TransformStream<ChunkType<OUTPUT>, ChunkType<OUTPUT>>

此轉換會輸出經平順化處理的 text-delta 與 reasoning-delta 區塊，並保留區塊識別碼、執行識別碼、來源與中繼資料。

相關內容

• MastraModelOutput
• ChunkType