nanakura/moeka-project
1
0
Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

refactor(stage-ui): make queue framework-agnostic (#526)

Browse Source 
* refactor(stage-ui): make queue framework-agnostic

* fix(stage-ui): update usages

* fix(stage-ui): trust the shift

* fix(stage-ui): typecheck

* fix(stage-ui): keep caught error unknown
This commit is contained in:
Makito
2025-08-30 18:40:58 +09:00
committed by GitHub
parent 9e226184b3
commit 536786331b
12 changed files with 147 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/stage-web/src/pages/queue.vue
+7 -7
View File
@@ -1,11 +1,11 @@
<script setup lang="ts">
import { sleep } from '@moeru/std'
import { useQueue } from '@proj-airi/stage-ui/composables/queue'
import { createQueue } from '@proj-airi/stage-ui/utils/queue'
import { onMounted, ref } from 'vue'
const temp = ref<string>('')
const audioQueue = useQueue<string>({
const audioQueue = createQueue<string>({
handlers: [
async (text) => {
// eslint-disable-next-line no-console
@@ -13,16 +13,16 @@ const audioQueue = useQueue<string>({
},
],
})
const ttsQueue = useQueue<string>({
const ttsQueue = createQueue<string>({
handlers: [
async (ctx) => {
// eslint-disable-next-line no-console
console.log('ready to stream speech audio for', ctx)
audioQueue.add(ctx.data)
audioQueue.enqueue(ctx.data)
},
],
})
const textQueue = useQueue<string>({
const textQueue = createQueue<string>({
handlers: [
async (ctx) => {
const endMarker = ['.', '?', '!']
@@ -40,7 +40,7 @@ const textQueue = useQueue<string>({
const afterPeriod = ctx.data.slice(periodIndex + 1)
temp.value += beforePeriod
ttsQueue.add(temp.value.trim())
ttsQueue.enqueue(temp.value.trim())
temp.value = afterPeriod
newEndPartDiscovered = true
@@ -129,7 +129,7 @@ const textParts = [
async function mockTextPartsStreamHandler() {
for (const part of textParts) {
await sleep(100)
textQueue.add(part)
textQueue.enqueue(part)
}
}
apps/stage-web/src/pages/test/queues/delays.vue
+2 -2
View File
@@ -26,9 +26,9 @@ function onSendMessage() {
processing.value = true
const tokens = messageInput.value.split('')
for (const token of tokens)
delaysQueue.add(token)
delaysQueue.enqueue(token)
delaysQueue.add(llmInferenceEndToken)
delaysQueue.enqueue(llmInferenceEndToken)
messageInput.value = ''
processing.value = false
}
apps/stage-web/src/pages/test/queues/emotions.vue
+4 -4
View File
@@ -1,9 +1,9 @@
<script setup lang="ts">
import type { Emotion } from '@proj-airi/stage-ui/constants/emotions'
import { useQueue } from '@proj-airi/stage-ui/composables/queue'
import { useEmotionsMessageQueue } from '@proj-airi/stage-ui/composables/queues'
import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
import { createQueue } from '@proj-airi/stage-ui/utils/queue'
import { Textarea } from '@proj-airi/ui'
import { ref } from 'vue'
@@ -12,7 +12,7 @@ const messagesProcessed = ref<string[]>([])
const emotionsProcessed = ref<string[]>([])
const processing = ref<boolean>(false)
const emotionsQueue = useQueue<Emotion>({
const emotionsQueue = createQueue<Emotion>({
handlers: [
async (ctx) => {
emotionsProcessed.value.push(ctx.data)
@@ -26,9 +26,9 @@ function onSendMessage() {
processing.value = true
const tokens = messageInput.value.split('')
for (const token of tokens)
emotionMessageContentQueue.add(token)
emotionMessageContentQueue.enqueue(token)
emotionMessageContentQueue.add(llmInferenceEndToken)
emotionMessageContentQueue.enqueue(llmInferenceEndToken)
messageInput.value = ''
processing.value = false
}
apps/stage-web/src/pages/test/queues/messages.vue
+4 -4
View File
@@ -1,7 +1,7 @@
<script setup lang="ts">
import { useQueue } from '@proj-airi/stage-ui/composables/queue'
import { useMessageContentQueue } from '@proj-airi/stage-ui/composables/queues'
import { llmInferenceEndToken } from '@proj-airi/stage-ui/constants'
import { createQueue } from '@proj-airi/stage-ui/utils/queue'
import { Textarea } from '@proj-airi/ui'
import { ref } from 'vue'
@@ -9,7 +9,7 @@ const messageInput = ref<string>('')
const ttsProcessed = ref<string[]>([])
const processing = ref<boolean>(false)
const ttsQueue = useQueue<string>({
const ttsQueue = createQueue<string>({
handlers: [
async (ctx) => {
ttsProcessed.value.push(ctx.data)
@@ -26,9 +26,9 @@ async function onSendMessage() {
// await sleep(100)
// messageContentQueue.add(token)
// }
messageContentQueue.add(messageInput.value)
messageContentQueue.enqueue(messageInput.value)
messageContentQueue.add(llmInferenceEndToken)
messageContentQueue.enqueue(llmInferenceEndToken)
messageInput.value = ''
processing.value = false
}
packages/stage-ui/src/components/Scenarios/Providers/SpeechStreamingPlayground.vue
+5 -5
View File
@@ -4,9 +4,9 @@ import type { TTSInputChunk } from '../../../utils/tts'
import { animate } from 'animejs'
import { ref } from 'vue'
import { useQueue } from '../../../composables/queue'
import { useMessageContentQueue } from '../../../composables/queues'
import { useAudioContext } from '../../../stores/audio'
import { createQueue } from '../../../utils/queue'
import { chunkTTSInput } from '../../../utils/tts'
const props = defineProps<{
@@ -21,7 +21,7 @@ const nowSpeaking = ref(false)
const ttsInputChunks = ref<TTSInputChunk[]>([])
const speechGenerationIndex = ref(-1)
const audioQueue = useQueue<{ audioBuffer: AudioBuffer, text: string }>({
const audioQueue = createQueue<{ audioBuffer: AudioBuffer, text: string }>({
handlers: [
(ctx) => {
return new Promise((resolve) => {
@@ -54,19 +54,19 @@ async function handleSpeechGeneration(ctx: { data: string }) {
// Decode the ArrayBuffer into an AudioBuffer
const audioBuffer = await audioContext.decodeAudioData(res)
await audioQueue.add({ audioBuffer, text: ctx.data })
audioQueue.enqueue({ audioBuffer, text: ctx.data })
}
catch (error) {
console.error('Speech generation failed:', error)
}
}
const ttsQueue = useQueue<string>({ handlers: [handleSpeechGeneration] })
const ttsQueue = createQueue<string>({ handlers: [handleSpeechGeneration] })
const messageContentQueue = useMessageContentQueue(ttsQueue)
async function testStreaming() {
await messageContentQueue.add(props.text)
messageContentQueue.enqueue(props.text)
}
async function testChunking() {
packages/stage-ui/src/components/Scenes/Stage.vue
+10 -15
View File
@@ -17,7 +17,6 @@ import { onMounted, onUnmounted, ref } from 'vue'
import Live2DScene from './Live2D.vue'
import VRMScene from './VRM.vue'
import { useQueue } from '../../composables/queue'
import { useDelayMessageQueue, useEmotionsMessageQueue, useMessageContentQueue } from '../../composables/queues'
import { llmInferenceEndToken } from '../../constants'
import { EMOTION_EmotionMotionName_value, EMOTION_VRMExpressionName_value, EmotionThinkMotionName } from '../../constants/emotions'
@@ -28,6 +27,7 @@ import { useSpeechStore } from '../../stores/modules/speech'
import { useProvidersStore } from '../../stores/providers'
import { useSettings } from '../../stores/settings'
import { useVRM } from '../../stores/vrm'
import { createQueue } from '../../utils/queue'
withDefaults(defineProps<{
paused?: boolean
@@ -76,7 +76,7 @@ const nowSpeaking = ref(false)
const lipSyncStarted = ref(false)
let currentAudioSource: AudioBufferSourceNode | null = null
const audioQueue = useQueue<{ audioBuffer: AudioBuffer, text: string }>({
const audioQueue = createQueue<{ audioBuffer: AudioBuffer, text: string }>({
handlers: [
(ctx) => {
return new Promise((resolve) => {
@@ -150,29 +150,24 @@ async function handleSpeechGeneration(ctx: { data: string }) {
// Decode the ArrayBuffer into an AudioBuffer
const audioBuffer = await audioContext.decodeAudioData(res)
await audioQueue.add({ audioBuffer, text: ctx.data })
audioQueue.enqueue({ audioBuffer, text: ctx.data })
}
catch (error) {
console.error('Speech generation failed:', error)
}
}
const ttsQueue = useQueue<string>({
const ttsQueue = createQueue<string>({
handlers: [
handleSpeechGeneration,
],
})
ttsQueue.on('add', (content) => {
// eslint-disable-next-line no-console
console.debug('ttsQueue added', content)
})
const messageContentQueue = useMessageContentQueue(ttsQueue)
const { currentMotion } = storeToRefs(useLive2d())
const emotionsQueue = useQueue<Emotion>({
const emotionsQueue = createQueue<Emotion>({
handlers: [
async (ctx) => {
if (stageModelRenderer.value === 'vrm') {
@@ -232,7 +227,7 @@ onBeforeMessageComposed(async () => {
catch {}
currentAudioSource = null
}
audioQueue.queue.value = []
audioQueue.clear()
setupAnalyser()
setupLipSync()
})
@@ -242,16 +237,16 @@ onBeforeSend(async () => {
})
onTokenLiteral(async (literal) => {
await messageContentQueue.add(literal)
messageContentQueue.enqueue(literal)
})
onTokenSpecial(async (special) => {
await delaysQueue.add(special)
await emotionMessageContentQueue.add(special)
delaysQueue.enqueue(special)
emotionMessageContentQueue.enqueue(special)
})
onStreamEnd(async () => {
await delaysQueue.add(llmInferenceEndToken)
delaysQueue.enqueue(llmInferenceEndToken)
})
onAssistantResponseEnd(async (_message) => {
packages/stage-ui/src/composables/index.ts
-1
View File
@@ -2,7 +2,6 @@ export * from './audio'
export * from './llmmarkerParser'
export * from './markdown'
export * from './micvad'
export * from './queue'
export * from './queues'
export * from './vrm'
export * from './whisper'
packages/stage-ui/src/composables/queue.ts
-123
View File
@@ -1,123 +0,0 @@
import type { Ref } from 'vue'
import { ref } from 'vue'
export interface HandlerContext<T> {
data: T
itemsToBeProcessed: () => number
emit: (eventName: string, ...params: any[]) => void
}
export interface Events<T> {
add: Array<(payload: T) => void>
pick: Array<(payload: T) => void>
processing: Array<(payload: T, handler: (param: HandlerContext<T>) => Promise<any>) => void>
error: Array<(payload: T, error: Error, handler: (param: HandlerContext<T>) => Promise<any>) => void>
processed: Array<<R>(payload: T, result: R, handler: (param: HandlerContext<T>) => Promise<any>) => void>
done: Array<(payload: T) => void>
}
export function useQueue<T>(options: {
handlers: Array<(ctx: HandlerContext<T>) => Promise<void>>
}) {
const queue = ref<T[]>([]) as Ref<T[]>
const isProcessing = ref(false)
const internalEventListeners: Events<T> = {
add: [],
pick: [],
processing: [],
error: [],
processed: [],
done: [],
}
const internalHandlerEventListeners: Record<string, Array<(...params: any[]) => void>> = {}
function on<E extends keyof Events<T>>(eventName: E, Listener: Events<T>[E][number]) {
internalEventListeners[eventName].push(Listener as any)
}
function emit<E extends keyof Events<T>>(eventName: E, ...params: Parameters<Events<T>[E][number]>) {
const handlers = internalEventListeners[eventName] as Events<T>[E]
handlers.forEach((handler) => {
(handler as any)(...params)
})
}
function onHandlerEvent(eventName: string, handler: (...params: any[]) => void) {
internalHandlerEventListeners[eventName] = internalHandlerEventListeners[eventName] || []
internalHandlerEventListeners[eventName].push(handler)
}
function emitHandlerEvent(eventName: string, ...params: any[]) {
const handlers = internalHandlerEventListeners[eventName] || []
handlers.forEach((handler) => {
handler(...params)
})
}
async function add(payload: T) {
queue.value.push(payload)
emit('add', payload)
}
function pick() {
const payload = queue.value.shift()
if (!payload)
return
emit('pick', payload)
return payload
}
// Listener for item add / enqueue event
async function addItemListener() {
if (isProcessing.value)
return
const payload = pick()
if (!payload)
return
isProcessing.value = true
for (const handler of options.handlers) {
// If there is a need to register customised listener for processing event, then this line of code should be rewritten
// handlers as the input parameter is only designed for the add event
emit('processing', payload, handler)
try {
// Use handler to deal with the newly enqueued item
const result = await handler({ data: payload, itemsToBeProcessed: () => queue.value.length, emit: emitHandlerEvent })
// If there is a need to register customised listener for processing event, then this line of code should be rewritten
// handlers as the input parameter is only designed for the add event
emit('processed', payload, result, handler)
}
catch (err) {
// If there is a need to register customised listener for processing event, then this line of code should be rewritten
// handlers as the input parameter is only designed for the add event
emit('error', payload, err as Error, handler)
continue
}
}
isProcessing.value = false
emit('done', payload)
// Process next item if any
if (queue.value.length > 0)
addItemListener()
}
on('add', addItemListener)
// Lilia: I'm not sure why do we need to register handleItem to 'done', if queue is not empty, addItemListner will continue to call addItemListner. Calling this function again when 'done' event is triggered will only lead to payload = undefined (since queue is already empty) and return
// Maybe leave 'done' event to be registered other customised listeners would be more reasonable
// on('done', addItemListener)
return {
add,
on,
onHandlerEvent,
queue,
}
}
export type UseQueueReturn<T> = ReturnType<typeof useQueue<T>>
packages/stage-ui/src/composables/queues.ts
+7 -8
View File
@@ -1,12 +1,12 @@
import type { Emotion } from '../constants/emotions'
import type { UseQueueReturn } from './queue'
import type { UseQueueReturn } from '../utils/queue'
import { sleep } from '@moeru/std'
import { EMOTION_VALUES } from '../constants/emotions'
import { createQueue } from '../utils/queue'
import { createControllableStream } from '../utils/stream'
import { chunkToTTSQueue } from '../utils/tts'
import { useQueue } from './queue'
export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>) {
function splitEmotion(content: string) {
@@ -27,14 +27,13 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
}
}
return useQueue<string>({
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)
await emotionsQueue.add(ctx.data as Emotion)
emotionsQueue.enqueue(ctx.data as Emotion)
return
}
@@ -44,7 +43,7 @@ export function useEmotionsMessageQueue(emotionsQueue: UseQueueReturn<Emotion>)
const { ok, emotion } = splitEmotion(ctx.data)
if (ok) {
ctx.emit('emotion', emotion)
await emotionsQueue.add(emotion)
emotionsQueue.enqueue(emotion)
}
}
},
@@ -87,7 +86,7 @@ export function useDelayMessageQueue() {
}
}
return useQueue<string>({
return createQueue<string>({
handlers: [
async (ctx) => {
// iterate through the message to find the emotions
@@ -107,7 +106,7 @@ export function useMessageContentQueue(ttsQueue: UseQueueReturn<string>) {
chunkToTTSQueue(stream.getReader(), ttsQueue)
return useQueue<string>({
return createQueue<string>({
handlers: [
async (ctx) => {
controller.enqueue(encoder.encode(ctx.data))
packages/stage-ui/src/stores/chat.ts
+4 -4
View File
@@ -7,9 +7,9 @@ import type { ChatAssistantMessage, ChatMessage, ChatSlices } from '../types/cha
import { defineStore, storeToRefs } from 'pinia'
import { ref, toRaw } from 'vue'
import { useQueue } from '../composables'
import { useLlmmarkerParser } from '../composables/llmmarkerParser'
import { useLLM } from '../stores/llm'
import { createQueue } from '../utils/queue'
import { TTS_FLUSH_INSTRUCTION } from '../utils/tts'
import { useAiriCardStore } from './modules'
@@ -144,7 +144,7 @@ export const useChatStore = defineStore('chat', () => {
minLiteralEmitLength: 24, // Avoid emitting literals too fast. This is a magic number and can be changed later.
})
const toolCallQueue = useQueue<ChatSlices>({
const toolCallQueue = createQueue<ChatSlices>({
handlers: [
async (ctx) => {
if (ctx.data.type === 'tool-call') {
@@ -184,13 +184,13 @@ export const useChatStore = defineStore('chat', () => {
headers,
async onStreamEvent(event: StreamEvent) {
if (event.type === 'tool-call') {
toolCallQueue.add({
toolCallQueue.enqueue({
type: 'tool-call',
toolCall: event,
})
}
else if (event.type === 'tool-result') {
toolCallQueue.add({
toolCallQueue.enqueue({
type: 'tool-call-result',
id: event.toolCallId,
result: event.result,
packages/stage-ui/src/utils/queue.ts
+102
View File
@@ -0,0 +1,102 @@
export interface HandlerContext<T> {
data: T
emit: (eventName: string, ...params: any[]) => void
}
export interface Events<T> {
enqueue: Array<(payload: T, queueLength: number) => void>
dequeue: Array<(payload: T, queueLength: number) => void>
process: Array<(payload: T, handler: (param: HandlerContext<T>) => Promise<any>) => void>
error: Array<(payload: T, error: unknown, handler: (param: HandlerContext<T>) => Promise<any>) => void>
result: Array<<R>(payload: T, result: R, handler: (param: HandlerContext<T>) => Promise<any>) => void>
drain: Array<() => void>
}
export function createQueue<T>(options: {
handlers: Array<(ctx: HandlerContext<T>) => Promise<void>>
}) {
const queue: T[] = []
let drainTask: Promise<any> | undefined
const internalEventListeners: Events<T> = {
enqueue: [],
dequeue: [],
process: [],
error: [],
result: [],
drain: [],
}
const internalHandlerEventListeners: Record<string, Array<(...params: any[]) => void>> = {}
function on<E extends keyof Events<T>>(eventName: E, listener: Events<T>[E][number]) {
internalEventListeners[eventName].push(listener as any)
}
function emit<E extends keyof Events<T>>(eventName: E, ...params: Parameters<Events<T>[E][number]>) {
const listeners = internalEventListeners[eventName] as Events<T>[E]
listeners.forEach(listener => (listener as any)(...params))
}
function onHandlerEvent(eventName: string, listener: (...params: any[]) => void) {
internalHandlerEventListeners[eventName] = internalHandlerEventListeners[eventName] || []
internalHandlerEventListeners[eventName].push(listener)
}
function emitHandlerEvent(eventName: string, ...params: any[]) {
const listeners = internalHandlerEventListeners[eventName] || []
listeners.forEach(listener => listener(...params))
}
function enqueue(payload: T) {
queue.push(payload)
emit('enqueue', payload, queue.length)
if (!drainTask) {
drainTask = drain()
}
}
function clear() {
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) {
emit('process', payload, handler)
try {
const result = await handler({ data: payload, emit: emitHandlerEvent })
emit('result', payload, result, handler)
}
catch (err) {
// Keep `unknown` and let the event listener handle the error type
emit('error', payload, err, handler)
continue
}
}
}
emit('drain')
drainTask = undefined
}
function length() {
return queue.length
}
return {
enqueue,
clear,
length,
on,
onHandlerEvent,
}
}
export type UseQueueReturn<T> = ReturnType<typeof createQueue<T>>
packages/stage-ui/src/utils/tts.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import type { ReaderLike } from 'clustr'
import type { UseQueueReturn } from '../composables/queue'
import type { UseQueueReturn } from './queue'
import { readGraphemeClusters } from 'clustr'
@@ -207,7 +207,7 @@ export async function chunkToTTSQueue(reader: ReaderLike, queue: UseQueueReturn<
// TODO: remove later
// eslint-disable-next-line no-console
console.debug('chunk to be pushed: ', chunk)
await queue.add(chunk.text)
queue.enqueue(chunk.text)
}
}
catch (e) {