smoothStream()

smoothStream() crée un flux de transformation expérimental qui met en mémoire tampon les deltas de texte et de raisonnement avant de les émettre sous forme de segments réguliers. Utilisez-le pour rendre le rythme des réponses diffusées plus constant lorsqu’un modèle émet des deltas irréguliers.

Les segments qui ne contiennent pas de texte sont transmis sans modification. Tout contenu mis en mémoire tampon est émis avant un segment de Tool, de contrôle ou de fin.

Exemple d’utilisation

Transmettez le fullStream d’un Agent au flux de transformation :

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

La transformation modifie uniquement le flux qui lui est transmis. Les propriétés de type Promise et les callbacks du MastraModelOutput d’origine, tels que result.text et onChunk, conservent la cadence initiale des segments du modèle.

Routes AI SDK

Importez smoothStream() depuis @mastra/ai-sdk afin de régulariser la sortie d’un Agent avant que handleChatStream() ne la convertisse en segments 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 })
}

L’export de @mastra/ai-sdk renvoie une fabrique de transformations réutilisable, de sorte que la configuration de la route crée un nouveau TransformStream pour chaque requête. L’export de @mastra/core/stream renvoie un TransformStream utilisable directement avec pipeThrough().

La fabrique réutilisable peut également être transmise à 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.
}

Paramètres

options?:

SmoothStreamOptions

Contrôle le délai et les limites des segments du flux transformé.

SmoothStreamOptions

delayInMs?:

number | null

Délai en millisecondes après chaque segment émis. Définissez cette valeur sur null pour désactiver le délai.

chunking?:

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

Contrôle la manière dont le texte et le raisonnement mis en mémoire tampon sont divisés en segments.

L’option chunking accepte les valeurs suivantes :

• 'word' : émet des mots complets, y compris les espaces qui les suivent.
• 'line' : émet le contenu jusqu’à chaque saut de ligne inclus.
• RegExp : émet le contenu jusqu’à la première correspondance incluse.
• Intl.Segmenter : utilise une segmentation adaptée aux paramètres régionaux, utile pour les langues qui ne séparent pas les mots par des espaces.
• SmoothStreamChunkDetector : appelle une fonction avec le tampon actuel. La fonction renvoie un préfixe non vide à émettre, ou null ou undefined pour attendre davantage de contenu.

Segmentation personnalisée

Utilisez une expression régulière pour définir une limite de segment :

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

Utilisez Intl.Segmenter pour une segmentation adaptée aux paramètres régionaux :

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

Utilisez une fonction de détection lorsque les limites des segments dépendent d’une logique personnalisée. La valeur renvoyée doit être un préfixe du tampon :

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

Valeur renvoyée

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

La transformation émet des segments text-delta et reasoning-delta régularisés. Elle conserve les identifiants des segments et des exécutions, les sources ainsi que les métadonnées.

Voir aussi

• MastraModelOutput
• ChunkType