From 4ba49a243aaaa5a14e93f41e164a3bcc1907b133 Mon Sep 17 00:00:00 2001 From: Neko Ayaka Date: Tue, 31 Mar 2026 15:51:24 +0800 Subject: [PATCH] fix(pipelines-audio): no longer concurrently tts --- packages/pipelines-audio/package.json | 4 +- .../src/speech-pipeline.test.ts | 190 ++++++++++++++++++ .../pipelines-audio/src/speech-pipeline.ts | 144 ++++++++----- packages/pipelines-audio/src/types.ts | 3 + packages/pipelines-audio/vitest.config.ts | 8 + 5 files changed, 302 insertions(+), 47 deletions(-) create mode 100644 packages/pipelines-audio/src/speech-pipeline.test.ts create mode 100644 packages/pipelines-audio/vitest.config.ts diff --git a/packages/pipelines-audio/package.json b/packages/pipelines-audio/package.json index 514bb48c8..88ab4990a 100644 --- a/packages/pipelines-audio/package.json +++ b/packages/pipelines-audio/package.json @@ -30,7 +30,9 @@ ], "scripts": { "build": "tsdown", - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "test": "vitest", + "test:run": "vitest run --config vitest.config.ts" }, "dependencies": { "@moeru/eventa": "catalog:", diff --git a/packages/pipelines-audio/src/speech-pipeline.test.ts b/packages/pipelines-audio/src/speech-pipeline.test.ts new file mode 100644 index 000000000..c54218c63 --- /dev/null +++ b/packages/pipelines-audio/src/speech-pipeline.test.ts @@ -0,0 +1,190 @@ +import type { PlaybackItem, TextSegment, TextToken, TtsRequest } from './types' + +import { describe, expect, it, vi } from 'vitest' + +import { createSpeechPipeline } from './speech-pipeline' + +function delay(ms: number) { + return new Promise(resolve => setTimeout(resolve, ms)) +} + +function deferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise + reject = rejectPromise + }) + + return { + promise, + resolve, + reject, + } +} + +function createSegmenter(texts: string[]) { + return (_tokens: ReadableStream, meta: { streamId: string, intentId: string }) => { + let index = 0 + + return new ReadableStream({ + pull(controller) { + const text = texts[index] + if (text == null) { + controller.close() + return + } + + controller.enqueue({ + streamId: meta.streamId, + intentId: meta.intentId, + segmentId: `${meta.streamId}:${index}`, + text, + special: null, + reason: 'flush', + createdAt: Date.now(), + }) + index += 1 + }, + }) + } +} + +function createPlaybackSpy() { + const scheduled: Array> = [] + + return { + scheduled, + playback: { + schedule(item: PlaybackItem) { + scheduled.push(item) + }, + stopAll: vi.fn(), + stopByIntent: vi.fn(), + stopByOwner: vi.fn(), + onStart: vi.fn(), + onEnd: vi.fn(), + onInterrupt: vi.fn(), + onReject: vi.fn(), + }, + } +} + +describe('createSpeechPipeline', () => { + it('preserves playback order when TTS completes out of order', async () => { + const { scheduled, playback } = createPlaybackSpy() + + const pipeline = createSpeechPipeline({ + ttsMaxConcurrent: 2, + segmenter: createSegmenter(['first', 'second', 'third']), + playback, + async tts(request) { + if (request.sequence === 0) + await delay(30) + else if (request.sequence === 1) + await delay(5) + + return request.text + }, + }) + + const intentFinished = new Promise((resolve) => { + pipeline.on('onIntentEnd', () => resolve()) + }) + + const intent = pipeline.openIntent() + intent.end() + + await intentFinished + + expect(scheduled.map(item => item.sequence)).toEqual([0, 1, 2]) + expect(scheduled.map(item => item.text)).toEqual(['first', 'second', 'third']) + }) + + it('prefetches TTS requests up to the configured concurrency', async () => { + const { playback } = createPlaybackSpy() + const startedRequests: number[] = [] + const pendingRequests = [ + deferred(), + deferred(), + deferred(), + ] + let inFlight = 0 + let maxInFlight = 0 + + const pipeline = createSpeechPipeline({ + ttsMaxConcurrent: 2, + segmenter: createSegmenter(['alpha', 'beta', 'gamma']), + playback, + async tts(request: TtsRequest) { + startedRequests.push(request.sequence) + inFlight += 1 + maxInFlight = Math.max(maxInFlight, inFlight) + + try { + return await pendingRequests[request.sequence]!.promise + } + finally { + inFlight -= 1 + } + }, + }) + + const intentFinished = new Promise((resolve) => { + pipeline.on('onIntentEnd', () => resolve()) + }) + + const intent = pipeline.openIntent() + intent.end() + + await delay(0) + + expect(startedRequests).toEqual([0, 1]) + expect(maxInFlight).toBe(2) + + pendingRequests[0]!.resolve('alpha') + pendingRequests[1]!.resolve('beta') + await delay(0) + + expect(startedRequests).toEqual([0, 1, 2]) + + pendingRequests[2]!.resolve('gamma') + await intentFinished + + expect(maxInFlight).toBe(2) + }) + + it('cancels in-flight TTS work without scheduling stale playback', async () => { + const { scheduled, playback } = createPlaybackSpy() + const abortedRequests: number[] = [] + + const pipeline = createSpeechPipeline({ + ttsMaxConcurrent: 2, + segmenter: createSegmenter(['left', 'right']), + playback, + tts(request, signal) { + return new Promise((resolve) => { + signal.addEventListener('abort', () => { + abortedRequests.push(request.sequence) + resolve(null) + }, { once: true }) + }) + }, + }) + + const intentCanceled = new Promise((resolve) => { + pipeline.on('onIntentCancel', () => resolve()) + }) + + const intent = pipeline.openIntent() + intent.end() + + await delay(0) + intent.cancel('test-cancel') + await intentCanceled + + expect(abortedRequests.sort()).toEqual([0, 1]) + expect(scheduled).toEqual([]) + }) +}) diff --git a/packages/pipelines-audio/src/speech-pipeline.ts b/packages/pipelines-audio/src/speech-pipeline.ts index 9e162e584..0627be8a7 100644 --- a/packages/pipelines-audio/src/speech-pipeline.ts +++ b/packages/pipelines-audio/src/speech-pipeline.ts @@ -22,6 +22,12 @@ import { createPushStream } from './stream' export interface SpeechPipelineOptions { tts: (request: TtsRequest, signal: AbortSignal) => Promise + /** + * Maximum number of concurrent TTS generation tasks. Default is 4. Must be at least 1. + * + * @default 4 + */ + ttsMaxConcurrent?: number playback: { schedule: (item: PlaybackItem) => void stopAll: (reason: string) => void @@ -58,6 +64,7 @@ export function createSpeechPipeline(options: SpeechPipelineOptions() @@ -86,11 +93,91 @@ export function createSpeechPipeline(options: SpeechPipelineOptions | null>() + const inFlightTasks = new Set>() + let nextRequestSequence = 0 + let nextSequenceToSchedule = 0 + + function scheduleCompletedRequests() { + while (completedRequests.has(nextSequenceToSchedule)) { + const completedRequest = completedRequests.get(nextSequenceToSchedule) ?? null + completedRequests.delete(nextSequenceToSchedule) + + if (completedRequest) { + options.playback.schedule({ + id: createId('playback'), + streamId: completedRequest.streamId, + intentId: completedRequest.intentId, + segmentId: completedRequest.segmentId, + sequence: completedRequest.sequence, + ownerId: intent.ownerId, + priority: intent.priority, + text: completedRequest.text, + special: completedRequest.special, + audio: completedRequest.audio, + createdAt: Date.now(), + }) + } + + nextSequenceToSchedule += 1 + } + } + + function createTtsTask(request: TtsRequest) { + const task = (async () => { + let audio: TAudio | null = null + try { + audio = await options.tts(request, intent.controller.signal) + } + catch (err) { + logger.warn('TTS generation failed:', err) + if (intent.controller.signal.aborted) + return + } + + if (intent.controller.signal.aborted) { + completedRequests.set(request.sequence, null) + scheduleCompletedRequests() + return + } + + if (!audio) { + completedRequests.set(request.sequence, null) + scheduleCompletedRequests() + return + } + + const ttsResult: TtsResult = { + streamId: request.streamId, + intentId: request.intentId, + segmentId: request.segmentId, + sequence: request.sequence, + text: request.text, + special: request.special, + audio, + createdAt: Date.now(), + } + + context.emit(speechPipelineEventMap.onTtsResult, ttsResult) + completedRequests.set(request.sequence, ttsResult) + scheduleCompletedRequests() + })() + .finally(() => { + inFlightTasks.delete(task) + }) + + inFlightTasks.add(task) + return task + } try { const reader = segmentStream.getReader() while (true) { + while (!intent.controller.signal.aborted && inFlightTasks.size >= ttsMaxConcurrent) { + await Promise.race(inFlightTasks) + } + const { value, done } = await reader.read() if (done) break @@ -112,6 +199,7 @@ export function createSpeechPipeline(options: SpeechPipelineOptions(options: SpeechPipelineOptions = { - streamId: request.streamId, - intentId: request.intentId, - segmentId: request.segmentId, - text: request.text, - special: request.special, - audio, - createdAt: Date.now(), - } - - context.emit(speechPipelineEventMap.onTtsResult, ttsResult) - - options.playback.schedule({ - id: createId('playback'), - streamId: ttsResult.streamId, - intentId: ttsResult.intentId, - segmentId: ttsResult.segmentId, - ownerId: intent.ownerId, - priority: intent.priority, - text: ttsResult.text, - special: ttsResult.special, - audio: ttsResult.audio, - createdAt: Date.now(), - }) + createTtsTask(request) } + await Promise.allSettled(inFlightTasks) + scheduleCompletedRequests() reader.releaseLock() } catch (err) { @@ -177,11 +226,14 @@ export function createSpeechPipeline(options: SpeechPipelineOptions { streamId: string intentId: string segmentId: string + sequence: number text: string special: string | null audio: TAudio @@ -48,6 +50,7 @@ export interface PlaybackItem { streamId: string intentId: string segmentId: string + sequence: number ownerId?: string priority: number text: string diff --git a/packages/pipelines-audio/vitest.config.ts b/packages/pipelines-audio/vitest.config.ts new file mode 100644 index 000000000..ceafc2412 --- /dev/null +++ b/packages/pipelines-audio/vitest.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from 'vitest/config' + +export default defineConfig({ + test: { + environment: 'node', + include: ['src/**/*.test.ts'], + }, +})