跳至主要內容

smoothStream()

smoothStream() 會建立實驗性 transform stream，在以一致的 chunk 發出文字及推理 delta 前先加以緩衝。當模型發出的 delta 不均勻時，可用此函數令串流回應以較穩定的速度顯示。

非文字 chunk 會原樣傳遞。任何已緩衝內容都會在 Tool、控制或完成 chunk 前發出。

使用範例
使用範例 的直接連結

將 Agent 的 fullStream 經由 transform 傳送：

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)
  }
}

transform 只會改變經管道傳送的串流。原有 MastraModelOutput 上的 promise 屬性及 callback（例如 result.text 和 onChunk）會保留模型原本的 chunk 時序。

AI SDK route
AI SDK route 的直接連結

從 @mastra/ai-sdk 匯入 smoothStream()，在 handleChatStream() 將 Agent 輸出轉換為 AI SDK UI chunk 前平滑處理：

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 的 export 會傳回可重用的 transform factory，讓 route 配置為每個請求建立新的 TransformStream。@mastra/core/stream 的 export 則會傳回可直接配合 pipeThrough() 使用的 TransformStream。

可重用的 factory 亦可傳入 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
控制轉換後串流的延遲及 chunk 邊界。
SmoothStreamOptions

delayInMs?:

number | null
每個 chunk 發出後的延遲毫秒數。將此值設為 null 可停用延遲。

chunking?:

'word' | 'line' | RegExp | SmoothStreamChunkDetector | Intl.Segmenter
控制如何將已緩衝的文字及推理內容分成 chunk。

chunking 選項接受：

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

自訂 chunk 分段
自訂 chunk 分段 的直接連結

使用 regular expression 定義 chunk 邊界：

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

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

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

當 chunk 邊界取決於自訂邏輯時，使用 detector 函數。傳回值必須是 buffer 的前綴：

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

transform 會發出經平滑處理的 text-delta 及 reasoning-delta chunk，並保留 chunk 標識符、run 標識符、來源及 metadata。

On this page