fix(stage-tamagotchi,stream-kit,pipelines-audio): couldn't handle spark:notify and has elegant way of handling tts & playback

This commit is contained in:
Neko Ayaka
2026-01-11 00:13:14 +08:00
parent 91a95ab3db
commit 3c68172442
34 changed files with 2227 additions and 709 deletions
+2
View File
@@ -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",
@@ -1,10 +1,10 @@
<script setup lang="ts">
import type { TTSChunkItem } from '@proj-airi/stage-ui/utils/tts'
import type { ChatProvider, SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils'
import { createPlaybackManager, createSpeechPipeline } from '@proj-airi/pipelines-audio'
import { ThreeScene } from '@proj-airi/stage-ui-three'
import { animations } from '@proj-airi/stage-ui-three/assets/vrm'
import { useDelayMessageQueue, useEmotionsMessageQueue, usePipelineCharacterSpeechPlaybackQueueStore, usePipelineWorkflowTextSegmentationStore } from '@proj-airi/stage-ui/composables/queues'
import { useDelayMessageQueue, useEmotionsMessageQueue } from '@proj-airi/stage-ui/composables/queues'
import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '@proj-airi/stage-ui/constants/emotions'
import { useAudioContext, useSpeakingStore } from '@proj-airi/stage-ui/stores/audio'
@@ -13,35 +13,25 @@ import { useConsciousnessStore } from '@proj-airi/stage-ui/stores/modules/consci
import { useSpeechStore } from '@proj-airi/stage-ui/stores/modules/speech'
import { useProvidersStore } from '@proj-airi/stage-ui/stores/providers'
import { useSettings } from '@proj-airi/stage-ui/stores/settings'
import { createQueue } from '@proj-airi/stage-ui/utils/queue'
import { createQueue } from '@proj-airi/stream-kit'
import { generateSpeech } from '@xsai/generate-speech'
import { storeToRefs } from 'pinia'
import { computed, onMounted, onUnmounted, ref } from 'vue'
// VRM scene refs
const sceneRef = ref<InstanceType<typeof ThreeScene>>()
const currentAudioSource = ref<AudioBufferSourceNode>()
// Playback + lip sync (VRM uses currentAudioSource)
const characterSpeechPlaybackQueue = usePipelineCharacterSpeechPlaybackQueueStore()
const { connectAudioContext, connectAudioAnalyser, clearAll, onPlaybackStarted, onPlaybackFinished } = characterSpeechPlaybackQueue
const { currentAudioSource, playbackQueue } = storeToRefs(characterSpeechPlaybackQueue)
// Audio context / analyser
const { audioContext } = useAudioContext()
connectAudioContext(audioContext)
const audioAnalyser = ref<AnalyserNode>()
function setupAnalyser() {
if (!audioAnalyser.value) {
if (!audioAnalyser.value)
audioAnalyser.value = audioContext.createAnalyser()
connectAudioAnalyser(audioAnalyser.value)
}
}
// Settings + force VRM model
const settingsStore = useSettings()
const { stageModelRenderer, stageModelSelected, stageModelSelectedUrl, stageViewControlsEnabled } = storeToRefs(settingsStore)
onMounted(async () => {
// Preserve existing VRM selection if available; otherwise fall back to preset VRM
const needsFallback = !stageModelSelectedUrl.value || stageModelRenderer.value !== 'vrm'
if (needsFallback)
stageModelSelected.value = 'preset-vrm-1'
@@ -50,30 +40,23 @@ onMounted(async () => {
setupAnalyser()
})
// Speech
const providersStore = useProvidersStore()
const speechStore = useSpeechStore()
const { activeSpeechProvider, activeSpeechVoice, activeSpeechModel, ssmlEnabled, pitch } = storeToRefs(speechStore)
const consciousnessStore = useConsciousnessStore()
const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
// Text segmentation
const textSegmentationStore = usePipelineWorkflowTextSegmentationStore()
const { onTextSegmented, clearHooks: clearTextSegmentationHooks } = textSegmentationStore
const { textSegmentationQueue } = storeToRefs(textSegmentationStore)
clearTextSegmentationHooks()
// Emotion/delay queues (special tokens)
const delaysQueue = useDelayMessageQueue()
const emotionMessageQueue = useEmotionsMessageQueue(createQueue({ handlers: [] }))
emotionMessageQueue.on('enqueue', (token) => {
log(` - special 入队:${token}`)
})
emotionMessageQueue.on('dequeue', (token) => {
log(`special 出队处理:${token}`)
})
// State
const { mouthOpenSize } = storeToRefs(useSpeakingStore())
const nowSpeaking = ref(false)
const currentMotion = ref<{ group: string }>({ group: EmotionThinkMotionName })
@@ -97,29 +80,69 @@ function log(line: string) {
logLines.value = [line, ...logLines.value].slice(0, 50)
}
// TTS generation handler
async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
try {
const playbackManager = createPlaybackManager<AudioBuffer>({
play: (item, signal) => {
return new Promise((resolve) => {
const source = audioContext.createBufferSource()
source.buffer = item.audio
source.connect(audioContext.destination)
if (audioAnalyser.value)
source.connect(audioAnalyser.value)
currentAudioSource.value = source
const stopPlayback = () => {
try {
source.stop()
source.disconnect()
}
catch {}
if (currentAudioSource.value === source)
currentAudioSource.value = undefined
resolve()
}
if (signal.aborted) {
stopPlayback()
return
}
signal.addEventListener('abort', stopPlayback, { once: true })
source.onended = () => {
signal.removeEventListener('abort', stopPlayback)
stopPlayback()
}
source.start(0)
})
},
maxVoices: 1,
maxVoicesPerOwner: 1,
overflowPolicy: 'queue',
ownerOverflowPolicy: 'steal-oldest',
})
const speechPipeline = createSpeechPipeline<AudioBuffer>({
tts: async (request, signal) => {
if (signal.aborted)
return null
if (!activeSpeechProvider.value || !activeSpeechVoice.value) {
console.warn('No active speech provider configured')
return
return null
}
const provider = await providersStore.getProviderInstance(activeSpeechProvider.value) as SpeechProviderWithExtraOptions<string, any>
if (!provider) {
console.error('Failed to initialize speech provider')
return
}
if (ctx.data.chunk === '' && !ctx.data.special)
return
if (ctx.data.chunk === '' && ctx.data.special) {
// log(`特殊标记:${ctx.data.special}`)
emotionMessageQueue.enqueue(ctx.data.special)
return
return null
}
if (!request.text && !request.special)
return null
const providerConfig = providersStore.getProviderConfig(activeSpeechProvider.value)
const input = ssmlEnabled.value
? speechStore.generateSSML(ctx.data.chunk, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: ctx.data.chunk
? speechStore.generateSSML(request.text, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: request.text
const res = await generateSpeech({
...provider.speech(activeSpeechModel.value, providerConfig),
@@ -127,24 +150,38 @@ async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
voice: activeSpeechVoice.value.id,
})
const audioBuffer = await audioContext.decodeAudioData(res)
log(` - 排队:${ctx.data.chunk}${ctx.data.special ? ` [special: ${ctx.data.special}]` : ''}`)
playbackQueue.value.enqueue({ audioBuffer, text: ctx.data.chunk, special: ctx.data.special })
}
catch (error) {
console.error('Speech generation failed:', error)
}
}
if (signal.aborted)
return null
const ttsQueue = createQueue<TTSChunkItem>({
handlers: [
handleSpeechGeneration,
],
log(` - 排队:${request.text}${request.special ? ` [special: ${request.special}]` : ''}`)
return audioContext.decodeAudioData(res)
},
playback: playbackManager,
})
// text segmentation hooks
onTextSegmented((chunkItem) => {
ttsQueue.enqueue(chunkItem)
speechPipeline.on('onSpecial', (segment) => {
if (segment.special)
emotionMessageQueue.enqueue(segment.special)
})
playbackManager.onStart(({ item }) => {
nowSpeaking.value = true
log(`播放开始:${item.text}`)
})
playbackManager.onEnd(({ item }) => {
nowSpeaking.value = false
mouthOpenSize.value = 0
if (item.special) {
log(`播放结束,special: ${item.special}`)
const motion = EMOTION_EmotionMotionName_value[item.special as keyof typeof EMOTION_EmotionMotionName_value]
const expression = EMOTION_VRMExpressionName_value[item.special as keyof typeof EMOTION_VRMExpressionName_value]
if (motion)
currentMotion.value = { group: motion }
if (expression)
sceneRef.value?.setExpression(expression)
}
})
async function sendChat() {
@@ -175,17 +212,19 @@ function resetChat() {
chatStore.cleanupMessages()
chatInput.value = ''
logLines.value = []
clearAll()
playbackManager.stopAll('reset')
}
// Chat hooks (reuse Stage pipeline but Live2D removed)
const { onBeforeMessageComposed, onBeforeSend, onTokenLiteral, onTokenSpecial, onStreamEnd } = chatStore
const { onBeforeMessageComposed, onBeforeSend, onTokenLiteral, onTokenSpecial, onStreamEnd, onAssistantResponseEnd } = chatStore
const chatHookCleanups: Array<() => void> = []
let currentIntent: ReturnType<typeof speechPipeline.openIntent> | null = null
chatHookCleanups.push(onBeforeMessageComposed(async () => {
clearAll()
playbackManager.stopAll('new-message')
setupAnalyser()
logLines.value = []
currentIntent?.cancel('new-message')
currentIntent = speechPipeline.openIntent({ priority: 'normal', behavior: 'queue' })
}))
chatHookCleanups.push(onBeforeSend(async () => {
@@ -193,40 +232,26 @@ chatHookCleanups.push(onBeforeSend(async () => {
}))
chatHookCleanups.push(onTokenLiteral(async (literal) => {
textSegmentationQueue.value.enqueue({ type: 'literal', value: literal })
currentIntent?.writeLiteral(literal)
}))
chatHookCleanups.push(onTokenSpecial(async (special) => {
textSegmentationQueue.value.enqueue({ type: 'special', value: special })
currentIntent?.writeSpecial(special)
}))
chatHookCleanups.push(onStreamEnd(async () => {
delaysQueue.enqueue(llmInferenceEndToken)
currentIntent?.writeFlush()
}))
// Wire playback to VRM + logs
onPlaybackFinished(({ special }) => {
nowSpeaking.value = false
mouthOpenSize.value = 0
if (special) {
log(`播放结束,special: ${special}`)
const motion = EMOTION_EmotionMotionName_value[special as keyof typeof EMOTION_EmotionMotionName_value]
const expression = EMOTION_VRMExpressionName_value[special as keyof typeof EMOTION_VRMExpressionName_value]
if (motion)
currentMotion.value = { group: motion }
if (expression)
sceneRef.value?.setExpression(expression)
}
})
onPlaybackStarted(({ text }) => {
nowSpeaking.value = true
log(`播放开始:${text}`)
})
chatHookCleanups.push(onAssistantResponseEnd(async () => {
currentIntent?.end()
currentIntent = null
}))
onUnmounted(() => {
chatHookCleanups.forEach(dispose => dispose?.())
clearAll()
playbackManager.stopAll('unmount')
})
</script>
@@ -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<ChatProvider>(activeProvider.value), [])
+2
View File
@@ -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",
@@ -1,10 +1,10 @@
<script setup lang="ts">
import type { TTSChunkItem } from '@proj-airi/stage-ui/utils/tts'
import type { ChatProvider, SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils'
import { createPlaybackManager, createSpeechPipeline } from '@proj-airi/pipelines-audio'
import { ThreeScene } from '@proj-airi/stage-ui-three'
import { animations } from '@proj-airi/stage-ui-three/assets/vrm'
import { useDelayMessageQueue, useEmotionsMessageQueue, usePipelineCharacterSpeechPlaybackQueueStore, usePipelineWorkflowTextSegmentationStore } from '@proj-airi/stage-ui/composables/queues'
import { useDelayMessageQueue, useEmotionsMessageQueue } from '@proj-airi/stage-ui/composables/queues'
import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '@proj-airi/stage-ui/constants/emotions'
import { useAudioContext, useSpeakingStore } from '@proj-airi/stage-ui/stores/audio'
@@ -13,35 +13,25 @@ import { useConsciousnessStore } from '@proj-airi/stage-ui/stores/modules/consci
import { useSpeechStore } from '@proj-airi/stage-ui/stores/modules/speech'
import { useProvidersStore } from '@proj-airi/stage-ui/stores/providers'
import { useSettings } from '@proj-airi/stage-ui/stores/settings'
import { createQueue } from '@proj-airi/stage-ui/utils/queue'
import { createQueue } from '@proj-airi/stream-kit'
import { generateSpeech } from '@xsai/generate-speech'
import { storeToRefs } from 'pinia'
import { computed, onMounted, onUnmounted, ref } from 'vue'
// VRM scene refs
const sceneRef = ref<InstanceType<typeof ThreeScene>>()
const currentAudioSource = ref<AudioBufferSourceNode>()
// Playback + lip sync (VRM uses currentAudioSource)
const characterSpeechPlaybackQueue = usePipelineCharacterSpeechPlaybackQueueStore()
const { connectAudioContext, connectAudioAnalyser, clearAll, onPlaybackStarted, onPlaybackFinished } = characterSpeechPlaybackQueue
const { currentAudioSource, playbackQueue } = storeToRefs(characterSpeechPlaybackQueue)
// Audio context / analyser
const { audioContext } = useAudioContext()
connectAudioContext(audioContext)
const audioAnalyser = ref<AnalyserNode>()
function setupAnalyser() {
if (!audioAnalyser.value) {
if (!audioAnalyser.value)
audioAnalyser.value = audioContext.createAnalyser()
connectAudioAnalyser(audioAnalyser.value)
}
}
// Settings + force VRM model
const settingsStore = useSettings()
const { stageModelRenderer, stageModelSelected, stageModelSelectedUrl, stageViewControlsEnabled } = storeToRefs(settingsStore)
onMounted(async () => {
// Preserve existing VRM selection if available; otherwise fall back to preset VRM
const needsFallback = !stageModelSelectedUrl.value || stageModelRenderer.value !== 'vrm'
if (needsFallback)
stageModelSelected.value = 'preset-vrm-1'
@@ -50,30 +40,23 @@ onMounted(async () => {
setupAnalyser()
})
// Speech
const providersStore = useProvidersStore()
const speechStore = useSpeechStore()
const { activeSpeechProvider, activeSpeechVoice, activeSpeechModel, ssmlEnabled, pitch } = storeToRefs(speechStore)
const consciousnessStore = useConsciousnessStore()
const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
// Text segmentation
const textSegmentationStore = usePipelineWorkflowTextSegmentationStore()
const { onTextSegmented, clearHooks: clearTextSegmentationHooks } = textSegmentationStore
const { textSegmentationQueue } = storeToRefs(textSegmentationStore)
clearTextSegmentationHooks()
// Emotion/delay queues (special tokens)
const delaysQueue = useDelayMessageQueue()
const emotionMessageQueue = useEmotionsMessageQueue(createQueue({ handlers: [] }))
emotionMessageQueue.on('enqueue', (token) => {
log(` - special 入队:${token}`)
})
emotionMessageQueue.on('dequeue', (token) => {
log(`special 出队处理:${token}`)
})
// State
const { mouthOpenSize } = storeToRefs(useSpeakingStore())
const nowSpeaking = ref(false)
const currentMotion = ref<{ group: string }>({ group: EmotionThinkMotionName })
@@ -97,29 +80,69 @@ function log(line: string) {
logLines.value = [line, ...logLines.value].slice(0, 50)
}
// TTS generation handler
async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
try {
const playbackManager = createPlaybackManager<AudioBuffer>({
play: (item, signal) => {
return new Promise((resolve) => {
const source = audioContext.createBufferSource()
source.buffer = item.audio
source.connect(audioContext.destination)
if (audioAnalyser.value)
source.connect(audioAnalyser.value)
currentAudioSource.value = source
const stopPlayback = () => {
try {
source.stop()
source.disconnect()
}
catch {}
if (currentAudioSource.value === source)
currentAudioSource.value = undefined
resolve()
}
if (signal.aborted) {
stopPlayback()
return
}
signal.addEventListener('abort', stopPlayback, { once: true })
source.onended = () => {
signal.removeEventListener('abort', stopPlayback)
stopPlayback()
}
source.start(0)
})
},
maxVoices: 1,
maxVoicesPerOwner: 1,
overflowPolicy: 'queue',
ownerOverflowPolicy: 'steal-oldest',
})
const speechPipeline = createSpeechPipeline<AudioBuffer>({
tts: async (request, signal) => {
if (signal.aborted)
return null
if (!activeSpeechProvider.value || !activeSpeechVoice.value) {
console.warn('No active speech provider configured')
return
return null
}
const provider = await providersStore.getProviderInstance(activeSpeechProvider.value) as SpeechProviderWithExtraOptions<string, any>
if (!provider) {
console.error('Failed to initialize speech provider')
return
}
if (ctx.data.chunk === '' && !ctx.data.special)
return
if (ctx.data.chunk === '' && ctx.data.special) {
// log(`特殊标记:${ctx.data.special}`)
emotionMessageQueue.enqueue(ctx.data.special)
return
return null
}
if (!request.text && !request.special)
return null
const providerConfig = providersStore.getProviderConfig(activeSpeechProvider.value)
const input = ssmlEnabled.value
? speechStore.generateSSML(ctx.data.chunk, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: ctx.data.chunk
? speechStore.generateSSML(request.text, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: request.text
const res = await generateSpeech({
...provider.speech(activeSpeechModel.value, providerConfig),
@@ -127,24 +150,38 @@ async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
voice: activeSpeechVoice.value.id,
})
const audioBuffer = await audioContext.decodeAudioData(res)
log(` - 排队:${ctx.data.chunk}${ctx.data.special ? ` [special: ${ctx.data.special}]` : ''}`)
playbackQueue.value.enqueue({ audioBuffer, text: ctx.data.chunk, special: ctx.data.special })
}
catch (error) {
console.error('Speech generation failed:', error)
}
}
if (signal.aborted)
return null
const ttsQueue = createQueue<TTSChunkItem>({
handlers: [
handleSpeechGeneration,
],
log(` - 排队:${request.text}${request.special ? ` [special: ${request.special}]` : ''}`)
return audioContext.decodeAudioData(res)
},
playback: playbackManager,
})
// text segmentation hooks
onTextSegmented((chunkItem) => {
ttsQueue.enqueue(chunkItem)
speechPipeline.on('onSpecial', (segment) => {
if (segment.special)
emotionMessageQueue.enqueue(segment.special)
})
playbackManager.onStart(({ item }) => {
nowSpeaking.value = true
log(`播放开始:${item.text}`)
})
playbackManager.onEnd(({ item }) => {
nowSpeaking.value = false
mouthOpenSize.value = 0
if (item.special) {
log(`播放结束,special: ${item.special}`)
const motion = EMOTION_EmotionMotionName_value[item.special as keyof typeof EMOTION_EmotionMotionName_value]
const expression = EMOTION_VRMExpressionName_value[item.special as keyof typeof EMOTION_VRMExpressionName_value]
if (motion)
currentMotion.value = { group: motion }
if (expression)
sceneRef.value?.setExpression(expression)
}
})
async function sendChat() {
@@ -175,17 +212,19 @@ function resetChat() {
chatStore.cleanupMessages()
chatInput.value = ''
logLines.value = []
clearAll()
playbackManager.stopAll('reset')
}
// Chat hooks (reuse Stage pipeline but Live2D removed)
const { onBeforeMessageComposed, onBeforeSend, onTokenLiteral, onTokenSpecial, onStreamEnd } = chatStore
const { onBeforeMessageComposed, onBeforeSend, onTokenLiteral, onTokenSpecial, onStreamEnd, onAssistantResponseEnd } = chatStore
const chatHookCleanups: Array<() => void> = []
let currentIntent: ReturnType<typeof speechPipeline.openIntent> | null = null
chatHookCleanups.push(onBeforeMessageComposed(async () => {
clearAll()
playbackManager.stopAll('new-message')
setupAnalyser()
logLines.value = []
currentIntent?.cancel('new-message')
currentIntent = speechPipeline.openIntent({ priority: 'normal', behavior: 'queue' })
}))
chatHookCleanups.push(onBeforeSend(async () => {
@@ -193,40 +232,26 @@ chatHookCleanups.push(onBeforeSend(async () => {
}))
chatHookCleanups.push(onTokenLiteral(async (literal) => {
textSegmentationQueue.value.enqueue({ type: 'literal', value: literal })
currentIntent?.writeLiteral(literal)
}))
chatHookCleanups.push(onTokenSpecial(async (special) => {
textSegmentationQueue.value.enqueue({ type: 'special', value: special })
currentIntent?.writeSpecial(special)
}))
chatHookCleanups.push(onStreamEnd(async () => {
delaysQueue.enqueue(llmInferenceEndToken)
currentIntent?.writeFlush()
}))
// Wire playback to VRM + logs
onPlaybackFinished(({ special }) => {
nowSpeaking.value = false
mouthOpenSize.value = 0
if (special) {
log(`播放结束,special: ${special}`)
const motion = EMOTION_EmotionMotionName_value[special as keyof typeof EMOTION_EmotionMotionName_value]
const expression = EMOTION_VRMExpressionName_value[special as keyof typeof EMOTION_VRMExpressionName_value]
if (motion)
currentMotion.value = { group: motion }
if (expression)
sceneRef.value?.setExpression(expression)
}
})
onPlaybackStarted(({ text }) => {
nowSpeaking.value = true
log(`播放开始:${text}`)
})
chatHookCleanups.push(onAssistantResponseEnd(async () => {
currentIntent?.end()
currentIntent = null
}))
onUnmounted(() => {
chatHookCleanups.forEach(dispose => dispose?.())
clearAll()
playbackManager.stopAll('unmount')
})
</script>
+1
View File
@@ -33,6 +33,7 @@
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@moeru/eventa": "^1.0.0-alpha.10",
"@moeru/std": "catalog:",
"clustr": "^1.0.2"
},
+39
View File
@@ -0,0 +1,39 @@
import type {
PlaybackEndEvent,
PlaybackInterruptEvent,
PlaybackRejectEvent,
PlaybackStartEvent,
TextSegment,
TtsRequest,
TtsResult,
} from './types'
import { defineEventa } from '@moeru/eventa'
export const speechSegmentEvent = defineEventa<TextSegment>('proj-airi:pipelines:output:speech:segment')
export const speechSpecialEvent = defineEventa<TextSegment>('proj-airi:pipelines:output:speech:special')
export const speechTtsRequestEvent = defineEventa<TtsRequest>('proj-airi:pipelines:output:speech:tts-request')
export const speechTtsResultEvent = defineEventa<TtsResult<any>>('proj-airi:pipelines:output:speech:tts-result')
export const speechPlaybackStartEvent = defineEventa<PlaybackStartEvent<any>>('proj-airi:pipelines:output:speech:playback-start')
export const speechPlaybackEndEvent = defineEventa<PlaybackEndEvent<any>>('proj-airi:pipelines:output:speech:playback-end')
export const speechPlaybackInterruptEvent = defineEventa<PlaybackInterruptEvent<any>>('proj-airi:pipelines:output:speech:playback-interrupt')
export const speechPlaybackRejectEvent = defineEventa<PlaybackRejectEvent<any>>('proj-airi:pipelines:output:speech:playback-reject')
export const speechIntentStartEvent = defineEventa<string>('proj-airi:pipelines:output:speech:intent-start')
export const speechIntentEndEvent = defineEventa<string>('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
+7 -1
View File
@@ -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'
@@ -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<TAudio> {
play: (item: PlaybackItem<TAudio>, signal: AbortSignal) => Promise<void>
maxVoices?: number
maxVoicesPerOwner?: number
overflowPolicy?: OverflowPolicy
ownerOverflowPolicy?: OwnerOverflowPolicy
}
export function createPlaybackManager<TAudio>(options: PlaybackManagerOptions<TAudio>) {
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<string, {
item: PlaybackItem<TAudio>
controller: AbortController
startedAt: number
}>()
const waiting: Array<{ item: PlaybackItem<TAudio>, enqueuedAt: number }> = []
const listeners = {
start: [] as Array<(event: PlaybackStartEvent<TAudio>) => void>,
end: [] as Array<(event: PlaybackEndEvent<TAudio>) => void>,
interrupt: [] as Array<(event: PlaybackInterruptEvent<TAudio>) => void>,
reject: [] as Array<(event: PlaybackRejectEvent<TAudio>) => void>,
}
function onStart(listener: (event: PlaybackStartEvent<TAudio>) => void) {
listeners.start.push(listener)
}
function onEnd(listener: (event: PlaybackEndEvent<TAudio>) => void) {
listeners.end.push(listener)
}
function onInterrupt(listener: (event: PlaybackInterruptEvent<TAudio>) => void) {
listeners.interrupt.push(listener)
}
function onReject(listener: (event: PlaybackRejectEvent<TAudio>) => void) {
listeners.reject.push(listener)
}
function emitStart(item: PlaybackItem<TAudio>) {
const event = { item, startedAt: Date.now() }
listeners.start.forEach(listener => listener(event))
}
function emitEnd(item: PlaybackItem<TAudio>) {
const event = { item, endedAt: Date.now() }
listeners.end.forEach(listener => listener(event))
}
function emitInterrupt(item: PlaybackItem<TAudio>, reason: string) {
const event = { item, reason, interruptedAt: Date.now() }
listeners.interrupt.forEach(listener => listener(event))
}
function emitReject(item: PlaybackItem<TAudio>, 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<TAudio>, 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<TAudio>, 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<TAudio>, controller: AbortController }, reason: string) {
entry.controller.abort(reason)
active.delete(entry.item.id)
emitInterrupt(entry.item, reason)
}
function canStart(item: PlaybackItem<TAudio>) {
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<TAudio>) {
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<TAudio>, 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<TAudio>) {
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,
}
}
+28
View File
@@ -0,0 +1,28 @@
import type { PriorityLevel, PriorityResolver } from './types'
const DEFAULT_LEVELS: Record<PriorityLevel, number> = {
critical: 300,
high: 200,
normal: 100,
low: 0,
}
export function createPriorityResolver(levels?: Partial<Record<PriorityLevel, number>>): 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
}
@@ -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<TtsInputChunk, void, unknown> {
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<string, any> | undefined
let afterNext: IteratorResult<string, any> | 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> | 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<TextToken>,
meta: { streamId: string, intentId: string },
options?: TtsInputChunkOptions,
) {
const { stream, write, close, error } = createPushStream<TextSegment>()
const pendingSpecials: string[] = []
const encoder = new TextEncoder()
const { stream: byteStream, write: writeBytes, close: closeBytes, error: errorBytes } = createPushStream<Uint8Array>()
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
}
@@ -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<TAudio> {
tts: (request: TtsRequest, signal: AbortSignal) => Promise<TAudio | null>
playback: {
schedule: (item: PlaybackItem<TAudio>) => void
stopAll: (reason: string) => void
stopByIntent: (intentId: string, reason: string) => void
stopByOwner: (ownerId: string, reason: string) => void
onStart: (listener: (event: { item: PlaybackItem<TAudio>, startedAt: number }) => void) => void
onEnd: (listener: (event: { item: PlaybackItem<TAudio>, endedAt: number }) => void) => void
onInterrupt: (listener: (event: { item: PlaybackItem<TAudio>, reason: string, interruptedAt: number }) => void) => void
onReject: (listener: (event: { item: PlaybackItem<TAudio>, reason: string }) => void) => void
}
logger?: LoggerLike
priority?: ReturnType<typeof createPriorityResolver>
segmenter?: (tokens: ReadableStream<TextToken>, meta: { streamId: string, intentId: string }) => ReadableStream<TextSegment>
}
interface IntentState {
intentId: string
streamId: string
priority: number
ownerId?: string
behavior: 'queue' | 'interrupt' | 'replace'
createdAt: number
controller: AbortController
stream: ReadableStream<TextToken>
closeStream: () => void
canceled: boolean
}
function createId(prefix: string) {
return `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`
}
export function createSpeechPipeline<TAudio>(options: SpeechPipelineOptions<TAudio>) {
const logger = options.logger ?? console
const priorityResolver = options.priority ?? createPriorityResolver()
const segmenter = options.segmenter ?? createTtsSegmentStream
const context = createContext()
const intents = new Map<string, IntentState>()
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<TAudio> = {
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<TextToken>()
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<K extends SpeechPipelineEventName>(event: K, listener: SpeechPipelineEvents<TAudio>[K]) {
return context.on(speechPipelineEventMap[event] as Eventa<any>, (payload) => {
listener(payload?.body ?? payload)
})
},
}
}
+61
View File
@@ -0,0 +1,61 @@
export interface StreamController<T> {
stream: ReadableStream<T>
write: (value: T) => void
close: () => void
error: (err: unknown) => void
isClosed: () => boolean
}
export function createPushStream<T>(): StreamController<T> {
let closed = false
let controller: ReadableStreamDefaultController<T> | null = null
const stream = new ReadableStream<T>({
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<T>(stream: ReadableStream<T>, handler: (value: T) => Promise<void> | 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()
}
}
+122
View File
@@ -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<TAudio> {
streamId: string
intentId: string
segmentId: string
text: string
special: string | null
audio: TAudio
createdAt: number
}
export interface PlaybackItem<TAudio> {
id: string
streamId: string
intentId: string
segmentId: string
ownerId?: string
priority: number
text: string
special: string | null
audio: TAudio
createdAt: number
}
export interface PlaybackStartEvent<TAudio> {
item: PlaybackItem<TAudio>
startedAt: number
}
export interface PlaybackEndEvent<TAudio> {
item: PlaybackItem<TAudio>
endedAt: number
}
export interface PlaybackInterruptEvent<TAudio> {
item: PlaybackItem<TAudio>
reason: string
interruptedAt: number
}
export interface PlaybackRejectEvent<TAudio> {
item: PlaybackItem<TAudio>
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<TextToken>
}
export interface SpeechPipelineEvents<TAudio> {
onSegment: (segment: TextSegment) => void
onSpecial: (segment: TextSegment) => void
onTtsRequest: (request: TtsRequest) => void
onTtsResult: (result: TtsResult<TAudio>) => void
onPlaybackStart: (event: PlaybackStartEvent<TAudio>) => void
onPlaybackEnd: (event: PlaybackEndEvent<TAudio>) => void
onPlaybackInterrupt: (event: PlaybackInterruptEvent<TAudio>) => void
onPlaybackReject: (event: PlaybackRejectEvent<TAudio>) => 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
}
+2
View File
@@ -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",
@@ -1,13 +1,11 @@
<script setup lang="ts">
import type { TTSInputChunk } from '../../../utils/tts'
import { createQueue } from '@proj-airi/stream-kit'
import { animate } from 'animejs'
import { storeToRefs } from 'pinia'
import { ref } from 'vue'
import { usePipelineWorkflowTextSegmentationStore } from '../../../composables/queues'
import { useAudioContext } from '../../../stores/audio'
import { createQueue } from '../../../utils/queue'
import { chunkTTSInput } from '../../../utils/tts'
const props = defineProps<{
@@ -17,8 +15,6 @@ const props = defineProps<{
voice: string
}>()
const { onTextSegmented } = usePipelineWorkflowTextSegmentationStore()
const { textSegmentationQueue } = storeToRefs(usePipelineWorkflowTextSegmentationStore())
const { audioContext } = useAudioContext()
const nowSpeaking = ref(false)
const ttsInputChunks = ref<TTSInputChunk[]>([])
@@ -28,14 +24,10 @@ const audioQueue = createQueue<{ audioBuffer: AudioBuffer, text: string }>({
handlers: [
(ctx) => {
return new Promise((resolve) => {
// Create an AudioBufferSourceNode
const source = audioContext.createBufferSource()
source.buffer = ctx.data.audioBuffer
// Connect the source to the AudioContext's destination (the speakers)
source.connect(audioContext.destination)
// Start playing the audio
nowSpeaking.value = true
source.start(0)
source.onended = () => {
@@ -55,7 +47,6 @@ async function handleSpeechGeneration(ctx: { data: string }) {
const res = await props.generateSpeech(input, props.voice, false)
// Decode the ArrayBuffer into an AudioBuffer
const audioBuffer = await audioContext.decodeAudioData(res)
audioQueue.enqueue({ audioBuffer, text: ctx.data })
}
@@ -66,16 +57,17 @@ async function handleSpeechGeneration(ctx: { data: string }) {
const ttsQueue = createQueue<string>({ handlers: [handleSpeechGeneration] })
onTextSegmented((chunk) => {
ttsQueue.enqueue(chunk.chunk)
})
async function testStreaming() {
textSegmentationQueue.value.enqueue({ type: 'literal', value: props.text })
speechGenerationIndex.value = -1
for await (const chunk of chunkTTSInput(props.text, { boost: 1, minimumWords: 4, maximumWords: 12 })) {
if (!chunk.text)
continue
ttsQueue.enqueue(chunk.text)
}
}
async function testChunking() {
const chunks = []
const chunks: TTSInputChunk[] = []
const stream = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(new TextEncoder().encode(props.text))
+123 -87
View File
@@ -5,34 +5,35 @@ import type { Profile } from '@proj-airi/model-driver-lipsync/shared/wlipsync'
import type { SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils'
import type { UnElevenLabsOptions } from 'unspeech'
import type { TextSegmentationItem } from '../../composables/queues'
import type { Emotion } from '../../constants/emotions'
import type { TTSChunkItem } from '../../utils/tts'
import { drizzle } from '@proj-airi/drizzle-duckdb-wasm'
import { getImportUrlBundles } from '@proj-airi/drizzle-duckdb-wasm/bundles/import-url-browser'
import { createLive2DLipSync } from '@proj-airi/model-driver-lipsync'
import { wlipsyncProfile } from '@proj-airi/model-driver-lipsync/shared/wlipsync'
import { createPlaybackManager, createSpeechPipeline } from '@proj-airi/pipelines-audio'
import { Live2DScene, useLive2d } from '@proj-airi/stage-ui-live2d'
import { ThreeScene, useModelStore } from '@proj-airi/stage-ui-three'
import { animations } from '@proj-airi/stage-ui-three/assets/vrm'
import { createQueue } from '@proj-airi/stream-kit'
import { useBroadcastChannel } from '@vueuse/core'
// import { createTransformers } from '@xsai-transformers/embed'
// import embedWorkerURL from '@xsai-transformers/embed/worker?worker&url'
// import { embed } from '@xsai/embed'
import { generateSpeech } from '@xsai/generate-speech'
import { storeToRefs } from 'pinia'
import { onMounted, onUnmounted, ref } from 'vue'
import { computed, onMounted, onUnmounted, ref } from 'vue'
import { useDelayMessageQueue, useEmotionsMessageQueue, usePipelineCharacterSpeechPlaybackQueueStore, usePipelineWorkflowTextSegmentationStore } from '../../composables/queues'
import { useDelayMessageQueue, useEmotionsMessageQueue } from '../../composables/queues'
import { llmInferenceEndToken } from '../../constants'
import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions'
import { useAudioContext, useSpeakingStore } from '../../stores/audio'
import { useChatStore } from '../../stores/chat'
import { useAiriCardStore } from '../../stores/modules'
import { useSpeechStore } from '../../stores/modules/speech'
import { useProvidersStore } from '../../stores/providers'
import { useSettings } from '../../stores/settings'
import { createQueue } from '../../utils/queue'
import { useSpeechRuntimeStore } from '../../stores/speech-runtime'
withDefaults(defineProps<{
paused?: boolean
@@ -50,28 +51,6 @@ const db = ref<DuckDBWasmDrizzleDatabase>()
const vrmViewerRef = ref<InstanceType<typeof ThreeScene>>()
const live2dSceneRef = ref<InstanceType<typeof Live2DScene>>()
const textSegmentationStore = usePipelineWorkflowTextSegmentationStore()
const { onTextSegmented, clearHooks: clearTextSegmentationHooks } = textSegmentationStore
const { textSegmentationQueue } = storeToRefs(textSegmentationStore)
// WORKAROUND: clear previous hooks to avoid duplicate calls
// due to re-mounting of this component when switching routes and stages.
// We may need to find a way to better orchestrate the lifecycle of the event
// listeners within specific scopes, e.g., perhaps, addEventListener with
// group tag, then we can remove them all once, or perhaps, we could implement
// every non-deterministic onXXX register function to remove registered listeners
// when the onUnmounted lifecycle hook is called.
// Another possible approach but not really work for every cases (such as character
// pipeline here, we may have multiple characters? Or multiple chat instances?, etc.)
// is to orchestrate the lifecycle of events for specific Character, sub-module like
// Dreaming procedure, etc. in each entity's own store with all lifecycle contained.
// We need better pattern (maybe learn from game engine) to power this kind of pipeline/
// event-driven workflow to avoid unexpected behaviors while maintain flexibility.
clearTextSegmentationHooks()
const characterSpeechPlaybackQueue = usePipelineCharacterSpeechPlaybackQueueStore()
const { connectAudioContext, connectAudioAnalyser, connectLipSyncNode, clearAll, onPlaybackStarted, onPlaybackFinished } = characterSpeechPlaybackQueue
const { currentAudioSource, playbackQueue } = storeToRefs(characterSpeechPlaybackQueue)
const settingsStore = useSettings()
const {
stageModelRenderer,
@@ -88,7 +67,7 @@ const {
} = storeToRefs(settingsStore)
const { mouthOpenSize } = storeToRefs(useSpeakingStore())
const { audioContext } = useAudioContext()
connectAudioContext(audioContext)
const currentAudioSource = ref<AudioBufferSourceNode>()
const { onBeforeMessageComposed, onBeforeSend, onTokenLiteral, onTokenSpecial, onStreamEnd, onAssistantResponseEnd } = useChatStore()
const chatHookCleanups: Array<() => void> = []
@@ -138,8 +117,11 @@ const lipSyncLoopId = ref<number>()
const live2dLipSync = ref<Live2DLipSync>()
const live2dLipSyncOptions: Live2DLipSyncOptions = { mouthUpdateIntervalMs: 50, mouthLerpWindowMs: 50 }
const { activeCard } = storeToRefs(useAiriCardStore())
const speechStore = useSpeechStore()
const { ssmlEnabled, activeSpeechProvider, activeSpeechModel, activeSpeechVoice, pitch } = storeToRefs(speechStore)
const activeCardId = computed(() => activeCard.value?.name ?? 'default')
const speechRuntimeStore = useSpeechRuntimeStore()
const { currentMotion } = storeToRefs(useLive2d())
@@ -178,43 +160,85 @@ function playSpecialToken(special: string) {
delaysQueue.enqueue(special)
emotionMessageContentQueue.enqueue(special)
}
onPlaybackFinished(({ special }) => {
playSpecialToken(special)
const lipSyncNode = ref<AudioNode>()
const playbackManager = createPlaybackManager<AudioBuffer>({
play: (item, signal) => {
return new Promise((resolve) => {
if (!audioContext) {
resolve()
return
}
const source = audioContext.createBufferSource()
currentAudioSource.value = source
source.buffer = item.audio
source.connect(audioContext.destination)
if (audioAnalyser.value)
source.connect(audioAnalyser.value)
if (lipSyncNode.value)
source.connect(lipSyncNode.value)
const stopPlayback = () => {
try {
source.stop()
source.disconnect()
}
catch {}
if (currentAudioSource.value === source)
currentAudioSource.value = undefined
resolve()
}
if (signal.aborted) {
stopPlayback()
return
}
signal.addEventListener('abort', stopPlayback, { once: true })
source.onended = () => {
signal.removeEventListener('abort', stopPlayback)
stopPlayback()
}
source.start(0)
})
},
maxVoices: 1,
maxVoicesPerOwner: 1,
overflowPolicy: 'queue',
ownerOverflowPolicy: 'steal-oldest',
})
async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
try {
const speechPipeline = createSpeechPipeline<AudioBuffer>({
tts: async (request, signal) => {
if (signal.aborted)
return null
if (!activeSpeechProvider.value) {
console.warn('No active speech provider configured')
return
return null
}
if (!activeSpeechVoice.value) {
console.warn('No active speech voice configured')
return
return null
}
const provider = await providersStore.getProviderInstance(activeSpeechProvider.value) as SpeechProviderWithExtraOptions<string, UnElevenLabsOptions>
if (!provider) {
console.error('Failed to initialize speech provider')
return
return null
}
// console.debug("ctx.data.chunk is empty? ", ctx.data.chunk === "")
// console.debug("ctx.data.special: ", ctx.data.special)
if (ctx.data.chunk === '' && !ctx.data.special)
return
// If special token only and chunk = ""
if (ctx.data.chunk === '' && ctx.data.special) {
playSpecialToken(ctx.data.special)
return
}
if (!request.text && !request.special)
return null
const providerConfig = providersStore.getProviderConfig(activeSpeechProvider.value)
const input = ssmlEnabled.value
? speechStore.generateSSML(ctx.data.chunk, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: ctx.data.chunk
? speechStore.generateSSML(request.text, activeSpeechVoice.value, { ...providerConfig, pitch: pitch.value })
: request.text
const res = await generateSpeech({
...provider.speech(activeSpeechModel.value, providerConfig),
@@ -222,22 +246,40 @@ async function handleSpeechGeneration(ctx: { data: TTSChunkItem }) {
voice: activeSpeechVoice.value.id,
})
const audioBuffer = await audioContext.decodeAudioData(res)
playbackQueue.value.enqueue({ audioBuffer, text: ctx.data.chunk, special: ctx.data.special })
}
catch (error) {
console.error('Speech generation failed:', error)
}
}
if (signal.aborted)
return null
const ttsQueue = createQueue<TTSChunkItem>({
handlers: [
handleSpeechGeneration,
],
return audioContext.decodeAudioData(res)
},
playback: playbackManager,
})
onTextSegmented((chunkItem) => {
ttsQueue.enqueue(chunkItem)
void speechRuntimeStore.registerHost(speechPipeline)
speechPipeline.on('onSpecial', (segment) => {
if (segment.special)
playSpecialToken(segment.special)
})
playbackManager.onEnd(({ item }) => {
if (item.special)
playSpecialToken(item.special)
nowSpeaking.value = false
mouthOpenSize.value = 0
})
playbackManager.onStart(({ item }) => {
nowSpeaking.value = true
// NOTICE: currently, postCaption, postPresent from useBroadcastChannel may throw error
// once we navigate away from the page that created the BroadcastChannel,
// as the channel gets closed on unmount, leading to "Failed to execute 'postMessage' on 'BroadcastChannel': The channel is closed."
// error that may block hooks or throw exceptions silently.
//
// TODO: we should consider better way to manage BroadcastChannel lifecycle to avoid such issues.
assistantCaption.value += ` ${item.text}`
postCaption({ type: 'caption-assistant', text: assistantCaption.value })
postPresent({ type: 'assistant-append', text: item.text })
})
function startLipSyncLoop() {
@@ -264,7 +306,7 @@ async function setupLipSync() {
try {
const lipSync = await createLive2DLipSync(audioContext, wlipsyncProfile as Profile, live2dLipSyncOptions)
live2dLipSync.value = lipSync
connectLipSyncNode(lipSync.node)
lipSyncNode.value = lipSync.node
await audioContext.resume()
startLipSyncLoop()
lipSyncStarted.value = true
@@ -278,18 +320,31 @@ async function setupLipSync() {
function setupAnalyser() {
if (!audioAnalyser.value) {
audioAnalyser.value = audioContext.createAnalyser()
connectAudioAnalyser(audioAnalyser.value)
}
}
let currentChatIntent: ReturnType<typeof speechRuntimeStore.openIntent> | null = null
chatHookCleanups.push(onBeforeMessageComposed(async () => {
clearAll()
playbackManager.stopAll('new-message')
setupAnalyser()
await setupLipSync()
// Reset assistant caption for a new message
assistantCaption.value = ''
postCaption({ type: 'caption-assistant', text: '' })
postPresent({ type: 'assistant-reset' })
if (currentChatIntent) {
currentChatIntent.cancel('new-message')
currentChatIntent = null
}
currentChatIntent = speechRuntimeStore.openIntent({
ownerId: activeCardId.value,
priority: 'normal',
behavior: 'queue',
})
}))
chatHookCleanups.push(onBeforeSend(async () => {
@@ -297,22 +352,21 @@ chatHookCleanups.push(onBeforeSend(async () => {
}))
chatHookCleanups.push(onTokenLiteral(async (literal) => {
// Only push to segmentation; visual presentation happens on playback start
textSegmentationQueue.value.enqueue({ type: 'literal', value: literal } as TextSegmentationItem)
currentChatIntent?.writeLiteral(literal)
}))
chatHookCleanups.push(onTokenSpecial(async (special) => {
// delaysQueue.enqueue(special)
// emotionMessageContentQueue.enqueue(special)
// Also push special token to the queue for emotion animation/delay and TTS playback synchronisation
textSegmentationQueue.value.enqueue({ type: 'special', value: special } as TextSegmentationItem)
currentChatIntent?.writeSpecial(special)
}))
chatHookCleanups.push(onStreamEnd(async () => {
delaysQueue.enqueue(llmInferenceEndToken)
currentChatIntent?.writeFlush()
}))
chatHookCleanups.push(onAssistantResponseEnd(async (_message) => {
currentChatIntent?.end()
currentChatIntent = null
// const res = await embed({
// ...transformersProvider.embed('Xenova/nomic-embed-text-v1'),
// input: message,
@@ -359,24 +413,6 @@ defineExpose({
canvasElement,
readRenderTargetRegionAtClientPoint,
})
onPlaybackFinished(() => {
nowSpeaking.value = false
mouthOpenSize.value = 0
})
onPlaybackStarted(({ text }) => {
nowSpeaking.value = true
// NOTICE: currently, postCaption, postPresent from useBroadcastChannel may throw error
// once we navigate away from the page that created the BroadcastChannel,
// as the channel gets closed on unmount, leading to "Failed to execute 'postMessage' on 'BroadcastChannel': The channel is closed."
// error that may block hooks or throw exceptions silently.
//
// TODO: we should consider better way to manage BroadcastChannel lifecycle to avoid such issues.
assistantCaption.value += ` ${text}`
postCaption({ type: 'caption-assistant', text: assistantCaption.value })
postPresent({ type: 'assistant-append', text })
})
</script>
<template>
@@ -2,7 +2,6 @@ export * from './audio'
export * from './canvas-alpha'
export * from './llm-marker-parser'
export * from './markdown'
export * from './micvad'
export * from './queues'
export * from './use-analytics'
export * from './use-build-info'
@@ -1,9 +1,157 @@
const TAG_OPEN = '<|'
const TAG_CLOSE = '|>'
interface MarkerToken {
type: 'literal' | 'special'
value: string
}
interface MarkerParserOptions {
minLiteralEmitLength?: number
}
interface StreamController<T> {
stream: ReadableStream<T>
write: (value: T) => void
close: () => void
error: (err: unknown) => void
}
function createPushStream<T>(): StreamController<T> {
let closed = false
let controller: ReadableStreamDefaultController<T> | null = null
const stream = new ReadableStream<T>({
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)
},
}
}
async function readStream<T>(stream: ReadableStream<T>, handler: (value: T) => Promise<void> | 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()
}
}
function createLlmMarkerParser(options?: MarkerParserOptions) {
const minLiteralEmitLength = Math.max(1, options?.minLiteralEmitLength ?? 1)
let buffer = ''
let inTag = false
return {
async consume(textPart: string, onLiteral: (value: string) => Promise<void> | void, onSpecial: (value: string) => Promise<void> | void) {
buffer += textPart
while (buffer.length > 0) {
if (!inTag) {
const openTagIndex = buffer.indexOf(TAG_OPEN)
if (openTagIndex < 0) {
if (buffer.length - 1 >= minLiteralEmitLength) {
const emit = buffer.slice(0, -1)
buffer = buffer[buffer.length - 1]
await onLiteral(emit)
}
break
}
if (openTagIndex > 0) {
const emit = buffer.slice(0, openTagIndex)
buffer = buffer.slice(openTagIndex)
await onLiteral(emit)
}
inTag = true
}
else {
const closeTagIndex = buffer.indexOf(TAG_CLOSE)
if (closeTagIndex < 0)
break
const emit = buffer.slice(0, closeTagIndex + TAG_CLOSE.length)
buffer = buffer.slice(closeTagIndex + TAG_CLOSE.length)
await onSpecial(emit)
inTag = false
}
}
},
async end(onLiteral: (value: string) => Promise<void> | void) {
if (!inTag && buffer.length > 0) {
await onLiteral(buffer)
buffer = ''
}
},
}
}
function createLlmMarkerStream(input: ReadableStream<string>, options?: MarkerParserOptions) {
const { stream, write, close, error } = createPushStream<MarkerToken>()
const parser = createLlmMarkerParser(options)
void readStream(input, async (chunk) => {
await parser.consume(
chunk,
async (literal) => {
if (!literal)
return
write({ type: 'literal', value: literal })
},
async (special) => {
write({ type: 'special', value: special })
},
)
})
.then(async () => {
await parser.end(async (literal) => {
if (!literal)
return
write({ type: 'literal', value: literal })
})
close()
})
.catch((err) => {
error(err)
})
return stream
}
/**
* A streaming parser for LLM responses that contain special markers (e.g., for tool calls).
* This composable is designed to be efficient and robust, using a regular expression
* This composable is designed to be efficient and robust, using a stream-based parser
* to handle special tags enclosed in `<|...|>`.
*
* @example
@@ -29,53 +177,26 @@ export function useLlmmarkerParser(options: {
*/
minLiteralEmitLength?: number
}) {
const minLiteralEmitLength = Math.max(1, options.minLiteralEmitLength ?? 1)
let buffer = ''
let inTag = false
let fullText = ''
const { stream, write, close } = createPushStream<string>()
const markerStream = createLlmMarkerStream(stream, { minLiteralEmitLength: options.minLiteralEmitLength })
const processing = readStream(markerStream, async (token) => {
if (token.type === 'literal')
await options.onLiteral?.(token.value)
if (token.type === 'special')
await options.onSpecial?.(token.value)
})
return {
/**
* Consumes a chunk of text from the stream.
* It processes the internal buffer to find and emit complete literal and special parts.
* Incomplete parts are kept in the buffer to be processed with the next chunk.
* @param textPart The chunk of text to consume.
*/
async consume(textPart: string) {
fullText += textPart
buffer += textPart
while (buffer.length > 0) {
if (!inTag) {
const openTagIndex = buffer.indexOf(TAG_OPEN)
if (openTagIndex < 0) {
if (buffer.length - 1 >= minLiteralEmitLength) {
const emit = buffer.slice(0, -1)
buffer = buffer[buffer.length - 1]
await options.onLiteral?.(emit)
}
break
}
if (openTagIndex > 0) {
const emit = buffer.slice(0, openTagIndex)
buffer = buffer.slice(openTagIndex)
await options.onLiteral?.(emit)
}
inTag = true
}
else {
const closeTagIndex = buffer.indexOf(TAG_CLOSE)
if (closeTagIndex < 0) {
break
}
const emit = buffer.slice(0, closeTagIndex + TAG_CLOSE.length)
buffer = buffer.slice(closeTagIndex + TAG_CLOSE.length)
await options.onSpecial?.(emit)
inTag = false
}
}
write(textPart)
},
/**
@@ -84,11 +205,8 @@ export function useLlmmarkerParser(options: {
* This should be called after the stream has ended.
*/
async end() {
// Incomplete tag should not be emitted as literals.
if (!inTag && buffer.length > 0) {
await options.onLiteral?.(buffer)
buffer = ''
}
close()
await processing
await options.onEnd?.(fullText)
},
}
@@ -1,74 +0,0 @@
import type { RealTimeVADOptions } from '@ricky0123/vad-web'
import type { MaybeRef } from 'vue'
import { merge } from '@moeru/std'
import { getDefaultRealTimeVADOptions, MicVAD } from '@ricky0123/vad-web'
import { usePermission } from '@vueuse/core'
import { tryOnMounted } from '@vueuse/shared'
import { onUnmounted, ref, toRef, unref, watch } from 'vue'
export function useMicVAD(deviceId: MaybeRef<ConstrainDOMString | undefined>, options: Partial<RealTimeVADOptions> & { auto?: boolean } = {}) {
const opts = merge<Omit<RealTimeVADOptions, 'stream'> & { auto?: boolean }, Partial<RealTimeVADOptions> & { auto?: boolean }>({
...getDefaultRealTimeVADOptions('v5'),
preSpeechPadMs: 30,
positiveSpeechThreshold: 0.5, // default is 0.5
negativeSpeechThreshold: 0.5 - 0.15, // default is 0.5 - 0.15
minSpeechMs: 30, // default is 9
auto: true,
}, options)
const micVad = ref<MicVAD>()
const microphoneAccess = usePermission('microphone')
async function update() {
if (micVad.value) {
micVad.value.destroy()
micVad.value = undefined
console.warn('existing MicVAD destroyed')
}
if (!microphoneAccess.value)
return
const id = unref(deviceId)
if (!id)
return
const media = await navigator.mediaDevices.getUserMedia({ audio: { deviceId: id } })
// Use of MicVAD is inspired by Open-LLM-VTuber
// Source code reference: https://github.com/t41372/Open-LLM-VTuber/blob/92cbf4349b84a68b0035bc825bc3d1d61fd0f063/static/index.html#L119
micVad.value = await MicVAD.new({
...opts,
getStream: async () => {
return media
},
})
if (opts.auto)
micVad.value.start()
}
watch(microphoneAccess, update, { immediate: true })
watch(toRef(deviceId), update, { immediate: true })
tryOnMounted(update)
onUnmounted(() => {
if (micVad.value) {
micVad.value.destroy()
micVad.value = undefined
}
})
return {
destroy: () => {
if (micVad.value) {
micVad.value.destroy()
micVad.value = undefined
}
},
start: () => {
if (micVad.value) {
micVad.value.start()
}
},
}
}
+3 -212
View File
@@ -1,26 +1,15 @@
import type { UseQueueReturn } from '@proj-airi/stream-kit'
import type { Emotion } from '../constants/emotions'
import type { UseQueueReturn } from '../utils/queue'
import type { TTSChunkItem } from '../utils/tts'
import { sleep } from '@moeru/std'
import { invoke } from '@vueuse/core'
import { defineStore } from 'pinia'
import { ref, shallowRef } from 'vue'
import { createQueue } from '@proj-airi/stream-kit'
import { EMOTION_VALUES } from '../constants/emotions'
import { createQueue } from '../utils/queue'
import { createControllableStream } from '../utils/stream'
import { chunkEmitter, TTS_SPECIAL_TOKEN } from '../utils/tts'
export interface TextSegmentationItem {
type: 'literal' | 'special'
value: string
}
export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>) {
function splitEmotion(content: string) {
for (const emotion of EMOTION_VALUES) {
// doesn't include the emotion, continue
if (!content.includes(emotion))
continue
@@ -39,16 +28,13 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
return createQueue<string>({
handlers: [
async (ctx) => {
// if the message is an emotion, push the last content to the message queue
if (EMOTION_VALUES.includes(ctx.data as Emotion)) {
ctx.emit('emotion', ctx.data as Emotion)
emotionsQueue.enqueue(ctx.data as Emotion)
return
}
// otherwise we should process the message to find the emotions
{
// iterate through the message to find the emotions
const { ok, emotion } = splitEmotion(ctx.data)
if (ok) {
ctx.emit('emotion', emotion)
@@ -62,7 +48,6 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
export function useDelayMessageQueue() {
function splitDelays(content: string) {
// doesn't include the delay, continue
if (!(/<\|DELAY:\d+\|>/i.test(content))) {
return {
ok: false,
@@ -98,7 +83,6 @@ export function useDelayMessageQueue() {
return createQueue<string>({
handlers: [
async (ctx) => {
// iterate through the message to find the emotions
const { ok, delay } = splitDelays(ctx.data)
if (ok) {
ctx.emit('delay', delay)
@@ -108,196 +92,3 @@ export function useDelayMessageQueue() {
],
})
}
export const usePipelineCharacterSpeechPlaybackQueueStore = defineStore('pipelines:character:speech', () => {
// Hooks
const onPlaybackStartedHooks = ref<Array<(payload: { text: string }) => Promise<void> | void>>([])
const onPlaybackFinishedHooks = ref<Array<(payload: { special: string }) => Promise<void> | void>>([])
// Hooks registers
function onPlaybackStarted(hook: (payload: { text: string }) => Promise<void> | void) {
onPlaybackStartedHooks.value.push(hook)
}
function onPlaybackFinished(hook: (payload: { special: string }) => Promise<void> | void) {
onPlaybackFinishedHooks.value.push(hook)
}
const currentAudioSource = shallowRef<AudioBufferSourceNode>()
const audioContext = shallowRef<AudioContext>()
const audioAnalyser = shallowRef<AnalyserNode>()
const lipSyncNode = shallowRef<AudioNode>()
function connectAudioContext(context: AudioContext) {
audioContext.value = context
}
function connectAudioAnalyser(analyser: AnalyserNode) {
audioAnalyser.value = analyser
}
function connectLipSyncNode(node: AudioNode) {
lipSyncNode.value = node
}
function clearPlaying() {
if (currentAudioSource) {
try {
currentAudioSource.value?.stop()
currentAudioSource.value?.disconnect()
}
catch {}
currentAudioSource.value = undefined
}
}
const playbackQueue = ref(invoke(() => {
return createQueue<{ audioBuffer: AudioBuffer, text: string, special: string | null }>({
handlers: [
(ctx) => {
return new Promise((resolve) => {
// NOTICE: here clearPlaying is called because that createQueue guarantees that only one handler is running at a time,
// so we can safely stop any currently playing audio before starting a new one. If multiple audios were to play
// simultaneously, this would lead to overlapping sounds.
//
// TODO: when migrating to better solution for audio playback management, be careful with this part.
// as without proper singleton gated, this may lead to audio cutoffs.
clearPlaying()
if (!audioContext.value) {
resolve()
return
}
// Create an AudioBufferSourceNode
const source = audioContext.value.createBufferSource()
source.buffer = ctx.data.audioBuffer
// Connect the source to the AudioContext's destination (the speakers)
source.connect(audioContext.value.destination)
// Connect the source to the analyzer
source.connect(audioAnalyser.value!)
// Connect to lip sync tap if provided
if (lipSyncNode.value)
source.connect(lipSyncNode.value)
// Start playing the audio
for (const hook of onPlaybackStartedHooks.value) {
try {
hook({ text: ctx.data.text })
}
catch (err) {
// NOTICE: onPlaybackStarted hook errors should not block audio playback.
// in currently use case of Stage.vue, BroadcastChannel is involved,
// navigating from pages may cause unexpected onUnmounted calls to close the channel,
// which throws error when posting message to closed channel.
//
// TODO: we should consider better way to manage BroadcastChannel lifecycle to avoid such issues.
console.error('Error in onPlaybackStarted hook:', err)
}
}
currentAudioSource.value = source
source.start(0)
source.onended = () => {
// Notify hooks regardless; consumers can decide how to use the special token (if any).
for (const hook of onPlaybackFinishedHooks.value) {
try {
hook({ special: ctx.data.special ?? '' })
}
catch (err) {
console.error('Error in onPlaybackFinished hook:', err)
}
}
if (currentAudioSource.value === source) {
currentAudioSource.value = undefined
}
resolve()
}
})
},
],
})
}))
function clearQueue() {
playbackQueue.value.clear()
}
function clearAll() {
clearPlaying()
clearQueue()
}
return {
onPlaybackStarted,
onPlaybackFinished,
connectAudioContext,
connectAudioAnalyser,
connectLipSyncNode,
clearPlaying,
clearQueue,
clearAll,
currentAudioSource,
playbackQueue,
}
})
export const usePipelineWorkflowTextSegmentationStore = defineStore('pipelines:workflows:text-segmentation', () => {
// Hooks
const onTextSegmentedHooks = ref<Array<(segment: TTSChunkItem) => Promise<void> | void>>([])
// Hooks registers
function onTextSegmented(hook: (segment: TTSChunkItem) => Promise<void> | void) {
onTextSegmentedHooks.value.push(hook)
}
function clearHooks() {
onTextSegmentedHooks.value = []
}
const textSegmentationQueue = ref(invoke(() => {
const textSegmentationStream = ref()
const textSegmentationStreamController = ref<ReadableStreamDefaultController<Uint8Array>>()
const encoder = new TextEncoder()
const { stream, controller } = createControllableStream<Uint8Array>()
textSegmentationStream.value = stream
textSegmentationStreamController.value = controller
// This is the queue for pending special tokens
const pendingSpecials: string[] = []
chunkEmitter(stream.getReader(), pendingSpecials, async (chunk) => {
for (const hook of onTextSegmentedHooks.value) {
await hook(chunk)
}
})
return createQueue<TextSegmentationItem>({
handlers: [
async (ctx) => {
if (ctx.data.type === 'literal') {
controller.enqueue(encoder.encode(ctx.data.value))
}
else {
// Special literal, need to be flushed in tts rechunking
// console.debug("TextSegmentationQueue: Special enqueue", encoder.encode(TTS_SPECIAL_TOKEN))
pendingSpecials.push(ctx.data.value)
controller.enqueue(encoder.encode(TTS_SPECIAL_TOKEN))
}
},
],
})
}))
return {
onTextSegmented,
clearHooks,
textSegmentationQueue,
}
})
@@ -144,13 +144,15 @@ describe('createStreamingCategorizer', () => {
const result = categorizer.end()
// Log what was recognized
console.info('📋 Test: should handle <think> tag with special tokens')
console.info(' Input text:', text)
console.info(' Segments found:', result.segments.length)
console.info(' Tag name:', result.segments[0]?.tagName)
console.info(' Segment content:', result.segments[0]?.content)
console.info(' Reasoning:', result.reasoning)
console.info(' Speech:', result.speech)
// eslint-disable-next-line no-console
console.log({
input: text,
segmentsFound: result.segments.length,
tagName: result.segments[0]?.tagName,
segmentContent: result.segments[0]?.content,
reasoning: result.reasoning,
speech: result.speech,
})
// Verify tag is recognized
expect(result.segments).toHaveLength(1)
@@ -180,15 +182,15 @@ describe('createStreamingCategorizer', () => {
const result = categorizer.end()
// Log what was recognized
console.info('📋 Test: should handle <think> with special tokens like <|EMOTE_HAPPY|>')
console.info(' Input text:', text)
console.info(' Segments found:', result.segments.length)
console.info(' Tag name:', result.segments[0]?.tagName)
console.info(' Segment content:', result.segments[0]?.content)
console.info(' Has <|EMOTE_HAPPY|>:', result.segments[0]?.content.includes('<|EMOTE_HAPPY|>'))
console.info(' Has <|DELAY:1|>:', result.segments[0]?.content.includes('<|DELAY:1|>'))
console.info(' Reasoning:', result.reasoning)
console.info(' Speech:', result.speech)
// eslint-disable-next-line no-console
console.log({
input: text,
segmentsFound: result.segments.length,
tagName: result.segments[0]?.tagName,
segmentContent: result.segments[0]?.content,
reasoning: result.reasoning,
speech: result.speech,
})
// Verify tag is recognized
expect(result.segments).toHaveLength(1)
@@ -0,0 +1,56 @@
import { defineEventa } from '@moeru/eventa'
import { createContext as createBroadcastChannelContext } from '@moeru/eventa/adapters/broadcast-channel'
export interface SpeechIntentStartPayload {
originId: string
intentId: string
streamId: string
ownerId?: string
priority?: number
behavior?: 'queue' | 'interrupt' | 'replace'
}
export interface SpeechIntentTokenPayload {
originId: string
intentId: string
streamId: string
sequence: number
value?: string
}
export interface SpeechIntentEndPayload {
originId: string
intentId: string
streamId: string
}
export interface SpeechIntentCancelPayload {
originId: string
intentId: string
streamId: string
reason?: string
}
export const speechIntentStartEvent = defineEventa<SpeechIntentStartPayload>('eventa:audio:speech:intent:start')
export const speechIntentLiteralEvent = defineEventa<SpeechIntentTokenPayload>('eventa:audio:speech:intent:literal')
export const speechIntentSpecialEvent = defineEventa<SpeechIntentTokenPayload>('eventa:audio:speech:intent:special')
export const speechIntentFlushEvent = defineEventa<SpeechIntentTokenPayload>('eventa:audio:speech:intent:flush')
export const speechIntentEndEvent = defineEventa<SpeechIntentEndPayload>('eventa:audio:speech:intent:end')
export const speechIntentCancelEvent = defineEventa<SpeechIntentCancelPayload>('eventa:audio:speech:intent:cancel')
const BUS_CHANNEL_NAME = 'proj-airi:pipelines:outputs:speech'
let context: ReturnType<typeof createBroadcastChannelContext>['context'] | undefined
let channel: BroadcastChannel | undefined
function getChannel() {
if (!channel)
channel = new BroadcastChannel(BUS_CHANNEL_NAME)
return channel
}
export function getSpeechBusContext() {
if (!context)
context = createBroadcastChannelContext(getChannel()).context
return context
}
@@ -0,0 +1,270 @@
import type { createSpeechPipeline, IntentHandle, IntentOptions, TextToken } from '@proj-airi/pipelines-audio'
import type { SpeechIntentStartPayload, SpeechIntentTokenPayload } from './bus'
import { createPushStream } from '@proj-airi/pipelines-audio'
import { Mutex } from 'es-toolkit'
import { nanoid } from 'nanoid'
import {
getSpeechBusContext,
speechIntentCancelEvent,
speechIntentEndEvent,
speechIntentFlushEvent,
speechIntentLiteralEvent,
speechIntentSpecialEvent,
speechIntentStartEvent,
} from './bus'
function createId(prefix: string) {
return `${prefix}-${nanoid()}`
}
export interface SpeechPipelineRuntime {
openIntent: (options?: IntentOptions) => IntentHandle
registerHost: (pipeline: ReturnType<typeof createSpeechPipeline<AudioBuffer>>) => Promise<void>
isHost: () => boolean
dispose: () => Promise<void>
}
export function createSpeechPipelineRuntime(): SpeechPipelineRuntime {
const mutex = new Mutex()
const originId = `speech-${nanoid()}`
let hostPipeline: ReturnType<typeof createSpeechPipeline<AudioBuffer>> | null = null
let hostReady = false
let bound = false
const remoteIntentMap = new Map<string, IntentHandle>()
const context = getSpeechBusContext()
function bindSpeechBusToHost() {
if (bound)
return
bound = true
context.on(speechIntentStartEvent, (evt) => {
const payload = (evt as { body?: SpeechIntentStartPayload })?.body
if (!payload || payload.originId === originId)
return
if (!hostPipeline)
return
if (remoteIntentMap.has(payload.intentId))
return
const intent = hostPipeline.openIntent({
intentId: payload.intentId,
streamId: payload.streamId,
ownerId: payload.ownerId,
priority: payload.priority,
behavior: payload.behavior,
})
remoteIntentMap.set(payload.intentId, intent)
})
const applyToken = (payload: SpeechIntentTokenPayload, writer: (intent: IntentHandle, value?: string) => void) => {
if (!payload || payload.originId === originId)
return
const intent = remoteIntentMap.get(payload.intentId)
if (!intent) {
if (!hostPipeline)
return
const fallback = hostPipeline.openIntent({ intentId: payload.intentId, streamId: payload.streamId })
remoteIntentMap.set(payload.intentId, fallback)
writer(fallback, payload.value)
return
}
writer(intent, payload.value)
}
context.on(speechIntentLiteralEvent, (evt) => {
const payload = evt?.body
if (!payload)
return
applyToken(payload, (intent, value) => {
if (value)
intent.writeLiteral(value)
})
})
context.on(speechIntentSpecialEvent, (evt) => {
const payload = evt?.body
if (!payload)
return
applyToken(payload, (intent, value) => {
if (value)
intent.writeSpecial(value)
})
})
context.on(speechIntentFlushEvent, (evt) => {
const payload = evt?.body
if (!payload)
return
applyToken(payload, (intent) => {
intent.writeFlush()
})
})
context.on(speechIntentEndEvent, (evt) => {
const payload = evt?.body
if (!payload || payload.originId === originId)
return
const intent = remoteIntentMap.get(payload.intentId)
if (!intent)
return
intent.end()
remoteIntentMap.delete(payload.intentId)
})
context.on(speechIntentCancelEvent, (evt) => {
const payload = evt?.body
if (!payload || payload.originId === originId)
return
const intent = remoteIntentMap.get(payload.intentId)
if (!intent)
return
intent.cancel(payload.reason)
remoteIntentMap.delete(payload.intentId)
})
}
function createRemoteIntent(options?: IntentOptions): IntentHandle {
const intentId = options?.intentId ?? createId('intent')
const streamId = options?.streamId ?? createId('stream')
const priority = typeof options?.priority === 'number' ? options?.priority : undefined
const behavior = options?.behavior
const ownerId = options?.ownerId
const { stream, write, close } = createPushStream<TextToken>()
let sequence = 0
let closed = false
context.emit(speechIntentStartEvent, {
originId,
intentId,
streamId,
ownerId,
priority,
behavior,
})
const handle: IntentHandle = {
intentId,
streamId,
ownerId,
priority: priority ?? 0,
stream,
writeLiteral(value: string) {
if (closed)
return
write({ type: 'literal', value, streamId, intentId, sequence, createdAt: Date.now() })
context.emit(speechIntentLiteralEvent, {
originId,
intentId,
streamId,
sequence: sequence++,
value,
})
},
writeSpecial(value: string) {
if (closed)
return
write({ type: 'special', value, streamId, intentId, sequence, createdAt: Date.now() })
context.emit(speechIntentSpecialEvent, {
originId,
intentId,
streamId,
sequence: sequence++,
value,
})
},
writeFlush() {
if (closed)
return
write({ type: 'flush', streamId, intentId, sequence, createdAt: Date.now() })
context.emit(speechIntentFlushEvent, {
originId,
intentId,
streamId,
sequence: sequence++,
})
},
end() {
if (closed)
return
closed = true
close()
context.emit(speechIntentEndEvent, {
originId,
intentId,
streamId,
})
},
cancel(reason?: string) {
if (closed)
return
closed = true
close()
context.emit(speechIntentCancelEvent, {
originId,
intentId,
streamId,
reason,
})
},
}
return handle
}
async function registerHost(pipeline: ReturnType<typeof createSpeechPipeline<AudioBuffer>>) {
await mutex.acquire()
try {
if (hostPipeline)
return
hostPipeline = pipeline
hostReady = true
bindSpeechBusToHost()
}
finally {
mutex.release()
}
}
function openIntent(options?: IntentOptions) {
if (hostPipeline)
return hostPipeline.openIntent(options)
return createRemoteIntent(options)
}
function isHost() {
return hostReady && !!hostPipeline
}
async function dispose() {
await mutex.acquire()
try {
hostPipeline = null
hostReady = false
remoteIntentMap.clear()
}
finally {
mutex.release()
}
}
return {
openIntent,
registerHost,
isHost,
dispose,
}
}
+30 -15
View File
@@ -4,7 +4,6 @@ import { createTestingPinia } from '@pinia/testing'
import { setActivePinia } from 'pinia'
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { TTS_FLUSH_INSTRUCTION } from '../utils/tts'
import { useCharacterStore } from './character'
import { useAiriCardStore } from './modules'
@@ -14,24 +13,39 @@ vi.mock('vue-i18n', () => ({
}),
}))
const enqueueSpy = vi.fn()
const writeLiteralSpy = vi.fn()
const writeFlushSpy = vi.fn()
const endSpy = vi.fn()
const cancelSpy = vi.fn()
vi.mock('../composables/queues', async () => {
const { defineStore } = await import('pinia')
const { ref } = await import('vue')
const openSpeechIntentSpy = vi.fn(() => ({
intentId: 'intent-test',
streamId: 'stream-test',
priority: 100,
stream: new ReadableStream(),
writeLiteral: writeLiteralSpy,
writeSpecial: vi.fn(),
writeFlush: writeFlushSpy,
end: endSpy,
cancel: cancelSpy,
}))
return {
usePipelineWorkflowTextSegmentationStore: defineStore('pipelines:workflows:text-segmentation', () => ({
textSegmentationQueue: ref({ enqueue: enqueueSpy }),
})),
}
})
vi.mock('../speech-runtime', () => ({
useSpeechRuntimeStore: () => ({
openIntent: openSpeechIntentSpy,
}),
}))
describe('store character', () => {
beforeEach(() => {
const pinia = createTestingPinia({ createSpy: vi.fn, stubActions: false })
setActivePinia(pinia)
enqueueSpy.mockClear()
writeLiteralSpy.mockClear()
writeFlushSpy.mockClear()
endSpy.mockClear()
cancelSpy.mockClear()
openSpeechIntentSpy.mockClear()
const airiCardStore = useAiriCardStore(pinia)
// @ts-expect-error - testing purpose
@@ -89,9 +103,10 @@ describe('store character', () => {
expect(store.reactions[0]?.sourceEventId).toBe('spark-1')
expect(store.reactions[0]?.createdAt).toBe(123456)
expect(enqueueSpy).toHaveBeenCalledWith({ type: 'literal', value: 'Hello' })
expect(enqueueSpy).toHaveBeenCalledWith({ type: 'literal', value: ' world' })
expect(enqueueSpy).toHaveBeenCalledWith({ type: 'literal', value: `${TTS_FLUSH_INSTRUCTION}${TTS_FLUSH_INSTRUCTION}` })
expect(writeLiteralSpy).toHaveBeenCalledWith('Hello')
expect(writeLiteralSpy).toHaveBeenCalledWith(' world')
expect(writeFlushSpy).toHaveBeenCalled()
expect(endSpy).toHaveBeenCalled()
nowSpy.mockRestore()
})
+67 -22
View File
@@ -1,12 +1,12 @@
import type { TextSegmentationItem } from '../../composables/queues'
import type { IntentHandle } from '@proj-airi/pipelines-audio'
import { nanoid } from 'nanoid'
import { defineStore, storeToRefs } from 'pinia'
import { computed, reactive, ref } from 'vue'
import { usePipelineWorkflowTextSegmentationStore } from '../../composables/queues'
import { TTS_FLUSH_INSTRUCTION } from '../../utils/tts'
import { useLlmmarkerParser } from '../../composables/llm-marker-parser'
import { useAiriCardStore } from '../modules'
import { useSpeechRuntimeStore } from '../speech-runtime'
export * from './orchestrator'
@@ -18,19 +18,47 @@ export interface CharacterSparkNotifyReaction {
metadata?: Record<string, unknown>
}
interface StreamingReactionState {
reaction: CharacterSparkNotifyReaction
intent: IntentHandle
parser: ReturnType<typeof useLlmmarkerParser>
}
const MAX_REACTIONS = 200
export const useCharacterStore = defineStore('character', () => {
const { activeCard, systemPrompt } = storeToRefs(useAiriCardStore())
const textSegmentationStore = usePipelineWorkflowTextSegmentationStore()
const { textSegmentationQueue } = storeToRefs(textSegmentationStore)
const name = computed(() => activeCard.value?.name ?? '')
const reactions = ref<CharacterSparkNotifyReaction[]>([])
const streamingReactions = ref<Map<string, CharacterSparkNotifyReaction>>(new Map())
const ownerId = computed(() => activeCard.value?.name ?? 'default')
function emitTextOutput(text: string) {
textSegmentationQueue.value.enqueue({ type: 'literal', value: text } as TextSegmentationItem)
const reactions = ref<CharacterSparkNotifyReaction[]>([])
const streamingReactions = ref<Map<string, StreamingReactionState>>(new Map())
const speechRuntimeStore = useSpeechRuntimeStore()
async function emitTextOutput(text: string) {
const intent = speechRuntimeStore.openIntent({
ownerId: ownerId.value,
priority: 'normal',
behavior: 'queue',
})
const parser = useLlmmarkerParser({
onLiteral: async (literal) => {
if (literal)
intent.writeLiteral(literal)
},
onSpecial: async (special) => {
if (special)
intent.writeSpecial(special)
},
})
await parser.consume(text)
await parser.end()
intent.writeFlush()
intent.end()
}
function onSparkNotifyReactionStreamEvent(sparkEventId: string, chunk: string, options?: { metadata?: Record<string, unknown> }) {
@@ -43,27 +71,45 @@ export const useCharacterStore = defineStore('character', () => {
metadata: options?.metadata,
}) satisfies CharacterSparkNotifyReaction
streamingReactions.value.set(sparkEventId, newReaction)
const intent = speechRuntimeStore.openIntent({
intentId: `spark:${sparkEventId}`,
ownerId: ownerId.value,
priority: 'high',
behavior: 'interrupt',
})
const parser = useLlmmarkerParser({
onLiteral: async (literal) => {
if (literal)
intent.writeLiteral(literal)
},
onSpecial: async (special) => {
if (special)
intent.writeSpecial(special)
},
})
streamingReactions.value.set(sparkEventId, { reaction: newReaction, intent, parser })
}
const reaction = streamingReactions.value.get(sparkEventId)!
reaction.message += chunk
emitTextOutput(chunk)
const state = streamingReactions.value.get(sparkEventId)!
state.reaction.message += chunk
void state.parser.consume(chunk)
}
function onSparkNotifyReactionStreamEnd(sparkEventId: string, fullText: string, options?: { metadata?: Record<string, unknown> }) {
if (!streamingReactions.value.has(sparkEventId)) {
const state = streamingReactions.value.get(sparkEventId)
if (!state)
return
}
const reaction = streamingReactions.value.get(sparkEventId)!
reaction.message = fullText
state.reaction.message = fullText
recordSparkNotifyReaction(sparkEventId, fullText, { metadata: options?.metadata })
streamingReactions.value.delete(sparkEventId)
emitTextOutput(`${TTS_FLUSH_INSTRUCTION}${TTS_FLUSH_INSTRUCTION}`)
void state.parser.end().then(() => {
state.intent.writeFlush()
state.intent.end()
streamingReactions.value.delete(sparkEventId)
})
}
function recordSparkNotifyReaction(sparkEventId: string, message: string, options?: { metadata?: Record<string, unknown> }) {
@@ -77,7 +123,6 @@ export const useCharacterStore = defineStore('character', () => {
reactions.value.push(newReaction)
// Trim reactions if exceeding max limit
if (reactions.value.length > MAX_REACTIONS) {
reactions.value.splice(0, reactions.value.length - MAX_REACTIONS)
}
+3 -6
View File
@@ -5,6 +5,7 @@ import type { StreamEvent, StreamOptions } from '../stores/llm'
import type { ChatAssistantMessage, ChatHistoryItem, ChatSlices, ChatStreamEventContext, ContextMessage, StreamingAssistantMessage } from '../types/chat'
import { ContextUpdateStrategy } from '@proj-airi/server-sdk'
import { createQueue } from '@proj-airi/stream-kit'
import { useLocalStorage } from '@vueuse/core'
import { defineStore, storeToRefs } from 'pinia'
import { computed, ref, toRaw, watch } from 'vue'
@@ -13,8 +14,6 @@ import { useAnalytics } from '../composables'
import { useLlmmarkerParser } from '../composables/llm-marker-parser'
import { categorizeResponse, createStreamingCategorizer } from '../composables/response-categoriser'
import { useLLM } from '../stores/llm'
import { createQueue } from '../utils/queue'
import { TTS_FLUSH_INSTRUCTION } from '../utils/tts'
import { useCharacterStore } from './character'
import { useConsciousnessStore } from './modules/consciousness'
@@ -409,6 +408,7 @@ export const useChatStore = defineStore('chat', () => {
if (shouldAbort())
return
console.log('literal', literal)
// Feed to categorizer first
categorizer.consume(literal)
@@ -422,6 +422,7 @@ export const useChatStore = defineStore('chat', () => {
if (speechOnly.trim()) {
streamingMessage.value.content += speechOnly
console.log('speechOnly', speechOnly)
// Emit TTS only for speech parts, not reasoning (clean data, no empty chunks)
await emitTokenLiteralHooks(speechOnly, streamingMessageContext)
@@ -568,10 +569,6 @@ export const useChatStore = defineStore('chat', () => {
sessionMessagesForSend.push(toRaw(streamingMessage.value))
}
// Instruct the TTS pipeline to flush by calling hooks directly
const flushSignal = `${TTS_FLUSH_INSTRUCTION}${TTS_FLUSH_INSTRUCTION}`
await emitTokenLiteralHooks(flushSignal, streamingMessageContext)
// Call the end-of-stream hooks
await emitStreamEndHooks(streamingMessageContext)
@@ -0,0 +1,30 @@
import { defineStore } from 'pinia'
import { createSpeechPipelineRuntime } from '../services/speech/pipeline-runtime'
export const useSpeechRuntimeStore = defineStore('speech-runtime', () => {
const runtime = createSpeechPipelineRuntime()
function openIntent(options?: Parameters<typeof runtime.openIntent>[0]) {
return runtime.openIntent(options)
}
async function registerHost(pipeline: Parameters<typeof runtime.registerHost>[0]) {
await runtime.registerHost(pipeline)
}
function isHost() {
return runtime.isHost()
}
async function dispose() {
await runtime.dispose()
}
return {
openIntent,
registerHost,
isHost,
dispose,
}
})
+36
View File
@@ -0,0 +1,36 @@
{
"name": "@proj-airi/stream-kit",
"type": "module",
"version": "0.8.1-beta.8",
"private": true,
"description": "Stream utilities for AIRI (queues, streams)",
"author": {
"name": "Moeru AI Project AIRI Team",
"email": "airi@moeru.ai",
"url": "https://github.com/moeru-ai"
},
"license": "MIT",
"repository": {
"type": "git",
"url": "https://github.com/moeru-ai/airi.git",
"directory": "packages/stream-kit"
},
"exports": {
".": {
"types": "./dist/index.d.mts",
"default": "./dist/index.mjs"
}
},
"main": "./dist/index.mjs",
"types": "./dist/index.d.mts",
"files": [
"README.md",
"dist",
"package.json"
],
"scripts": {
"build": "tsdown",
"typecheck": "tsc --noEmit"
},
"devDependencies": {}
}
+1
View File
@@ -0,0 +1 @@
export * from './queue'
@@ -59,14 +59,8 @@ export function createQueue<T>(options: {
queue.length = 0
}
// Internal
// Drain the queue and call the handlers with each dequeued item
async function drain() {
while (queue.length > 0) {
// The async func should never yield at here, and shift() should
// always return an item from the queue.
// We cannot check for `undefined` here, because there could be
// someone using the queue to enqueue and dequeue `undefined`.
const payload = queue.shift() as T
emit('dequeue', payload, queue.length)
for (const handler of options.handlers) {
@@ -76,7 +70,6 @@ export function createQueue<T>(options: {
emit('result', payload, result, handler)
}
catch (err) {
// Keep `unknown` and let the event listener handle the error type
emit('error', payload, err, handler)
continue
}
+11
View File
@@ -0,0 +1,11 @@
{
"extends": "../../tsconfig.json",
"compilerOptions": {
"lib": ["ESNext", "DOM", "DOM.Iterable"],
"types": []
},
"include": [
"./src/**/*",
"./index.ts"
]
}
+6
View File
@@ -0,0 +1,6 @@
import { defineConfig } from 'tsdown'
export default defineConfig({
entry: ['src/index.ts'],
dts: true,
})
+23
View File
@@ -472,6 +472,9 @@ importers:
'@proj-airi/i18n':
specifier: workspace:^
version: link:../../packages/i18n
'@proj-airi/pipelines-audio':
specifier: workspace:^
version: link:../../packages/pipelines-audio
'@proj-airi/server-sdk':
specifier: workspace:^
version: link:../../packages/server-sdk
@@ -487,6 +490,9 @@ importers:
'@proj-airi/stage-ui-three-performance-runtime':
specifier: workspace:^
version: link:../../packages/stage-ui-three-performance-runtime
'@proj-airi/stream-kit':
specifier: workspace:^
version: link:../../packages/stream-kit
'@proj-airi/ui':
specifier: workspace:^
version: link:../../packages/ui
@@ -1263,6 +1269,9 @@ importers:
'@proj-airi/model-driver-mediapipe':
specifier: workspace:^
version: link:../../packages/model-driver-mediapipe
'@proj-airi/pipelines-audio':
specifier: workspace:^
version: link:../../packages/pipelines-audio
'@proj-airi/server-sdk':
specifier: workspace:^
version: link:../../packages/server-sdk
@@ -1281,6 +1290,9 @@ importers:
'@proj-airi/stage-ui-three-performance-runtime':
specifier: workspace:^
version: link:../../packages/stage-ui-three-performance-runtime
'@proj-airi/stream-kit':
specifier: workspace:^
version: link:../../packages/stream-kit
'@proj-airi/ui':
specifier: workspace:^
version: link:../../packages/ui
@@ -1854,6 +1866,9 @@ importers:
packages/pipelines-audio:
dependencies:
'@moeru/eventa':
specifier: ^1.0.0-alpha.10
version: 1.0.0-alpha.10(electron@39.2.7)(h3@2.0.1-rc.5(crossws@0.4.1(srvx@0.9.8(patch_hash=f0151386fdcbcb6f53833cf8f66926ef2eb31b71a2782d6e0960dec832ef108d))))
'@moeru/std':
specifier: 'catalog:'
version: 0.1.0-beta.14
@@ -2178,6 +2193,9 @@ importers:
'@proj-airi/model-driver-lipsync':
specifier: workspace:^
version: link:../model-driver-lipsync
'@proj-airi/pipelines-audio':
specifier: workspace:^
version: link:../pipelines-audio
'@proj-airi/server-sdk':
specifier: workspace:^
version: link:../server-sdk
@@ -2443,6 +2461,9 @@ importers:
'@proj-airi/lobe-icons':
specifier: ^1.0.18
version: 1.0.18
'@proj-airi/stream-kit':
specifier: workspace:*
version: link:../stream-kit
'@proj-airi/vite-plugin-warpdrive':
specifier: workspace:*
version: link:../vite-plugin-warpdrive
@@ -2656,6 +2677,8 @@ importers:
specifier: ^3.5.25
version: 3.5.25(typescript@5.9.3)
packages/stream-kit: {}
packages/tauri-plugin-mcp:
dependencies:
'@tauri-apps/api':