diff --git a/apps/stage-pocket/package.json b/apps/stage-pocket/package.json index a2134e558..0469215d8 100644 --- a/apps/stage-pocket/package.json +++ b/apps/stage-pocket/package.json @@ -33,11 +33,13 @@ "@proj-airi/font-cjkfonts-allseto": "workspace:^", "@proj-airi/font-xiaolai": "workspace:^", "@proj-airi/i18n": "workspace:^", + "@proj-airi/pipelines-audio": "workspace:^", "@proj-airi/server-sdk": "workspace:^", "@proj-airi/stage-layouts": "workspace:^", "@proj-airi/stage-ui": "workspace:^", "@proj-airi/stage-ui-three": "workspace:^", "@proj-airi/stage-ui-three-performance-runtime": "workspace:^", + "@proj-airi/stream-kit": "workspace:^", "@proj-airi/ui": "workspace:^", "@proj-airi/ui-transitions": "workspace:^", "@standard-schema/spec": "^1.1.0", diff --git a/apps/stage-pocket/src/pages/devtools/performance-playground.vue b/apps/stage-pocket/src/pages/devtools/performance-playground.vue index a1e866038..34d959c1c 100644 --- a/apps/stage-pocket/src/pages/devtools/performance-playground.vue +++ b/apps/stage-pocket/src/pages/devtools/performance-playground.vue @@ -1,10 +1,10 @@ diff --git a/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue index 8f325d5df..a6ab0ba6b 100644 --- a/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue +++ b/apps/stage-tamagotchi/src/renderer/components/InteractiveArea.vue @@ -3,11 +3,9 @@ import type { ChatHistoryItem } from '@proj-airi/stage-ui/types/chat' import type { ChatProvider } from '@xsai-ext/providers/utils' import { ChatHistory } from '@proj-airi/stage-ui/components' -import { useMicVAD } from '@proj-airi/stage-ui/composables' import { useChatStore } from '@proj-airi/stage-ui/stores/chat' import { useConsciousnessStore } from '@proj-airi/stage-ui/stores/modules/consciousness' import { useProvidersStore } from '@proj-airi/stage-ui/stores/providers' -import { useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings' import { BasicTextarea } from '@proj-airi/ui' import { storeToRefs } from 'pinia' import { computed, ref, watch } from 'vue' @@ -16,11 +14,8 @@ import { useI18n } from 'vue-i18n' import { widgetsTools } from '../stores/tools/builtin/widgets' const messageInput = ref('') -const listening = ref(false) const attachments = ref<{ type: 'image', data: string, mimeType: string, url: string }[]>([]) -const { askPermission } = useSettingsAudioDevice() -const { enabled, selectedAudioInput } = storeToRefs(useSettingsAudioDevice()) const chatStore = useChatStore() const { send, onAfterMessageComposed, discoverToolsCompatibility, cleanupMessages } = chatStore const { messages, sending, streamingMessage } = storeToRefs(chatStore) @@ -100,49 +95,6 @@ function removeAttachment(index: number) { } } -const { destroy, start } = useMicVAD(selectedAudioInput, { - onSpeechStart: () => { - // TODO: interrupt the playback - // TODO: interrupt any of the ongoing TTS - // TODO: interrupt any of the ongoing LLM requests - // TODO: interrupt any of the ongoing animation of Live2D or VRM - // TODO: once interrupted, we should somehow switch to listen or thinking - // emotion / expression? - listening.value = true - }, - // VAD misfire means while speech end is detected but - // the frames of the segment of the audio buffer - // is not enough to be considered as a speech segment - // which controlled by the `minSpeechFrames` parameter - onVADMisfire: () => { - // TODO: do audio buffer send to whisper - listening.value = false - }, - onSpeechEnd: (buffer) => { - // TODO: do audio buffer send to whisper - listening.value = false - handleTranscription(buffer) - }, - auto: false, -}) - -function handleTranscription(_buffer: Float32Array) { - // eslint-disable-next-line no-alert - alert('Transcription is not implemented yet') -} - -watch(enabled, async (value) => { - if (value === false) { - destroy() - } - else { - await askPermission() - start() - } -}, { - immediate: true, -}) - watch([activeProvider, activeModel], async () => { if (activeProvider.value && activeModel.value) { await discoverToolsCompatibility(activeModel.value, await providersStore.getProviderInstance(activeProvider.value), []) diff --git a/apps/stage-web/package.json b/apps/stage-web/package.json index 28d1541ea..cc86521fe 100644 --- a/apps/stage-web/package.json +++ b/apps/stage-web/package.json @@ -29,12 +29,14 @@ "@proj-airi/font-xiaolai": "workspace:^", "@proj-airi/i18n": "workspace:^", "@proj-airi/model-driver-mediapipe": "workspace:^", + "@proj-airi/pipelines-audio": "workspace:^", "@proj-airi/server-sdk": "workspace:^", "@proj-airi/stage-layouts": "workspace:^", "@proj-airi/stage-shared": "workspace:^", "@proj-airi/stage-ui": "workspace:^", "@proj-airi/stage-ui-three": "workspace:^", "@proj-airi/stage-ui-three-performance-runtime": "workspace:^", + "@proj-airi/stream-kit": "workspace:^", "@proj-airi/ui": "workspace:^", "@proj-airi/ui-transitions": "workspace:^", "@standard-schema/spec": "^1.1.0", diff --git a/apps/stage-web/src/pages/devtools/performance-playground.vue b/apps/stage-web/src/pages/devtools/performance-playground.vue index a1e866038..34d959c1c 100644 --- a/apps/stage-web/src/pages/devtools/performance-playground.vue +++ b/apps/stage-web/src/pages/devtools/performance-playground.vue @@ -1,10 +1,10 @@ diff --git a/packages/pipelines-audio/package.json b/packages/pipelines-audio/package.json index ebbec397d..5a54cabd6 100644 --- a/packages/pipelines-audio/package.json +++ b/packages/pipelines-audio/package.json @@ -33,6 +33,7 @@ "typecheck": "tsc --noEmit" }, "dependencies": { + "@moeru/eventa": "^1.0.0-alpha.10", "@moeru/std": "catalog:", "clustr": "^1.0.2" }, diff --git a/packages/pipelines-audio/src/eventa.ts b/packages/pipelines-audio/src/eventa.ts new file mode 100644 index 000000000..cd634f5ea --- /dev/null +++ b/packages/pipelines-audio/src/eventa.ts @@ -0,0 +1,39 @@ +import type { + PlaybackEndEvent, + PlaybackInterruptEvent, + PlaybackRejectEvent, + PlaybackStartEvent, + TextSegment, + TtsRequest, + TtsResult, +} from './types' + +import { defineEventa } from '@moeru/eventa' + +export const speechSegmentEvent = defineEventa('proj-airi:pipelines:output:speech:segment') +export const speechSpecialEvent = defineEventa('proj-airi:pipelines:output:speech:special') +export const speechTtsRequestEvent = defineEventa('proj-airi:pipelines:output:speech:tts-request') +export const speechTtsResultEvent = defineEventa>('proj-airi:pipelines:output:speech:tts-result') +export const speechPlaybackStartEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-start') +export const speechPlaybackEndEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-end') +export const speechPlaybackInterruptEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-interrupt') +export const speechPlaybackRejectEvent = defineEventa>('proj-airi:pipelines:output:speech:playback-reject') +export const speechIntentStartEvent = defineEventa('proj-airi:pipelines:output:speech:intent-start') +export const speechIntentEndEvent = defineEventa('proj-airi:pipelines:output:speech:intent-end') +export const speechIntentCancelEvent = defineEventa<{ intentId: string, reason?: string }>('proj-airi:pipelines:output:speech:intent-cancel') + +export const speechPipelineEventMap = { + onSegment: speechSegmentEvent, + onSpecial: speechSpecialEvent, + onTtsRequest: speechTtsRequestEvent, + onTtsResult: speechTtsResultEvent, + onPlaybackStart: speechPlaybackStartEvent, + onPlaybackEnd: speechPlaybackEndEvent, + onPlaybackInterrupt: speechPlaybackInterruptEvent, + onPlaybackReject: speechPlaybackRejectEvent, + onIntentStart: speechIntentStartEvent, + onIntentEnd: speechIntentEndEvent, + onIntentCancel: speechIntentCancelEvent, +} as const + +export type SpeechPipelineEventName = keyof typeof speechPipelineEventMap diff --git a/packages/pipelines-audio/src/index.ts b/packages/pipelines-audio/src/index.ts index 336ce12bb..e39356c99 100644 --- a/packages/pipelines-audio/src/index.ts +++ b/packages/pipelines-audio/src/index.ts @@ -1 +1,7 @@ -export {} +export * from './eventa' +export * from './managers/playback-manager' +export * from './priority' +export * from './processors/tts-chunker' +export * from './speech-pipeline' +export * from './stream' +export * from './types' diff --git a/packages/pipelines-audio/src/managers/playback-manager.ts b/packages/pipelines-audio/src/managers/playback-manager.ts new file mode 100644 index 000000000..7c9506c46 --- /dev/null +++ b/packages/pipelines-audio/src/managers/playback-manager.ts @@ -0,0 +1,277 @@ +import type { + PlaybackEndEvent, + PlaybackInterruptEvent, + PlaybackItem, + PlaybackRejectEvent, + PlaybackStartEvent, +} from '../types' + +export type OverflowPolicy = 'queue' | 'reject' | 'steal-oldest' | 'steal-lowest-priority' +export type OwnerOverflowPolicy = 'reject' | 'steal-oldest' + +export interface PlaybackManagerOptions { + play: (item: PlaybackItem, signal: AbortSignal) => Promise + maxVoices?: number + maxVoicesPerOwner?: number + overflowPolicy?: OverflowPolicy + ownerOverflowPolicy?: OwnerOverflowPolicy +} + +export function createPlaybackManager(options: PlaybackManagerOptions) { + const maxVoices = Math.max(1, options.maxVoices ?? 1) + const maxVoicesPerOwner = options.maxVoicesPerOwner + const overflowPolicy = options.overflowPolicy ?? 'queue' + const ownerOverflowPolicy = options.ownerOverflowPolicy ?? 'steal-oldest' + + const active = new Map + controller: AbortController + startedAt: number + }>() + + const waiting: Array<{ item: PlaybackItem, enqueuedAt: number }> = [] + + const listeners = { + start: [] as Array<(event: PlaybackStartEvent) => void>, + end: [] as Array<(event: PlaybackEndEvent) => void>, + interrupt: [] as Array<(event: PlaybackInterruptEvent) => void>, + reject: [] as Array<(event: PlaybackRejectEvent) => void>, + } + + function onStart(listener: (event: PlaybackStartEvent) => void) { + listeners.start.push(listener) + } + + function onEnd(listener: (event: PlaybackEndEvent) => void) { + listeners.end.push(listener) + } + + function onInterrupt(listener: (event: PlaybackInterruptEvent) => void) { + listeners.interrupt.push(listener) + } + + function onReject(listener: (event: PlaybackRejectEvent) => void) { + listeners.reject.push(listener) + } + + function emitStart(item: PlaybackItem) { + const event = { item, startedAt: Date.now() } + listeners.start.forEach(listener => listener(event)) + } + + function emitEnd(item: PlaybackItem) { + const event = { item, endedAt: Date.now() } + listeners.end.forEach(listener => listener(event)) + } + + function emitInterrupt(item: PlaybackItem, reason: string) { + const event = { item, reason, interruptedAt: Date.now() } + listeners.interrupt.forEach(listener => listener(event)) + } + + function emitReject(item: PlaybackItem, reason: string) { + const event = { item, reason } + listeners.reject.forEach(listener => listener(event)) + } + + function countByOwner(ownerId?: string) { + if (!ownerId) + return 0 + let count = 0 + for (const entry of active.values()) { + if (entry.item.ownerId === ownerId) + count += 1 + } + return count + } + + function chooseVictimByPriority() { + let victim: { item: PlaybackItem, controller: AbortController, startedAt: number } | undefined + for (const entry of active.values()) { + if (!victim) + victim = entry + else if (entry.item.priority < victim.item.priority) + victim = entry + } + return victim + } + + function chooseVictimOldest(ownerId?: string) { + let victim: { item: PlaybackItem, controller: AbortController, startedAt: number } | undefined + for (const entry of active.values()) { + if (ownerId && entry.item.ownerId !== ownerId) + continue + if (!victim || entry.startedAt < victim.startedAt) + victim = entry + } + return victim + } + + function stopActive(entry: { item: PlaybackItem, controller: AbortController }, reason: string) { + entry.controller.abort(reason) + active.delete(entry.item.id) + emitInterrupt(entry.item, reason) + } + + function canStart(item: PlaybackItem) { + if (active.size >= maxVoices) + return { ok: false, reason: 'overflow' as const } + if (maxVoicesPerOwner && item.ownerId) { + if (countByOwner(item.ownerId) >= maxVoicesPerOwner) + return { ok: false, reason: 'owner-overflow' as const } + } + return { ok: true as const } + } + + function start(item: PlaybackItem) { + const controller = new AbortController() + const startedAt = Date.now() + active.set(item.id, { item, controller, startedAt }) + emitStart(item) + + void options.play(item, controller.signal) + .then(() => { + if (!active.has(item.id)) + return + active.delete(item.id) + emitEnd(item) + void tryStartWaiting() + }) + .catch((err) => { + if (!active.has(item.id)) + return + active.delete(item.id) + emitInterrupt(item, err instanceof Error ? err.message : 'playback-error') + void tryStartWaiting() + }) + } + + function tryStartWaiting() { + if (waiting.length === 0) + return + + const candidates = waiting + .slice() + .sort((a, b) => (b.item.priority - a.item.priority) || (a.enqueuedAt - b.enqueuedAt)) + + for (const candidate of candidates) { + const { ok, reason } = canStart(candidate.item) + if (!ok) { + if (reason === 'owner-overflow' && ownerOverflowPolicy === 'steal-oldest') { + const victim = chooseVictimOldest(candidate.item.ownerId) + if (victim) + stopActive(victim, 'owner-overflow') + } + continue + } + + const index = waiting.indexOf(candidate) + if (index >= 0) + waiting.splice(index, 1) + + start(candidate.item) + if (active.size >= maxVoices) + break + } + } + + function handleOverflow(item: PlaybackItem, reason: 'overflow' | 'owner-overflow') { + if (reason === 'owner-overflow') { + if (ownerOverflowPolicy === 'reject') { + emitReject(item, 'owner-overflow') + return + } + + const victim = chooseVictimOldest(item.ownerId) + if (victim) { + stopActive(victim, 'owner-overflow') + waiting.push({ item, enqueuedAt: Date.now() }) + void tryStartWaiting() + return + } + } + + switch (overflowPolicy) { + case 'reject': + emitReject(item, 'overflow') + break + case 'queue': + waiting.push({ item, enqueuedAt: Date.now() }) + break + case 'steal-oldest': { + const victim = chooseVictimOldest() + if (victim) + stopActive(victim, 'steal-oldest') + waiting.push({ item, enqueuedAt: Date.now() }) + void tryStartWaiting() + break + } + case 'steal-lowest-priority': { + const victim = chooseVictimByPriority() + if (victim && victim.item.priority <= item.priority) { + stopActive(victim, 'steal-lowest-priority') + waiting.push({ item, enqueuedAt: Date.now() }) + void tryStartWaiting() + } + else { + emitReject(item, 'lower-priority') + } + break + } + } + } + + function schedule(item: PlaybackItem) { + const { ok, reason } = canStart(item) + if (ok) { + start(item) + return + } + + handleOverflow(item, reason) + } + + function stopAll(reason: string) { + for (const entry of active.values()) { + stopActive(entry, reason) + } + waiting.length = 0 + } + + function stopByIntent(intentId: string, reason: string) { + for (const entry of active.values()) { + if (entry.item.intentId !== intentId) + continue + stopActive(entry, reason) + } + + for (let i = waiting.length - 1; i >= 0; i -= 1) { + if (waiting[i]?.item.intentId === intentId) + waiting.splice(i, 1) + } + } + + function stopByOwner(ownerId: string, reason: string) { + for (const entry of active.values()) { + if (entry.item.ownerId !== ownerId) + continue + stopActive(entry, reason) + } + + for (let i = waiting.length - 1; i >= 0; i -= 1) { + if (waiting[i]?.item.ownerId === ownerId) + waiting.splice(i, 1) + } + } + + return { + schedule, + stopAll, + stopByIntent, + stopByOwner, + onStart, + onEnd, + onInterrupt, + onReject, + } +} diff --git a/packages/pipelines-audio/src/priority.ts b/packages/pipelines-audio/src/priority.ts new file mode 100644 index 000000000..ff6471747 --- /dev/null +++ b/packages/pipelines-audio/src/priority.ts @@ -0,0 +1,28 @@ +import type { PriorityLevel, PriorityResolver } from './types' + +const DEFAULT_LEVELS: Record = { + critical: 300, + high: 200, + normal: 100, + low: 0, +} + +export function createPriorityResolver(levels?: Partial>): PriorityResolver { + const resolved = { ...DEFAULT_LEVELS, ...levels } + + return { + resolve(priority?: PriorityLevel | number) { + if (priority == null) + return resolved.normal + if (typeof priority === 'number') + return priority + return resolved[priority] ?? resolved.normal + }, + } +} + +export function comparePriority(a: number, b: number) { + if (a === b) + return 0 + return a > b ? 1 : -1 +} diff --git a/packages/pipelines-audio/src/processors/tts-chunker.ts b/packages/pipelines-audio/src/processors/tts-chunker.ts new file mode 100644 index 000000000..49f3147ee --- /dev/null +++ b/packages/pipelines-audio/src/processors/tts-chunker.ts @@ -0,0 +1,299 @@ +import type { ReaderLike } from 'clustr' + +import type { TextSegment, TextToken } from '../types' + +import { readGraphemeClusters } from 'clustr' + +import { createPushStream } from '../stream' + +export const TTS_FLUSH_INSTRUCTION = '\u200B' +export const TTS_SPECIAL_TOKEN = '\u2063' + +const keptPunctuations = new Set('?？!！') +const hardPunctuations = new Set('.。?？!！…⋯～~\n\t\r') +const softPunctuations = new Set(',，、–—:：;；《》「」') + +export interface TtsInputChunk { + text: string + words: number + reason: 'boost' | 'limit' | 'hard' | 'flush' | 'special' +} + +export interface TtsInputChunkOptions { + boost?: number + minimumWords?: number + maximumWords?: number +} + +export interface TtsChunkItem { + chunk: string + special: string | null + reason: 'boost' | 'limit' | 'hard' | 'flush' | 'special' +} + +export async function* chunkTtsInput( + input: string | ReaderLike, + options?: TtsInputChunkOptions, +): AsyncGenerator { + const { + boost = 2, + minimumWords = 4, + maximumWords = 12, + } = options ?? {} + + const iterator = readGraphemeClusters( + typeof input === 'string' + ? new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(input)) + controller.close() + }, + }).getReader() + : input, + ) + + const segmenter = new Intl.Segmenter(undefined, { granularity: 'word' }) // I love Intl.Segmenter + + let yieldCount = 0 + let buffer = '' + let chunk = '' + let chunkWordsCount = 0 + + let previousValue: string | undefined + let current = await iterator.next() + + while (!current.done) { + let value = current.value + + if (value.length > 1) { + previousValue = value + current = await iterator.next() + continue + } + + const flush = value === TTS_FLUSH_INSTRUCTION + const special = value === TTS_SPECIAL_TOKEN + const hard = hardPunctuations.has(value) + const soft = softPunctuations.has(value) + const kept = keptPunctuations.has(value) + let next: IteratorResult | undefined + let afterNext: IteratorResult | undefined + + if (flush || special || hard || soft) { + switch (value) { + case '.': + case ',': { + if (previousValue !== undefined && /\d/.test(previousValue)) { + next = await iterator.next() + if (!next.done && next.value && /\d/.test(next.value)) { + buffer += value + current = next + next = undefined + continue + } + } + else if (value === '.') { + next = await iterator.next() + if (!next.done && next.value && next.value === '.') { + afterNext = await iterator.next() + if (!afterNext.done && afterNext.value && afterNext.value === '.') { + value = '…' + next = undefined + afterNext = undefined + } + } + } + } + } + + if (buffer.length === 0) { + if (special) { + yield { + text: '', + words: 0, + reason: 'special', + } + yieldCount++ + chunkWordsCount = 0 + } + + previousValue = value + current = await iterator.next() + continue + } + + const words = [...segmenter.segment(buffer)].filter(w => w.isWordLike) + + if (chunkWordsCount > minimumWords && chunkWordsCount + words.length > maximumWords) { + const text = kept ? chunk.trim() + value : chunk.trim() + yield { + text, + words: chunkWordsCount, + reason: 'limit', + } + yieldCount++ + chunk = '' + chunkWordsCount = 0 + } + + chunk += buffer + value + chunkWordsCount += words.length + buffer = '' + + if (special) { + const text = chunk.slice(0, -1).trim() + yield { + text, + words: chunkWordsCount, + reason: 'special', + } + yieldCount++ + chunk = '' + chunkWordsCount = 0 + } + else if (flush || hard || chunkWordsCount > maximumWords || yieldCount < boost) { + const text = chunk.trim() + yield { + text, + words: chunkWordsCount, + reason: flush ? 'flush' : hard ? 'hard' : chunkWordsCount > maximumWords ? 'limit' : 'boost', + } + yieldCount++ + chunk = '' + chunkWordsCount = 0 + } + + previousValue = value + if (next !== undefined) { + if (afterNext !== undefined) { + current = afterNext + next = undefined + afterNext = undefined + } + else { + current = next + next = undefined + } + } + else { + current = await iterator.next() + } + continue + } + + buffer += value + previousValue = value + next = await iterator.next() + current = next + } + + // TODO: remove later + // eslint-disable-next-line no-console + console.debug('while loop ends, chunk/buffer:', chunk, buffer) + if (chunk.length > 0 || buffer.length > 0) { + const text = (chunk + buffer).trim() + yield { + text, + words: chunkWordsCount + [...segmenter.segment(buffer)].filter(w => w.isWordLike).length, + reason: 'flush', + } + } +} + +export async function chunkEmitter( + reader: ReaderLike, + pendingSpecials: string[], + options: TtsInputChunkOptions | undefined, + handler: (ttsSegment: TtsChunkItem) => Promise | void, +) { + const sanitizeChunk = (text: string) => + text + .replaceAll(TTS_SPECIAL_TOKEN, '') + .replaceAll(TTS_FLUSH_INSTRUCTION, '') + .trim() + + try { + for await (const chunk of chunkTtsInput(reader, options)) { + // TODO: remove later + + if (chunk.reason === 'special') { + const specialToken = pendingSpecials.shift() + // console.debug("special yield:", specialToken) + await handler({ chunk: sanitizeChunk(chunk.text), special: specialToken ?? null, reason: chunk.reason }) + } + else { + await handler({ chunk: sanitizeChunk(chunk.text), special: null, reason: chunk.reason }) + } + } + } + catch (e) { + console.error('Error chunking stream to TTS queue:', e) + } +} + +export function createTtsSegmentStream( + tokens: ReadableStream, + meta: { streamId: string, intentId: string }, + options?: TtsInputChunkOptions, +) { + const { stream, write, close, error } = createPushStream() + const pendingSpecials: string[] = [] + const encoder = new TextEncoder() + + const { stream: byteStream, write: writeBytes, close: closeBytes, error: errorBytes } = createPushStream() + + void (async () => { + const reader = tokens.getReader() + try { + while (true) { + const { value, done } = await reader.read() + if (done) + break + if (!value) + continue + + if (value.type === 'literal') { + if (value.value) + writeBytes(encoder.encode(value.value)) + } + else if (value.type === 'special') { + pendingSpecials.push(value.value ?? '') + writeBytes(encoder.encode(TTS_SPECIAL_TOKEN)) + } + else if (value.type === 'flush') { + writeBytes(encoder.encode(TTS_FLUSH_INSTRUCTION)) + } + } + closeBytes() + } + catch (err) { + errorBytes(err) + } + finally { + reader.releaseLock() + } + })() + + void (async () => { + try { + const reader = byteStream.getReader() + await chunkEmitter(reader, pendingSpecials, options, async (chunk) => { + write({ + streamId: meta.streamId, + intentId: meta.intentId, + segmentId: `${meta.streamId}:${Date.now()}:${Math.random().toString(36).slice(2, 8)}`, + text: chunk.chunk, + special: chunk.special, + reason: chunk.reason, + createdAt: Date.now(), + }) + }) + close() + } + catch (err) { + error(err) + } + })() + + return stream +} diff --git a/packages/pipelines-audio/src/speech-pipeline.ts b/packages/pipelines-audio/src/speech-pipeline.ts new file mode 100644 index 000000000..9e162e584 --- /dev/null +++ b/packages/pipelines-audio/src/speech-pipeline.ts @@ -0,0 +1,330 @@ +import type { Eventa } from '@moeru/eventa' + +import type { SpeechPipelineEventName } from './eventa' +import type { + IntentHandle, + IntentOptions, + LoggerLike, + PlaybackItem, + SpeechPipelineEvents, + TextSegment, + TextToken, + TtsRequest, + TtsResult, +} from './types' + +import { createContext } from '@moeru/eventa' + +import { speechPipelineEventMap } from './eventa' +import { createPriorityResolver } from './priority' +import { createTtsSegmentStream } from './processors/tts-chunker' +import { createPushStream } from './stream' + +export interface SpeechPipelineOptions { + tts: (request: TtsRequest, signal: AbortSignal) => Promise + playback: { + schedule: (item: PlaybackItem) => void + stopAll: (reason: string) => void + stopByIntent: (intentId: string, reason: string) => void + stopByOwner: (ownerId: string, reason: string) => void + onStart: (listener: (event: { item: PlaybackItem, startedAt: number }) => void) => void + onEnd: (listener: (event: { item: PlaybackItem, endedAt: number }) => void) => void + onInterrupt: (listener: (event: { item: PlaybackItem, reason: string, interruptedAt: number }) => void) => void + onReject: (listener: (event: { item: PlaybackItem, reason: string }) => void) => void + } + logger?: LoggerLike + priority?: ReturnType + segmenter?: (tokens: ReadableStream, meta: { streamId: string, intentId: string }) => ReadableStream +} + +interface IntentState { + intentId: string + streamId: string + priority: number + ownerId?: string + behavior: 'queue' | 'interrupt' | 'replace' + createdAt: number + controller: AbortController + stream: ReadableStream + closeStream: () => void + canceled: boolean +} + +function createId(prefix: string) { + return `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}` +} + +export function createSpeechPipeline(options: SpeechPipelineOptions) { + const logger = options.logger ?? console + const priorityResolver = options.priority ?? createPriorityResolver() + const segmenter = options.segmenter ?? createTtsSegmentStream + const context = createContext() + + const intents = new Map() + const pending: IntentState[] = [] + let activeIntent: IntentState | null = null + + options.playback.onStart(event => context.emit(speechPipelineEventMap.onPlaybackStart, event)) + options.playback.onEnd(event => context.emit(speechPipelineEventMap.onPlaybackEnd, event)) + options.playback.onInterrupt(event => context.emit(speechPipelineEventMap.onPlaybackInterrupt, event)) + options.playback.onReject(event => context.emit(speechPipelineEventMap.onPlaybackReject, event)) + + function enqueueIntent(intent: IntentState) { + pending.push(intent) + } + + function pickNextIntent() { + if (pending.length === 0) + return null + pending.sort((a, b) => (b.priority - a.priority) || (a.createdAt - b.createdAt)) + return pending.shift() ?? null + } + + async function runIntent(intent: IntentState) { + activeIntent = intent + context.emit(speechPipelineEventMap.onIntentStart, intent.intentId) + + const tokenStream = intent.stream + const segmentStream = segmenter(tokenStream, { streamId: intent.streamId, intentId: intent.intentId }) + + try { + const reader = segmentStream.getReader() + + while (true) { + const { value, done } = await reader.read() + if (done) + break + if (!value) + continue + if (intent.canceled || intent.controller.signal.aborted) { + await reader.cancel() + break + } + + context.emit(speechPipelineEventMap.onSegment, value) + + if (value.text === '' && value.special) { + context.emit(speechPipelineEventMap.onSpecial, value) + continue + } + + const request: TtsRequest = { + streamId: value.streamId, + intentId: value.intentId, + segmentId: value.segmentId, + text: value.text, + special: value.special, + priority: intent.priority, + createdAt: Date.now(), + } + + context.emit(speechPipelineEventMap.onTtsRequest, request) + + 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) + break + continue + } + + if (intent.controller.signal.aborted) + break + + if (!audio) + continue + + const ttsResult: TtsResult = { + 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(), + }) + } + + reader.releaseLock() + } + catch (err) { + logger.warn('Speech pipeline intent failed:', err) + } + finally { + if (intent.canceled) { + context.emit(speechPipelineEventMap.onIntentCancel, { intentId: intent.intentId, reason: intent.controller.signal.reason as string | undefined }) + } + else { + context.emit(speechPipelineEventMap.onIntentEnd, intent.intentId) + } + + intents.delete(intent.intentId) + activeIntent = null + + const next = pickNextIntent() + if (next) + void runIntent(next) + } + } + + function openIntent(optionsInput?: IntentOptions): IntentHandle { + const intentId = optionsInput?.intentId ?? createId('intent') + const streamId = optionsInput?.streamId ?? createId('stream') + const priority = priorityResolver.resolve(optionsInput?.priority) + const behavior = optionsInput?.behavior ?? 'queue' + const ownerId = optionsInput?.ownerId + + const controller = new AbortController() + const { stream, write, close } = createPushStream() + let sequence = 0 + + const intent: IntentState = { + intentId, + streamId, + priority, + ownerId, + behavior, + createdAt: Date.now(), + controller, + stream, + closeStream: close, + canceled: false, + } + + intents.set(intentId, intent) + + const handle: IntentHandle = { + intentId, + streamId, + priority, + ownerId, + stream, + writeLiteral(text: string) { + if (intent.canceled) + return + write({ + type: 'literal', + value: text, + streamId, + intentId, + sequence: sequence++, + createdAt: Date.now(), + }) + }, + writeSpecial(special: string) { + if (intent.canceled) + return + write({ + type: 'special', + value: special, + streamId, + intentId, + sequence: sequence++, + createdAt: Date.now(), + }) + }, + writeFlush() { + if (intent.canceled) + return + write({ + type: 'flush', + streamId, + intentId, + sequence: sequence++, + createdAt: Date.now(), + }) + }, + end() { + close() + }, + cancel(reason?: string) { + cancelIntent(intentId, reason) + }, + } + + if (!activeIntent) { + void runIntent(intent) + return handle + } + + if (behavior === 'replace') { + cancelIntent(activeIntent.intentId, 'replace') + void runIntent(intent) + return handle + } + + if (behavior === 'interrupt' && intent.priority >= activeIntent.priority) { + cancelIntent(activeIntent.intentId, 'interrupt') + void runIntent(intent) + return handle + } + + enqueueIntent(intent) + return handle + } + + function cancelIntent(intentId: string, reason?: string) { + const intent = intents.get(intentId) + if (!intent) + return + intent.canceled = true + intent.controller.abort(reason ?? 'canceled') + intent.closeStream() + + if (activeIntent?.intentId === intentId) { + options.playback.stopByIntent(intentId, reason ?? 'canceled') + return + } + + const index = pending.findIndex(item => item.intentId === intentId) + if (index >= 0) + pending.splice(index, 1) + } + + function interrupt(reason: string) { + if (activeIntent) + cancelIntent(activeIntent.intentId, reason) + } + + function stopAll(reason: string) { + for (const intent of intents.values()) { + intent.canceled = true + intent.controller.abort(reason) + intent.closeStream() + } + pending.length = 0 + intents.clear() + activeIntent = null + options.playback.stopAll(reason) + } + + return { + openIntent, + cancelIntent, + interrupt, + stopAll, + on(event: K, listener: SpeechPipelineEvents[K]) { + return context.on(speechPipelineEventMap[event] as Eventa, (payload) => { + listener(payload?.body ?? payload) + }) + }, + } +} diff --git a/packages/pipelines-audio/src/stream.ts b/packages/pipelines-audio/src/stream.ts new file mode 100644 index 000000000..327042854 --- /dev/null +++ b/packages/pipelines-audio/src/stream.ts @@ -0,0 +1,61 @@ +export interface StreamController { + stream: ReadableStream + write: (value: T) => void + close: () => void + error: (err: unknown) => void + isClosed: () => boolean +} + +export function createPushStream(): StreamController { + let closed = false + let controller: ReadableStreamDefaultController | null = null + + const stream = new ReadableStream({ + start(ctrl) { + controller = ctrl + }, + cancel() { + closed = true + }, + }) + + return { + stream, + write(value) { + if (!controller || closed) + return + controller.enqueue(value) + }, + close() { + if (!controller || closed) + return + closed = true + controller.close() + }, + error(err) { + if (!controller || closed) + return + closed = true + controller.error(err) + }, + isClosed() { + return closed + }, + } +} + +export async function readStream(stream: ReadableStream, handler: (value: T) => Promise | void) { + const reader = stream.getReader() + try { + while (true) { + const { value, done } = await reader.read() + if (done) + break + + await handler(value as T) + } + } + finally { + reader.releaseLock() + } +} diff --git a/packages/pipelines-audio/src/types.ts b/packages/pipelines-audio/src/types.ts new file mode 100644 index 000000000..eb24fdb64 --- /dev/null +++ b/packages/pipelines-audio/src/types.ts @@ -0,0 +1,122 @@ +export type PriorityLevel = 'critical' | 'high' | 'normal' | 'low' + +export interface PriorityResolver { + resolve: (priority?: PriorityLevel | number) => number +} + +export interface TextToken { + type: 'literal' | 'special' | 'flush' + value?: string + streamId: string + intentId: string + sequence: number + createdAt: number +} + +export interface TextSegment { + streamId: string + intentId: string + segmentId: string + text: string + special: string | null + reason: 'boost' | 'limit' | 'hard' | 'flush' | 'special' + createdAt: number +} + +export interface TtsRequest { + streamId: string + intentId: string + segmentId: string + text: string + special: string | null + priority: number + createdAt: number +} + +export interface TtsResult { + streamId: string + intentId: string + segmentId: string + text: string + special: string | null + audio: TAudio + createdAt: number +} + +export interface PlaybackItem { + id: string + streamId: string + intentId: string + segmentId: string + ownerId?: string + priority: number + text: string + special: string | null + audio: TAudio + createdAt: number +} + +export interface PlaybackStartEvent { + item: PlaybackItem + startedAt: number +} + +export interface PlaybackEndEvent { + item: PlaybackItem + endedAt: number +} + +export interface PlaybackInterruptEvent { + item: PlaybackItem + reason: string + interruptedAt: number +} + +export interface PlaybackRejectEvent { + item: PlaybackItem + reason: string +} + +export type IntentBehavior = 'queue' | 'interrupt' | 'replace' + +export interface IntentOptions { + intentId?: string + streamId?: string + priority?: PriorityLevel | number + ownerId?: string + behavior?: IntentBehavior +} + +export interface IntentHandle { + intentId: string + streamId: string + priority: number + ownerId?: string + writeLiteral: (text: string) => void + writeSpecial: (special: string) => void + writeFlush: () => void + end: () => void + cancel: (reason?: string) => void + stream: ReadableStream +} + +export interface SpeechPipelineEvents { + onSegment: (segment: TextSegment) => void + onSpecial: (segment: TextSegment) => void + onTtsRequest: (request: TtsRequest) => void + onTtsResult: (result: TtsResult) => void + onPlaybackStart: (event: PlaybackStartEvent) => void + onPlaybackEnd: (event: PlaybackEndEvent) => void + onPlaybackInterrupt: (event: PlaybackInterruptEvent) => void + onPlaybackReject: (event: PlaybackRejectEvent) => void + onIntentStart: (intentId: string) => void + onIntentEnd: (intentId: string) => void + onIntentCancel: (intentId: string, reason?: string) => void +} + +export interface LoggerLike { + debug: (message: string, ...args: unknown[]) => void + info: (message: string, ...args: unknown[]) => void + warn: (message: string, ...args: unknown[]) => void + error: (message: string, ...args: unknown[]) => void +} diff --git a/packages/stage-ui/package.json b/packages/stage-ui/package.json index 2969dfc29..ea63f1fcc 100644 --- a/packages/stage-ui/package.json +++ b/packages/stage-ui/package.json @@ -61,6 +61,7 @@ "@proj-airi/font-xiaolai": "workspace:^", "@proj-airi/i18n": "workspace:^", "@proj-airi/model-driver-lipsync": "workspace:^", + "@proj-airi/pipelines-audio": "workspace:^", "@proj-airi/server-sdk": "workspace:^", "@proj-airi/stage-shared": "workspace:^", "@proj-airi/stage-ui-live2d": "workspace:^", @@ -151,6 +152,7 @@ "@moeru/std": "catalog:", "@pinia/testing": "catalog:", "@proj-airi/lobe-icons": "^1.0.18", + "@proj-airi/stream-kit": "workspace:*", "@proj-airi/vite-plugin-warpdrive": "workspace:*", "@types/audioworklet": "catalog:", "@types/culori": "^4.0.1", diff --git a/packages/stage-ui/src/components/scenarios/providers/speech-streaming-playground.vue b/packages/stage-ui/src/components/scenarios/providers/speech-streaming-playground.vue index d6daec6b1..e2b0b89b9 100644 --- a/packages/stage-ui/src/components/scenarios/providers/speech-streaming-playground.vue +++ b/packages/stage-ui/src/components/scenarios/providers/speech-streaming-playground.vue @@ -1,13 +1,11 @@