From b04cc02a5a6f20b42cf887ca7acbea3e244426d9 Mon Sep 17 00:00:00 2001 From: Neko Date: Wed, 12 Aug 2026 00:58:53 +0800 Subject: [PATCH] fix(stage-ui): preserve streaming transcript corrections (#2261) --- .../composables/use-transcriptions.test.ts | 57 +++++++- .../src/composables/use-transcriptions.ts | 14 +- .../use-streaming-transcription-input.test.ts | 56 ++++++++ .../use-streaming-transcription-input.ts | 88 ++++++++++++ .../providers/aliyun-nls/provider.test.ts | 135 ++++++++++++++++++ .../providers/aliyun-nls/provider.ts | 61 ++++++-- .../stream-transcription/index.test.ts | 32 +++++ .../providers/stream-transcription/index.ts | 34 ++++- .../stage-ui/src/stores/modules/hearing.ts | 34 +++-- .../streaming-transcription-consumers.test.ts | 13 ++ .../streaming-transcription-consumers.ts | 8 ++ 11 files changed, 497 insertions(+), 35 deletions(-) create mode 100644 packages/stage-ui/src/composables/use-streaming-transcription-input.test.ts create mode 100644 packages/stage-ui/src/composables/use-streaming-transcription-input.ts create mode 100644 packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.test.ts diff --git a/packages/stage-layouts/src/composables/use-transcriptions.test.ts b/packages/stage-layouts/src/composables/use-transcriptions.test.ts index b1a692d04..0ba0392b0 100644 --- a/packages/stage-layouts/src/composables/use-transcriptions.test.ts +++ b/packages/stage-layouts/src/composables/use-transcriptions.test.ts @@ -18,10 +18,16 @@ function createMockStore() { } const mockTranscribedContent = 'test content' +interface MockStreamingCallbacks { + onSentenceEnd: (delta: string) => void + onSpeechEnd?: (text: string) => void + onTranscriptionUpdate?: (text: string) => void +} + function createMockPipeline() { return { removeStreamingTranscriptionConsumer: vi.fn(), - transcribeForMediaStream: vi.fn().mockImplementation((_stream, options: { onSentenceEnd: (delta: string) => void }) => { + transcribeForMediaStream: vi.fn().mockImplementation((_stream, options: MockStreamingCallbacks) => { options.onSentenceEnd(mockTranscribedContent) }), stopStreamingTranscription: vi.fn().mockResolvedValue(undefined), @@ -264,6 +270,55 @@ describe('useTranscriptions', () => { expect(mockInput.value).toBe(`${prependText} ${mockTranscribedContent}`) }) + it('replaces volatile snapshots when the provider corrects text', async () => { + // ROOT CAUSE: + // + // The input consumer only accepted final deltas. It had no operation for + // replacing a provider-owned draft when the provider corrected that text. + const mockInput = ref('prefix') + const observedInputs: string[] = [] + mockHearingStore.configured.value = true + mockAudioDevice.stream.value = { id: 'stream-1' } as any + mockAudioDevice.enabled.value = true + mockHearingPipeline.supportsStreamInput.value = true + mockHearingPipeline.transcribeForMediaStream.mockImplementation((_stream, options: MockStreamingCallbacks) => { + options.onTranscriptionUpdate?.('今天天气很号') + observedInputs.push(mockInput.value) + options.onTranscriptionUpdate?.('今天天气很好') + observedInputs.push(mockInput.value) + options.onSentenceEnd('今天天气很好') + }) + + const { startStreamingTranscription } + = useTranscriptions({ ...createOptions(), messageInputRef: mockInput }) + + await startStreamingTranscription() + + expect(observedInputs).toEqual(['prefix 今天天气很号', 'prefix 今天天气很好']) + expect(mockInput.value).toBe('prefix 今天天气很好') + }) + + it('preserves manual input changes during a volatile transcription', async () => { + const mockInput = ref('prefix') + mockHearingStore.configured.value = true + mockAudioDevice.stream.value = { id: 'stream-1' } as any + mockAudioDevice.enabled.value = true + mockHearingPipeline.supportsStreamInput.value = true + mockHearingPipeline.transcribeForMediaStream.mockImplementation((_stream, options: MockStreamingCallbacks) => { + options.onTranscriptionUpdate?.('provider draft') + mockInput.value = 'manual edit' + options.onTranscriptionUpdate?.('provider correction') + options.onSentenceEnd('provider final') + }) + + const { startStreamingTranscription } + = useTranscriptions({ ...createOptions(), messageInputRef: mockInput }) + + await startStreamingTranscription() + + expect(mockInput.value).toBe('manual edit') + }) + it('should trigger auto-send after delay', async () => { const mockInput = ref('') const mockSendMessage = vi.fn() diff --git a/packages/stage-layouts/src/composables/use-transcriptions.ts b/packages/stage-layouts/src/composables/use-transcriptions.ts index cae576879..9b08839f0 100644 --- a/packages/stage-layouts/src/composables/use-transcriptions.ts +++ b/packages/stage-layouts/src/composables/use-transcriptions.ts @@ -1,5 +1,6 @@ import type { MaybeRefOrGetter, Ref } from 'vue' +import { useStreamingTranscriptionInput } from '@proj-airi/stage-ui/composables/use-streaming-transcription-input' import { useHearingSpeechInputPipeline, useHearingStore } from '@proj-airi/stage-ui/stores/modules/hearing' import { useProviderStore } from '@proj-airi/stage-ui/stores/providers/provider' import { useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings' @@ -28,6 +29,7 @@ export function useTranscriptions(options: TranscriptionOptions) { const isListening = ref(false) const transcriptionConsumerId = `interactive-area:${useId()}` + const streamingInput = useStreamingTranscriptionInput(messageInput) // Auto-send logic let autoSendTimeout: ReturnType | undefined @@ -60,6 +62,7 @@ export function useTranscriptions(options: TranscriptionOptions) { const stopStreaming = async () => { removeStreamingTranscriptionConsumer(transcriptionConsumerId) + streamingInput.clear() if (!isListening.value) return @@ -181,15 +184,13 @@ export function useTranscriptions(options: TranscriptionOptions) { await transcribeForMediaStream(stream.value, { consumerId: transcriptionConsumerId, onSentenceEnd: (delta) => { - if (delta && delta.trim()) { - console.info('Received transcription delta:', delta, { source: 'useTranscriptions' }) - // Append transcribed text to message input - const currentText = messageInput.value.trim() - messageInput.value = currentText ? `${currentText} ${delta}` : delta + if (streamingInput.commit(delta)) { + console.info('Received final transcription:', delta, { source: 'useTranscriptions' }) debouncedAutoSend() } }, - // Omit onSpeechEnd to avoid re-adding user-deleted text; use sentence deltas only. + onSpeechEnd: streamingInput.clear, + onTranscriptionUpdate: streamingInput.replace, }) // Only set listening to true if transcription started successfully @@ -198,6 +199,7 @@ export function useTranscriptions(options: TranscriptionOptions) { console.info('Streaming transcription initiated successfully', { source: 'useTranscriptions' }) } catch (err) { + streamingInput.clear() console.error('Transcription error:', err, { source: 'useTranscriptions' }) isListening.value = false throw err diff --git a/packages/stage-ui/src/composables/use-streaming-transcription-input.test.ts b/packages/stage-ui/src/composables/use-streaming-transcription-input.test.ts new file mode 100644 index 000000000..4063b7ad5 --- /dev/null +++ b/packages/stage-ui/src/composables/use-streaming-transcription-input.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest' +import { shallowRef } from 'vue' + +import { useStreamingTranscriptionInput } from './use-streaming-transcription-input' + +describe('streaming transcription input', () => { + it('replaces provider corrections and clears only the owned suffix', () => { + const input = shallowRef('manual prefix') + const transcription = useStreamingTranscriptionInput(input) + + transcription.replace('今天天气很号') + transcription.replace('今天天气很好') + + expect(input.value).toBe('manual prefix 今天天气很好') + + transcription.clear() + + expect(input.value).toBe('manual prefix') + }) + + it('keeps manual edits after the user changes the volatile suffix', () => { + const input = shallowRef('manual prefix') + const transcription = useStreamingTranscriptionInput(input) + + transcription.replace('provider draft') + input.value = 'user replacement' + transcription.replace('provider correction') + transcription.clear() + + expect(input.value).toBe('user replacement') + }) + + it('commits the final correction as stable input', () => { + const input = shallowRef('manual prefix') + const transcription = useStreamingTranscriptionInput(input) + + transcription.replace('provider draft') + + expect(transcription.commit('provider final')).toBe(true) + expect(input.value).toBe('manual prefix provider final') + + transcription.clear() + + expect(input.value).toBe('manual prefix provider final') + }) + + it('removes the provider draft when the final transcript is empty', () => { + const input = shallowRef('manual prefix') + const transcription = useStreamingTranscriptionInput(input) + + transcription.replace('provider draft') + + expect(transcription.commit('')).toBe(false) + expect(input.value).toBe('manual prefix') + }) +}) diff --git a/packages/stage-ui/src/composables/use-streaming-transcription-input.ts b/packages/stage-ui/src/composables/use-streaming-transcription-input.ts new file mode 100644 index 000000000..7ef6a0520 --- /dev/null +++ b/packages/stage-ui/src/composables/use-streaming-transcription-input.ts @@ -0,0 +1,88 @@ +import type { Ref } from 'vue' + +function joinInputAndTranscription(input: string, transcription: string) { + return [input.trimEnd(), transcription.trim()].filter(Boolean).join(' ') +} + +/** + * Applies replaceable streaming transcription text to an editable input. + * Manual input changes detach the current provider-owned suffix. + */ +export function useStreamingTranscriptionInput(input: Ref) { + let volatileTranscription = '' + let volatileTranscriptionDetached = false + + function reset() { + volatileTranscription = '' + volatileTranscriptionDetached = false + } + + function clear() { + if (volatileTranscription && input.value.endsWith(volatileTranscription)) + input.value = input.value.slice(0, -volatileTranscription.length).trimEnd() + + reset() + } + + function replace(text: string) { + if (volatileTranscriptionDetached) + return + + const nextTranscription = text.trim() + if (!nextTranscription) { + clear() + return + } + + if (!volatileTranscription) { + volatileTranscription = nextTranscription + input.value = joinInputAndTranscription(input.value, nextTranscription) + return + } + + if (!input.value.endsWith(volatileTranscription)) { + volatileTranscriptionDetached = true + volatileTranscription = '' + return + } + + const stableInput = input.value.slice(0, -volatileTranscription.length) + volatileTranscription = nextTranscription + input.value = joinInputAndTranscription(stableInput, nextTranscription) + } + + function commit(text: string) { + if (volatileTranscriptionDetached) { + reset() + return false + } + + const finalTranscription = text.trim() + if (!finalTranscription) { + clear() + return false + } + + if (!volatileTranscription) { + input.value = joinInputAndTranscription(input.value, finalTranscription) + return true + } + + if (!input.value.endsWith(volatileTranscription)) { + reset() + return false + } + + const stableInput = input.value.slice(0, -volatileTranscription.length) + input.value = joinInputAndTranscription(stableInput, finalTranscription) + reset() + return true + } + + return { + clear, + commit, + replace, + reset, + } +} diff --git a/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.test.ts b/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.test.ts new file mode 100644 index 000000000..7635ecd70 --- /dev/null +++ b/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.test.ts @@ -0,0 +1,135 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { createAliyunNLSProvider } from './provider' + +class FakeWebSocket extends EventTarget { + static readonly CONNECTING = 0 + static readonly OPEN = 1 + static readonly CLOSING = 2 + static readonly CLOSED = 3 + static readonly instances: FakeWebSocket[] = [] + + binaryType: BinaryType = 'blob' + onclose: ((event: { code: number, reason: string }) => void) | null = null + onerror: ((event: Event) => void) | null = null + onmessage: ((event: MessageEvent) => void) | null = null + onopen: ((event: Event) => void) | null = null + readyState = FakeWebSocket.CONNECTING + sent: unknown[] = [] + + constructor(readonly url: string | URL) { + super() + FakeWebSocket.instances.push(this) + } + + close(code = 1000, reason = '') { + this.readyState = FakeWebSocket.CLOSED + this.onclose?.({ code, reason }) + } + + open() { + this.readyState = FakeWebSocket.OPEN + this.onopen?.(new Event('open')) + } + + receive(payload: unknown) { + this.onmessage?.(new MessageEvent('message', { data: JSON.stringify(payload) })) + } + + send(data: unknown) { + this.sent.push(data) + } +} + +afterEach(() => { + FakeWebSocket.instances.length = 0 + vi.unstubAllGlobals() +}) + +describe('aliyun NLS provider', () => { + it('forwards interim corrections and the final sentence as transcript snapshots', async () => { + // ROOT CAUSE: + // + // Aliyun emits each volatile hypothesis as `TranscriptionResultChanged`. + // The provider ignored these events and emitted only the final sentence. + vi.stubGlobal('WebSocket', FakeWebSocket) + vi.stubGlobal('fetch', vi.fn(async () => new Response( + JSON.stringify({ + NlsRequestId: 'nls-request', + RequestId: 'request', + ErrMsg: '', + Token: { + ExpireTime: Math.floor(Date.now() / 1000) + 3600, + Id: 'token', + UserId: 'user', + }, + }), + { headers: { 'Content-Type': 'application/json' } }, + ))) + + const inputAudioStream = new ReadableStream() + const provider = createAliyunNLSProvider('access-key-id', 'access-key-secret', 'app-key') + const speech = provider.speech('aliyun-nls-v1', { inputAudioStream }) + if (!speech.fetch) + throw new Error('The Aliyun provider did not create its streaming transport.') + + const response = await speech.fetch(new URL('https://example.invalid/transcription'), {}) + const responseText = response.text() + + await vi.waitFor(() => expect(FakeWebSocket.instances).toHaveLength(1)) + const socket = FakeWebSocket.instances[0] + if (!socket) + throw new Error('The Aliyun provider did not open its streaming socket.') + socket.open() + await vi.waitFor(() => expect(socket.sent).toHaveLength(1)) + const startEvent = JSON.parse(socket.sent[0] as string) + expect(startEvent.payload.enable_intermediate_result).toBe(true) + + socket.receive(serverEvent('SentenceBegin', { index: 1, time: 200 })) + socket.receive(serverEvent('TranscriptionResultChanged', { + index: 1, + result: '今天天气很号', + status: 20000000, + time: 1000, + })) + socket.receive(serverEvent('TranscriptionResultChanged', { + index: 1, + result: '今天天气很好', + status: 20000000, + time: 1200, + })) + socket.receive(serverEvent('SentenceEnd', { + begin_time: 200, + confidence: 0.95, + index: 1, + result: '今天天气很好', + stash_result: { + beginTime: 0, + currentTime: 1200, + sentenceId: 2, + text: '', + }, + status: 20000000, + time: 1200, + })) + socket.receive(serverEvent('TranscriptionCompleted', undefined)) + + await expect(responseText).resolves.toContain('"type":"transcript.text.snapshot","text":"今天天气很号","isFinal":false') + await expect(responseText).resolves.toContain('"type":"transcript.text.snapshot","text":"今天天气很好","isFinal":true') + }) +}) + +function serverEvent(name: string, payload: unknown) { + return { + header: { + appkey: 'app-key', + message_id: 'message-id', + name, + namespace: 'SpeechTranscriber', + status: 20000000, + status_message: 'SUCCESS', + task_id: 'task-id', + }, + payload, + } +} diff --git a/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.ts b/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.ts index 21e27b153..bd79ad9ef 100644 --- a/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.ts +++ b/packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.ts @@ -1,5 +1,6 @@ import type { SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils' +import type { StreamTranscriptionSnapshot } from '../../stream-transcription' import type { EventStartTranscription, ServerEvent, ServerEvents } from './session' import { tryCatch } from '@moeru/std' @@ -58,6 +59,7 @@ function createWaiter(timeoutMs: number, abortSignal?: AbortSignal) { } const DEFAULT_SESSION_OPTIONS: EventStartTranscription['payload'] = { + enable_intermediate_result: true, format: 'pcm', sample_rate: 16000, } @@ -104,12 +106,16 @@ function toArrayBuffer(chunk: AudioChunk): ArrayBuffer { const sseEncoder = new TextEncoder() -function encodeSSE(payload: { delta: string, type: 'transcript.text.delta' | 'transcript.text.done' }): Uint8Array { +type AliyunTranscriptionSSEEvent + = | StreamTranscriptionSnapshot + | { delta: string, type: 'transcript.text.done' } + +function encodeSSE(payload: AliyunTranscriptionSSEEvent): Uint8Array { return sseEncoder.encode(`data: ${JSON.stringify(payload)}\n\n`) } interface InternalRealtimeOptions extends CreateAliyunStreamTranscriptionOptions { - onSentenceFinal?: (payload: ServerEvents['SentenceEnd']) => Promise | void + onTranscriptSnapshot?: (snapshot: StreamTranscriptionSnapshot) => Promise | void idleTimeoutMs?: number stopAckTimeoutMs?: number } @@ -125,7 +131,7 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise | undefined + const sentenceStartMillisecondsByIndex = new Map() + const sentenceTextByIndex = new Map() + + function transcriptSnapshot( + payload: Pick, + isFinal: boolean, + ): StreamTranscriptionSnapshot { + sentenceTextByIndex.set(payload.index, payload.result) + const orderedText = [...sentenceTextByIndex.entries()] + .sort(([leftIndex], [rightIndex]) => leftIndex - rightIndex) + .map(([, text]) => text.trim()) + .filter(Boolean) + .join('\n') + const startMilliseconds = Math.min(...sentenceStartMillisecondsByIndex.values(), payload.time) + + return { + type: 'transcript.text.snapshot', + text: orderedText, + isFinal, + locale: 'und', + startMilliseconds, + durationMilliseconds: Math.max(0, payload.time - startMilliseconds), + } + } async function requestStop(reason?: unknown) { if (stopping) @@ -253,9 +283,22 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise { - const text = payload.result ? `${payload.result}\n` : '' - if (text) - controller.enqueue(encodeSSE({ delta: text, type: 'transcript.text.delta' })) - - controller.enqueue(encodeSSE({ delta: '', type: 'transcript.text.done' })) + onTranscriptSnapshot: async (snapshot) => { + controller.enqueue(encodeSSE(snapshot)) }, }).then((handle) => { sessionHandle = handle diff --git a/packages/stage-ui/src/libs/providers/stream-transcription/index.test.ts b/packages/stage-ui/src/libs/providers/stream-transcription/index.test.ts index eaf2d2dd5..b0c48f231 100644 --- a/packages/stage-ui/src/libs/providers/stream-transcription/index.test.ts +++ b/packages/stage-ui/src/libs/providers/stream-transcription/index.test.ts @@ -31,4 +31,36 @@ describe('streamTranscription', () => { it('rejects requests without an audio input', () => { expect(() => streamTranscription({})).toThrow('Audio stream or file is required') }) + + it('replaces volatile transcript snapshots instead of appending corrections', async () => { + // ROOT CAUSE: + // + // The adapter only accumulated `transcript.text.delta` events. Providers + // that emit complete volatile hypotheses could not replace incorrect text. + const encoder = new TextEncoder() + const responseBody = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode('data: {"type":"transcript.text.snapshot","text":"今天天气很号","isFinal":false,"locale":"zh-CN","startMilliseconds":0,"durationMilliseconds":1000}\n\n')) + controller.enqueue(encoder.encode('data: {"type":"transcript.text.snapshot","text":"今天天气很好","isFinal":true,"locale":"zh-CN","startMilliseconds":0,"durationMilliseconds":1200}\n\n')) + controller.close() + }, + }) + const audioStream = new ReadableStream({ + start(controller) { + controller.close() + }, + }) + + const result = streamTranscription({ + baseURL: 'https://example.invalid/transcription', + fetch: async () => new Response(responseBody), + inputAudioStream: audioStream, + }) + const updates = [] + for await (const update of result.fullStream) + updates.push(update) + + expect(updates).toHaveLength(2) + expect(await result.text).toBe('今天天气很好') + }) }) diff --git a/packages/stage-ui/src/libs/providers/stream-transcription/index.ts b/packages/stage-ui/src/libs/providers/stream-transcription/index.ts index d3dd7a355..ae6fbc323 100644 --- a/packages/stage-ui/src/libs/providers/stream-transcription/index.ts +++ b/packages/stage-ui/src/libs/providers/stream-transcription/index.ts @@ -3,6 +3,23 @@ import type { StreamTranscriptionDelta, StreamTranscriptionResult } from '@xsai/ type AudioChunk = ArrayBuffer | ArrayBufferView +/** A complete transcript snapshot that replaces earlier volatile text. */ +export interface StreamTranscriptionSnapshot { + durationMilliseconds: number + isFinal: boolean + locale: string + startMilliseconds: number + text: string + type: 'transcript.text.snapshot' +} + +export type AIRIStreamTranscriptionDelta = StreamTranscriptionDelta | StreamTranscriptionSnapshot + +/** xsAI stream result with AIRI's replaceable snapshot event. */ +export interface AIRIStreamTranscriptionResult extends Omit { + fullStream: ReadableStream +} + /** Options for adapting an SSE transcription request to xsAI stream results. */ export interface StreamTranscriptionOptions { abortSignal?: AbortSignal @@ -33,7 +50,7 @@ function resolveAudioStream(options: StreamTranscriptionOptions): ReadableStream return stream as ReadableStream } -function parseSSELine(line: string): StreamTranscriptionDelta | undefined { +function parseSSELine(line: string): AIRIStreamTranscriptionDelta | undefined { if (!line || !line.startsWith('data:')) return undefined @@ -42,14 +59,14 @@ function parseSSELine(line: string): StreamTranscriptionDelta | undefined { if (!data) return undefined - return JSON.parse(data) as StreamTranscriptionDelta + return JSON.parse(data) as AIRIStreamTranscriptionDelta } function createSSETransformer() { const decoder = new TextDecoder() let buffer = '' - return new TransformStream({ + return new TransformStream({ transform: (chunk, controller) => { buffer += decoder.decode(chunk, { stream: true }) const lines = buffer.split('\n') @@ -78,16 +95,16 @@ function createSSETransformer() { * The provider owns transport details. This adapter owns only request input, * SSE parsing, and the result streams consumed by Hearing. */ -export function streamTranscription(options: StreamTranscriptionOptions): StreamTranscriptionResult { +export function streamTranscription(options: StreamTranscriptionOptions): AIRIStreamTranscriptionResult { const audioStream = resolveAudioStream(options) const fetcher = options.fetch ?? globalThis.fetch const deferredText = createDeferred() let text = '' let textStreamCtrl: ReadableStreamDefaultController | undefined - let fullStreamCtrl: ReadableStreamDefaultController | undefined + let fullStreamCtrl: ReadableStreamDefaultController | undefined - const fullStream = new ReadableStream({ + const fullStream = new ReadableStream({ start(controller) { fullStreamCtrl = controller }, @@ -119,13 +136,16 @@ export function streamTranscription(options: StreamTranscriptionOptions): Stream await response.body .pipeThrough(createSSETransformer()) - .pipeTo(new WritableStream({ + .pipeTo(new WritableStream({ write: (chunk) => { fullStreamCtrl?.enqueue(chunk) if (chunk.type === 'transcript.text.delta') { text += chunk.delta textStreamCtrl?.enqueue(chunk.delta) } + else if (chunk.type === 'transcript.text.snapshot') { + text = chunk.text + } }, close: () => { fullStreamCtrl?.close() diff --git a/packages/stage-ui/src/stores/modules/hearing.ts b/packages/stage-ui/src/stores/modules/hearing.ts index 8e4df1620..f30abaf98 100644 --- a/packages/stage-ui/src/stores/modules/hearing.ts +++ b/packages/stage-ui/src/stores/modules/hearing.ts @@ -1,8 +1,9 @@ import type { Span } from '@opentelemetry/api' import type { TranscriptionProviderWithExtraOptions } from '@xsai-ext/providers/utils' import type { WithUnknown } from '@xsai/shared' -import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription' +import type { StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription' +import type { AIRIStreamTranscriptionResult } from '../../libs/providers/stream-transcription' import type { StreamingTranscriptionCallbacks, StreamingTranscriptionConsumer } from './streaming-transcription-consumers' import { errorMessageFrom, tryCatch } from '@moeru/std' @@ -90,11 +91,11 @@ export interface StreamTranscriptionStreamInputOptions extends Omit } -export type StreamTranscription = (options: WithUnknown) => StreamTranscriptionResult +export type StreamTranscription = (options: WithUnknown) => AIRIStreamTranscriptionResult type GenerateTranscriptionResponse = Awaited> type HearingTranscriptionGenerateResult = GenerateTranscriptionResponse & { mode: 'generate' } -type HearingTranscriptionStreamResult = StreamTranscriptionResult & { mode: 'stream' } +type HearingTranscriptionStreamResult = AIRIStreamTranscriptionResult & { mode: 'stream' } export type HearingTranscriptionResult = HearingTranscriptionGenerateResult | HearingTranscriptionStreamResult type HearingTranscriptionInput = File | { @@ -573,6 +574,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech const streamingCallbacks = { onSentenceEnd: (delta: string) => streamingConsumers.emitSentenceEnd(delta), onSpeechEnd: (text: string) => streamingConsumers.emitSpeechEnd(text), + onTranscriptionUpdate: (text: string) => streamingConsumers.emitTranscriptionUpdate(text), } const { trackAudioDeviceUnavailable, @@ -841,26 +843,34 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech session: NonNullable, result: HearingTranscriptionResult, ) { - if (result.mode !== 'stream' || !result.textStream) + if (result.mode !== 'stream' || !result.fullStream) return const sessionSpan = asrSpan const sessionCallbacks = session.callbacks void (async () => { let fullText = '' + let latestSnapshotIsFinal = false try { - const reader = result.textStream.getReader() + const reader = result.fullStream.getReader() while (true) { const { done, value } = await reader.read() if (done) break - if (!value) + if (value.type === 'transcript.text.snapshot') { + latestSnapshotIsFinal = value.isFinal + fullText = value.text + sessionCallbacks?.onTranscriptionUpdate?.(fullText) + continue + } + if (value.type !== 'transcript.text.delta' || !value.delta) continue - fullText += value - sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: value }) - sessionCallbacks?.onSentenceEnd?.(value) + fullText += value.delta + sessionCallbacks?.onTranscriptionUpdate?.(fullText) + sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: value.delta }) + sessionCallbacks?.onSentenceEnd?.(value.delta) } } catch (err) { @@ -868,6 +878,10 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech console.error('Error reading text stream:', err) } finally { + if (latestSnapshotIsFinal && fullText.trim()) { + sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: fullText }) + sessionCallbacks?.onSentenceEnd?.(fullText) + } sessionSpan?.setAttribute(IOAttributes.ASRText, fullText) sessionSpan?.end() if (asrSpan === sessionSpan) @@ -972,7 +986,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech supportsStreamInput: supportsStreamInput.value, hasStream: !!stream, providerId: activeTranscriptionProvider.value, - hasCallbacks: !!(options.onSentenceEnd || options.onSpeechEnd), + hasCallbacks: !!(options.onSentenceEnd || options.onSpeechEnd || options.onTranscriptionUpdate), }) if (!supportsStreamInput.value) { diff --git a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts index e0591ba41..9a929a557 100644 --- a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts +++ b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts @@ -60,4 +60,17 @@ describe('streaming transcription consumers', () => { consoleError.mockRestore() }) + + it('routes complete transcript updates independently from final sentences', () => { + const consumers = new StreamingTranscriptionConsumers() + const onSentenceEnd = vi.fn() + const onTranscriptionUpdate = vi.fn() + consumers.register({ consumerId: 'input', onSentenceEnd, onTranscriptionUpdate }) + + consumers.emitTranscriptionUpdate('provider correction') + + expect(onTranscriptionUpdate).toHaveBeenCalledOnce() + expect(onTranscriptionUpdate).toHaveBeenCalledWith('provider correction') + expect(onSentenceEnd).not.toHaveBeenCalled() + }) }) diff --git a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts index faead8b21..6797d8f69 100644 --- a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts +++ b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts @@ -2,6 +2,8 @@ export interface StreamingTranscriptionCallbacks { onSentenceEnd?: (delta: string) => void onSpeechEnd?: (text: string) => void + /** Receives the complete current transcript after each provider update. */ + onTranscriptionUpdate?: (text: string) => void } /** A consumer with a stable identity and its current callbacks. */ @@ -24,6 +26,7 @@ export class StreamingTranscriptionConsumers { this.consumers.set(consumer.consumerId, { onSentenceEnd: consumer.onSentenceEnd, onSpeechEnd: consumer.onSpeechEnd, + onTranscriptionUpdate: consumer.onTranscriptionUpdate, }) } @@ -42,6 +45,11 @@ export class StreamingTranscriptionConsumers { this.emit('onSpeechEnd', text) } + /** Sends the complete current transcript to all current consumers. */ + emitTranscriptionUpdate(text: string) { + this.emit('onTranscriptionUpdate', text) + } + private emit(callbackName: keyof StreamingTranscriptionCallbacks, text: string) { for (const [consumerId, callbacks] of this.consumers) { try {