BatchPartsProcessor

BatchPartsProcessor 是一个输出 Processor，它会在流式传输期间将多个 stream part 批量合并，以降低发送频率。此 Processor 可减少网络开销、通过合并小型文本 chunk 改善用户体验，并通过控制何时向客户端发送 part 来优化流式传输性能。

用法示例

import { BatchPartsProcessor } from '@mastra/core/processors'

const processor = new BatchPartsProcessor({
  batchSize: 5,
  maxWaitTime: 100,
  emitOnNonText: true,
})

构造函数参数

options?:

Options

批量合并 stream part 的配置选项

Options

batchSize?:

number

发送前要批量合并的 part 数量

maxWaitTime?:

number

发送一个批次前的最长等待时间（以毫秒为单位）。如已设置，即使当前批次尚未达到 batchSize，也会发送

emitOnNonText?:

boolean

遇到非文本 part 时是否立即发送

返回值

id:

string

Processor 标识符，设为 'batch-parts'

name?:

string

可选的 Processor 显示名称

processOutputStream:

(args: { part: ChunkType; streamParts: ChunkType[]; state: Record<string, any>; abort: (reason?: string) => never }) => Promise<ChunkType | null>

处理流式输出 part，以将其批量合并

flush:

(state?: BatchPartsState) => ChunkType | null

在流结束时强制刷新所有剩余的批量 part

扩展用法示例

src/mastra/agents/batched-agent.ts

import { Agent } from '@mastra/core/agent'
import { BatchPartsProcessor } from '@mastra/core/processors'

export const agent = new Agent({
  id: 'batched-agent',
  name: 'batched-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5.6-sol',
  outputProcessors: [
    new BatchPartsProcessor({
      batchSize: 5,
      maxWaitTime: 100,
      emitOnNonText: true,
    }),
  ],
})

相关内容

• Guardrails