From 39bcc158bb0bff1e22b469950681f53db5b0fd87 Mon Sep 17 00:00:00 2001 From: jialin Date: Wed, 4 Mar 2026 15:17:36 +0800 Subject: [PATCH] feat: add stream mode for tts --- src/pages/playground/speech/hooks/index.ts | 2 + .../speech/hooks/useNonStreamTTS.ts | 81 +++++ .../playground/speech/hooks/useStreamTTS.ts | 281 ++++++++++++++++++ src/pages/playground/speech/tts.tsx | 121 +++++--- 4 files changed, 442 insertions(+), 43 deletions(-) create mode 100644 src/pages/playground/speech/hooks/index.ts create mode 100644 src/pages/playground/speech/hooks/useNonStreamTTS.ts create mode 100644 src/pages/playground/speech/hooks/useStreamTTS.ts diff --git a/src/pages/playground/speech/hooks/index.ts b/src/pages/playground/speech/hooks/index.ts new file mode 100644 index 00000000..361c5ec9 --- /dev/null +++ b/src/pages/playground/speech/hooks/index.ts @@ -0,0 +1,2 @@ +export { useNonStreamTTS } from './useNonStreamTTS'; +export { useStreamTTS } from './useStreamTTS'; diff --git a/src/pages/playground/speech/hooks/useNonStreamTTS.ts b/src/pages/playground/speech/hooks/useNonStreamTTS.ts new file mode 100644 index 00000000..1ce8a749 --- /dev/null +++ b/src/pages/playground/speech/hooks/useNonStreamTTS.ts @@ -0,0 +1,81 @@ +import { useCallback, useRef, useState } from 'react'; +import { textToSpeech } from '../../apis'; +import { extractErrorMessage } from '../../config'; + +interface UseNonStreamTTSParams { + onSuccess?: (result: { url: string; type: string }) => void; + onError?: (error: any) => void; +} + +interface TTSParams { + model: string; + voice: string; + response_format: string; + speed?: number; + input: string; + [key: string]: any; +} + +export const useNonStreamTTS = (params?: UseNonStreamTTSParams) => { + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + const controllerRef = useRef(null); + + const generate = useCallback( + async (ttsParams: TTSParams) => { + try { + setLoading(true); + setError(null); + + // Abort previous request if exists + controllerRef.current?.abort(); + controllerRef.current = new AbortController(); + const signal = controllerRef.current.signal; + + const res: any = await textToSpeech({ + data: ttsParams, + signal + }); + + if ((res?.status_code && res?.status_code !== 200) || res?.error) { + const errorMessage = extractErrorMessage(res); + setError({ + error: true, + errorMessage + }); + params?.onError?.(errorMessage); + return null; + } + + params?.onSuccess?.(res); + return res; + } catch (err: any) { + const res = err?.response?.data; + if (res?.error) { + const errorMessage = extractErrorMessage(res); + setError({ + error: true, + errorMessage + }); + params?.onError?.(errorMessage); + } + return null; + } finally { + setLoading(false); + } + }, + [params] + ); + + const abort = useCallback(() => { + controllerRef.current?.abort(); + setLoading(false); + }, []); + + return { + generate, + abort, + loading, + error + }; +}; diff --git a/src/pages/playground/speech/hooks/useStreamTTS.ts b/src/pages/playground/speech/hooks/useStreamTTS.ts new file mode 100644 index 00000000..1ad1d0ee --- /dev/null +++ b/src/pages/playground/speech/hooks/useStreamTTS.ts @@ -0,0 +1,281 @@ +import { fetchChunkedData } from '@/utils/fetch-chunk-data'; +import { useCallback, useRef, useState } from 'react'; +import { AUDIO_TEXT_TO_SPEECH_API } from '../../apis'; +import { extractErrorMessage } from '../../config'; + +interface UseStreamTTSParams { + onChunk?: (chunk: ArrayBuffer) => void; + onComplete?: (audioUrl: string) => void; // Return complete audio URL when done + onError?: (error: any) => void; + autoPlay?: boolean; +} + +interface TTSParams { + model: string; + voice: string; + response_format: string; + speed?: number; + input: string; + stream?: boolean; + [key: string]: any; +} + +/** + * Audio chunk queue manager for smooth playback + * Uses a single Audio element for better performance and seamless playback + */ +class AudioQueue { + private queue: Blob[] = []; + private audioElement: HTMLAudioElement; + private isPlaying = false; + private currentIndex = 0; + private onComplete?: () => void; + private streamEnded = false; + private minBufferSize = 2; // Minimum chunks to buffer before starting playback + private maxQueueSize = 10; // Maximum queue size to prevent memory issues + private currentBlobUrl: string | null = null; + + constructor(onComplete?: () => void) { + this.onComplete = onComplete; + // Create a single Audio element for the entire playback session + this.audioElement = new Audio(); + this.setupAudioListeners(); + } + + private setupAudioListeners() { + this.audioElement.onended = () => { + this.cleanupCurrentBlob(); + this.playNext(); + }; + + this.audioElement.onerror = () => { + console.error('Audio playback error'); + this.cleanupCurrentBlob(); + this.playNext(); + }; + } + + private cleanupCurrentBlob() { + if (this.currentBlobUrl) { + URL.revokeObjectURL(this.currentBlobUrl); + this.currentBlobUrl = null; + } + } + + addChunk(chunk: Blob) { + this.queue.push(chunk); + + // Start playback if we have enough buffer + if (!this.isPlaying && this.queue.length >= this.minBufferSize) { + this.playNext(); + } + } + + // Check if queue is full (for backpressure) + isFull(): boolean { + return this.queue.length - this.currentIndex >= this.maxQueueSize; + } + + // Get current queue size (unplayed chunks) + getQueueSize(): number { + return this.queue.length - this.currentIndex; + } + + private async playNext() { + if (this.currentIndex >= this.queue.length) { + // If stream has ended and no more chunks, complete + if (this.streamEnded) { + this.isPlaying = false; + this.onComplete?.(); + } + return; + } + + this.isPlaying = true; + const chunk = this.queue[this.currentIndex]; + this.currentIndex++; + + // Reuse the same Audio element, just update the src + this.cleanupCurrentBlob(); + this.currentBlobUrl = URL.createObjectURL(chunk); + this.audioElement.src = this.currentBlobUrl; + + try { + await this.audioElement.play(); + } catch (error) { + console.error('Failed to play audio chunk:', error); + this.cleanupCurrentBlob(); + this.playNext(); + } + } + + markStreamEnded() { + this.streamEnded = true; + // If not playing and has remaining chunks, start playing + if (!this.isPlaying && this.currentIndex < this.queue.length) { + this.playNext(); + } + } + + stop() { + this.isPlaying = false; + this.audioElement.pause(); + this.cleanupCurrentBlob(); + } + + clear() { + this.stop(); + this.queue = []; + this.currentIndex = 0; + this.streamEnded = false; + } +} + +export const useStreamTTS = (params?: UseStreamTTSParams) => { + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + const [progress, setProgress] = useState(0); + const controllerRef = useRef(null); + const audioQueueRef = useRef(null); + + const generate = useCallback( + async (ttsParams: TTSParams) => { + try { + setLoading(true); + setError(null); + setProgress(0); + + // Abort previous request if exists + controllerRef.current?.abort(); + audioQueueRef.current?.clear(); + + controllerRef.current = new AbortController(); + const signal = controllerRef.current.signal; + + // Create audio queue for smooth playback + audioQueueRef.current = new AudioQueue(); + + // Add stream parameter + const streamParams = { + ...ttsParams, + stream: true + }; + + const result = await fetchChunkedData({ + url: AUDIO_TEXT_TO_SPEECH_API, + data: streamParams, + signal + }); + + if ('error' in result) { + const errorMessage = extractErrorMessage(result.data); + setError({ + error: true, + errorMessage + }); + params?.onError?.(errorMessage); + return; + } + + const { reader } = result; + + if (!reader) { + throw new Error('Failed to get reader from response'); + } + + // Read stream data + let audioChunks: Uint8Array[] = []; + let allAudioChunks: Uint8Array[] = []; // Collect all chunks for final URL + let chunkCount = 0; + + while (true) { + // Backpressure: wait if queue is full + while (audioQueueRef.current?.isFull()) { + await new Promise((resolve) => { + setTimeout(resolve, 100); + }); + } + + const { done, value } = await reader.read(); + + if (done) { + // Process remaining chunks for playback + if (audioChunks.length > 0) { + const blob = new Blob(audioChunks as any, { + type: `audio/${ttsParams.response_format || 'mp3'}` + }); + audioQueueRef.current?.addChunk(blob); + params?.onChunk?.(blob.arrayBuffer() as any); + } + audioQueueRef.current?.markStreamEnded(); + + // Create complete audio URL from all chunks + if (allAudioChunks.length > 0) { + const completeBlob = new Blob(allAudioChunks as any, { + type: `audio/${ttsParams.response_format || 'mp3'}` + }); + const completeUrl = URL.createObjectURL(completeBlob); + params?.onComplete?.(completeUrl); + } + break; + } + + if (value) { + audioChunks.push(value); + allAudioChunks.push(value); // Keep all chunks for final URL + chunkCount++; + + // For fast API responses, batch chunks to avoid too many small audio elements + // Create a blob every N chunks or when chunk size exceeds threshold + const totalSize = audioChunks.reduce( + (sum, chunk) => sum + chunk.length, + 0 + ); + const shouldFlush = chunkCount >= 5 || totalSize >= 50000; // ~50KB threshold + + if (shouldFlush) { + const blob = new Blob(audioChunks as any, { + type: `audio/${ttsParams.response_format || 'mp3'}` + }); + audioQueueRef.current?.addChunk(blob); + params?.onChunk?.(blob.arrayBuffer() as any); + + audioChunks = []; + chunkCount = 0; + setProgress((prev) => prev + 1); + } + } + } + } catch (err: any) { + if (err.name === 'AbortError') { + console.log('Stream aborted'); + return; + } + + const errorMessage = err?.message || 'Stream processing failed'; + setError({ + error: true, + errorMessage + }); + params?.onError?.(errorMessage); + } finally { + setLoading(false); + } + }, + [params] + ); + + const abort = useCallback(() => { + controllerRef.current?.abort(); + audioQueueRef.current?.stop(); + setLoading(false); + }, []); + + return { + generate, + abort, + loading, + error, + progress + }; +}; diff --git a/src/pages/playground/speech/tts.tsx b/src/pages/playground/speech/tts.tsx index a1484c66..5478a249 100644 --- a/src/pages/playground/speech/tts.tsx +++ b/src/pages/playground/speech/tts.tsx @@ -14,16 +14,16 @@ import React, { useRef, useState } from 'react'; -import { AUDIO_TEXT_TO_SPEECH_API, CHAT_API, textToSpeech } from '../apis'; +import { AUDIO_TEXT_TO_SPEECH_API } from '../apis'; import MessageInput from '../components/message-input'; import RightContainer from '../components/right-container'; import ViewCommonCode from '../components/view-common-code'; -import { extractErrorMessage } from '../config'; import { MessageItem } from '../config/types'; import '../style/ground-llm.less'; import '../style/system-message-wrap.less'; import { TextToSpeechCode } from '../view-code/audio'; import TTSDataForm from './forms/tts-form'; +import { useNonStreamTTS, useStreamTTS } from './hooks'; interface MessageProps { modelList: Global.BaseOption[]; @@ -55,11 +55,58 @@ const GroundTTS: React.FC = forwardRef((props, ref) => { const [loading, setLoading] = useState(false); const [tokenResult, setTokenResult] = useState(null); const [collapse, setCollapse] = useState(false); - const controllerRef = useRef(null); const checkvalueRef = useRef(true); const [currentPrompt, setCurrentPrompt] = useState(''); const formRef = useRef(null); + // Initialize non-stream TTS hook + const nonStreamTTS = useNonStreamTTS({ + onSuccess: (result) => { + setMessageList([ + { + input: currentPrompt, + voice: parameters.voice, + format: parameters.response_format, + speed: parameters.speed, + uid: messageId.current, + autoplay: checkvalueRef.current, + audioUrl: result.url + } + ]); + }, + onError: (error) => { + setTokenResult({ + error: true, + errorMessage: error + }); + setMessageList([]); + } + }); + + // Initialize stream TTS hook + const streamTTS = useStreamTTS({ + autoPlay: checkvalueRef.current, + onComplete: (audioUrl) => { + // Update messageList with complete audio URL when stream finishes + setMessageList((prev) => { + if (prev.length > 0) { + const updated = [...prev]; + updated[updated.length - 1].audioUrl = audioUrl; + return updated; + } + return prev; + }); + console.log('Stream playback completed'); + }, + onError: (error) => { + setTokenResult({ + error: true, + errorMessage: error + }); + setMessageList([]); + } + }); + useImperativeHandle(ref, () => { return { viewCode() { @@ -104,7 +151,8 @@ const GroundTTS: React.FC = forwardRef((props, ref) => { }; const handleStopConversation = () => { - controllerRef.current?.abort?.(); + nonStreamTTS.abort(); + streamTTS.abort(); setLoading(false); }; @@ -116,61 +164,48 @@ const GroundTTS: React.FC = forwardRef((props, ref) => { try { await formRef.current?.form.validateFields(); if (!parameters.model) return; + setLoading(true); setMessageId(); setTokenResult(null); - setCurrentPrompt(current?.content || ''); + const inputText = current?.content || currentPrompt; + setCurrentPrompt(inputText); setMessageList([]); setRouteCache(routeCachekey['/playground/speech'], true); - controllerRef.current?.abort?.(); - controllerRef.current = new AbortController(); - const signal = controllerRef.current.signal; - const params = { ...dropEmptyFields(parameters), - input: current?.content || currentPrompt + input: inputText }; - const res: any = await textToSpeech({ - data: params, - url: CHAT_API, - signal - }); setParams(params); - console.log('result:', res); - - if ((res?.status_code && res?.status_code !== 200) || res?.error) { - setTokenResult({ - error: true, - errorMessage: extractErrorMessage(res) - }); - setMessageList([]); - return; + // Choose stream or non-stream based on parameters + if (parameters.stream) { + // Stream mode: audio will play in real-time + setMessageList([ + { + input: inputText, + voice: parameters.voice, + format: parameters.response_format, + speed: parameters.speed, + uid: messageId.current, + autoplay: checkvalueRef.current, + audioUrl: '' // No URL for stream mode, audio plays in real-time + } + ]); + await streamTTS.generate(params); + } else { + // Non-stream mode: get complete audio URL + await nonStreamTTS.generate(params); } - - setMessageList([ - { - input: current?.content || currentPrompt, - voice: parameters.voice, - format: parameters.response_format, - speed: parameters.speed, - uid: messageId.current, - autoplay: checkvalueRef.current, - audioUrl: res.url - } - ]); } catch (error: any) { - const res = error?.response?.data; console.log('error:', error); - if (res?.error) { - setTokenResult({ - error: true, - errorMessage: extractErrorMessage(res) - }); - } + setTokenResult({ + error: true, + errorMessage: error?.message || 'Unknown error' + }); } finally { setLoading(false); setRouteCache(routeCachekey['/playground/speech'], false);