diff --git a/packages/stage-pages/src/pages/devtools/providers-transcription-realtime-aliyun-nls.vue b/packages/stage-pages/src/pages/devtools/providers-transcription-realtime-aliyun-nls.vue index 6183f05fa..8656c5889 100644 --- a/packages/stage-pages/src/pages/devtools/providers-transcription-realtime-aliyun-nls.vue +++ b/packages/stage-pages/src/pages/devtools/providers-transcription-realtime-aliyun-nls.vue @@ -3,9 +3,8 @@ import type { ServerEvent, ServerEvents } from '@proj-airi/stage-ui/stores/provi import vadWorkletUrl from '@proj-airi/stage-ui/workers/vad/process.worklet?worker&url' -import { createAliyunNLSProvider } from '@proj-airi/stage-ui/stores/providers/aliyun/stream-transcription' +import { createAliyunNLSProvider, streamAliyunTranscription } from '@proj-airi/stage-ui/stores/providers/aliyun/stream-transcription' import { Button, FieldInput, FieldSelect } from '@proj-airi/ui' -import { streamTranscription } from '@xsai/stream-transcription' import { computed, nextTick, onBeforeUnmount, reactive, ref, shallowRef, watch } from 'vue' type AliyunRegion @@ -159,7 +158,7 @@ async function startRecording() { appendLog('Initializing realtime transcription session') - const transcriptionResult = streamTranscription({ + const transcriptionResult = streamAliyunTranscription({ ...createAliyunNLSProvider( credentials.accessKeyId.trim(), credentials.accessKeySecret.trim(), @@ -183,9 +182,8 @@ async function startRecording() { } }, }), - inputStream: audioStream, inputAudioStream: audioStream, - } as unknown as Parameters[0]) + } as unknown as Parameters[0]) transcriptionTextPromise.value = transcriptionResult.text isTranscribing.value = true diff --git a/packages/stage-ui/src/stores/modules/hearing.ts b/packages/stage-ui/src/stores/modules/hearing.ts index 1e7d5d215..a3476b92d 100644 --- a/packages/stage-ui/src/stores/modules/hearing.ts +++ b/packages/stage-ui/src/stores/modules/hearing.ts @@ -3,12 +3,12 @@ import type { WithUnknown } from '@xsai/shared' import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription' import { generateTranscription } from '@xsai/generate-transcription' -import { streamTranscription } from '@xsai/stream-transcription' import { defineStore, storeToRefs } from 'pinia' import { computed, ref } from 'vue' import { createResettableLocalStorage, createResettableRef } from '../../utils/resettable' import { useProvidersStore } from '../providers' +import { streamAliyunTranscription } from '../providers/aliyun/stream-transcription' export interface StreamTranscriptionFileInputOptions extends Omit { file: Blob @@ -36,7 +36,7 @@ interface HearingTranscriptionInvokeOptions { } const STREAM_TRANSCRIPTION_EXECUTORS: Record = { - 'aliyun-nls-transcription': streamTranscription as StreamTranscription, + 'aliyun-nls-transcription': streamAliyunTranscription, } export const useHearingStore = defineStore('hearing-store', () => { diff --git a/packages/stage-ui/src/stores/providers/aliyun/stream-transcription.ts b/packages/stage-ui/src/stores/providers/aliyun/stream-transcription.ts index c245bf681..8792d57cc 100644 --- a/packages/stage-ui/src/stores/providers/aliyun/stream-transcription.ts +++ b/packages/stage-ui/src/stores/providers/aliyun/stream-transcription.ts @@ -1,5 +1,6 @@ import type { SpeechProviderWithExtraOptions } from '@xsai-ext/shared-providers' -import type { StreamTranscriptionDelta } from '@xsai/stream-transcription' +import type { CommonRequestOptions } from '@xsai/shared' +import type { StreamTranscriptionDelta, StreamTranscriptionResult } from '@xsai/stream-transcription' import type { EventStartTranscription, ServerEvent, ServerEvents } from './' @@ -40,6 +41,15 @@ export interface AliyunStreamTranscriptionHandle { close: () => Promise } +interface AliyunStreamTranscriptionOptions extends AliyunRealtimeSpeechExtraOptions { + baseURL?: CommonRequestOptions['baseURL'] + fetch?: CommonRequestOptions['fetch'] + headers?: HeadersInit + file?: Blob + fileName?: string + inputStream?: ReadableStream +} + function toArrayBuffer(chunk: AudioChunk): ArrayBuffer { if (chunk instanceof ArrayBuffer) return chunk @@ -60,6 +70,67 @@ function encodeSSE(payload: StreamTranscriptionDelta): Uint8Array { return sseEncoder.encode(`data: ${JSON.stringify(payload)}\n\n`) } +// NOTICE: Copied/adapted from @xsai/stream-transcription SSE parsing to keep behavior consistent. +// Ref: @xsai/stream-transcription@0.4.0-beta.8 (dist/index.js parseChunk/transformChunk). +function parseSSELine(line: string): StreamTranscriptionDelta | undefined { + if (!line || !line.startsWith('data:')) + return undefined + + const content = line.slice('data:'.length) + const data = content.startsWith(' ') ? content.slice(1) : content + if (!data) + return undefined + + return JSON.parse(data) as StreamTranscriptionDelta +} + +function aliyunChunkTransformer() { + const decoder = new TextDecoder() + let buffer = '' + + return new TransformStream({ + transform: (chunk, controller) => { + buffer += decoder.decode(chunk, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + + for (const line of lines) { + const parsed = parseSSELine(line) + if (parsed) + controller.enqueue(parsed) + } + }, + flush: (controller) => { + if (!buffer) + return + const parsed = parseSSELine(buffer) + if (parsed) + controller.enqueue(parsed) + }, + }) +} + +// NOTICE: Copied/adapted from @xsai/stream-transcription delayed promise helper. +// Ref: @xsai/stream-transcription@0.4.0-beta.8 (dist/index.js DelayedPromise usage). +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + const promise = new Promise((res, rej) => { + resolve = res + reject = rej + }) + + return { promise, resolve, reject } +} + +function resolveAudioStream(options: AliyunStreamTranscriptionOptions): ReadableStream { + const stream = options.inputAudioStream ?? options.inputStream ?? options.file?.stream() + if (!stream) + throw new TypeError('Audio stream or file is required for Aliyun streaming transcription.') + + return stream as ReadableStream +} + interface InternalRealtimeOptions extends CreateAliyunStreamTranscriptionOptions { onSentenceFinal?: (payload: ServerEvents['SentenceEnd']) => Promise | void } @@ -114,8 +185,8 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise session.stop(websocket)) - websocket.close(1000, 'client closed') + mayThrow(() => session.stop(websocket)) + websocket.close(1000, 'client closed') } else { mayThrow(() => websocket?.close()) @@ -198,6 +269,87 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise() + + let text = '' + let textStreamCtrl: ReadableStreamDefaultController | undefined + let fullStreamCtrl: ReadableStreamDefaultController | undefined + + const fullStream = new ReadableStream({ + start(controller) { + fullStreamCtrl = controller + }, + }) + + const textStream = new ReadableStream({ + start(controller) { + textStreamCtrl = controller + }, + }) + + const doStream = async () => { + const requestTarget = options.baseURL instanceof URL + ? options.baseURL + : new URL(typeof options.baseURL === 'string' ? options.baseURL : 'http://localhost') + const response = await fetcher(requestTarget, { + body: audioStream, + headers: options.headers, + method: 'POST', + signal: options.abortSignal, + }) + + if (!response.ok) + throw new Error(`Aliyun streaming transcription request failed with status ${response.status}`) + + if (!response.body) + throw new Error('Streaming transcription response is missing a readable body.') + + await response.body + .pipeThrough(aliyunChunkTransformer()) + .pipeTo(new WritableStream({ + write: (chunk) => { + fullStreamCtrl?.enqueue(chunk) + if (chunk.type === 'transcript.text.delta') { + text += chunk.delta + textStreamCtrl?.enqueue(chunk.delta) + } + }, + close: () => { + fullStreamCtrl?.close() + textStreamCtrl?.close() + }, + abort: (reason) => { + fullStreamCtrl?.error(reason) + textStreamCtrl?.error(reason) + }, + })) + } + + void (async () => { + try { + await doStream() + deferredText.resolve(text) + } + catch (error) { + fullStreamCtrl?.error(error) + textStreamCtrl?.error(error) + deferredText.reject(error) + } + })() + + // REVIEW: We mirrored the streaming orchestration from @xsai/stream-transcription instead of + // patching the upstream package because Aliyun uses a custom websocket+fetch bridge (no FormData). + // Keeping it local avoids diverging from the published package while we wait for upstream support. + return { + fullStream, + text: deferredText.promise, + textStream, + } +} + export function createAliyunNLSProvider( accessKeyId: string, accessKeySecret: string,