diff --git a/apps/stage-web/src/components/Layouts/InteractiveArea.vue b/apps/stage-web/src/components/Layouts/InteractiveArea.vue
index 7501511f9..1b04307be 100644
--- a/apps/stage-web/src/components/Layouts/InteractiveArea.vue
+++ b/apps/stage-web/src/components/Layouts/InteractiveArea.vue
@@ -1,9 +1,11 @@
@@ -29,7 +32,7 @@ const isLoading = ref(true)
+import type { ChatHistoryMessage } from '@proj-airi/stage-ui/components'
import type { ChatProvider } from '@xsai-ext/shared-providers'
import { ChatHistory, HearingConfigDialog } from '@proj-airi/stage-ui/components'
@@ -11,7 +12,7 @@ import { useSettings, useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/
import { BasicTextarea, useTheme } from '@proj-airi/ui'
import { useResizeObserver, useScreenSafeArea } from '@vueuse/core'
import { storeToRefs } from 'pinia'
-import { onMounted, onUnmounted, ref, useTemplateRef, watch } from 'vue'
+import { computed, onMounted, onUnmounted, ref, useTemplateRef, watch } from 'vue'
import { useI18n } from 'vue-i18n'
import { RouterLink } from 'vue-router'
@@ -24,7 +25,9 @@ import { BackgroundDialogPicker } from '../Backgrounds'
const { isDark, toggleDark } = useTheme()
const hearingDialogOpen = ref(false)
-const { messages, sending, streamingMessage } = storeToRefs(useChatStore())
+const chatStore = useChatStore()
+const { messages, sending, streamingMessage } = storeToRefs(chatStore)
+const historyMessages = computed(() => messages.value as unknown as ChatHistoryMessage[])
const viewControlsActiveMode = ref<'x' | 'y' | 'z' | 'scale'>('scale')
const viewControlsInputsRef = useTemplateRef>('viewControlsInputs')
@@ -41,7 +44,7 @@ useResizeObserver(document.documentElement, () => screenSafeArea.update())
const { themeColorsHueDynamic, stageViewControlsEnabled } = storeToRefs(useSettings())
const settingsAudioDevice = useSettingsAudioDevice()
const { enabled, selectedAudioInput, stream, audioInputs } = storeToRefs(settingsAudioDevice)
-const { send, onAfterMessageComposed, discoverToolsCompatibility, cleanupMessages } = useChatStore()
+const { send, onAfterMessageComposed, discoverToolsCompatibility, cleanupMessages } = chatStore
const { t } = useI18n()
const { audioContext } = useAudioContext()
const { startAnalyzer, stopAnalyzer, volumeLevel } = useAudioAnalyzer()
@@ -62,16 +65,20 @@ async function handleSend() {
return
}
+ const textToSend = messageInput.value
+ messageInput.value = ''
+
try {
const providerConfig = providersStore.getProviderConfig(activeProvider.value)
- await send(messageInput.value, {
+ await send(textToSend, {
chatProvider: await providersStore.getProviderInstance(activeProvider.value) as ChatProvider,
model: activeModel.value,
providerConfig,
})
}
catch (error) {
+ messageInput.value = textToSend
messages.value.pop()
messages.value.push({
role: 'error',
@@ -113,7 +120,6 @@ watch(hearingDialogOpen, (value) => {
})
onAfterMessageComposed(async () => {
- messageInput.value = ''
})
watch([activeProvider, activeModel], async () => {
@@ -139,7 +145,7 @@ onMounted(() => {
{
})
onAfterMessageComposed(async () => {
- messageInput.value = ''
})
const { startAnalyzer, stopAnalyzer, volumeLevel } = useAudioAnalyzer()
diff --git a/packages/stage-ui/src/components/scenarios/chat/ChatHistory.vue b/packages/stage-ui/src/components/scenarios/chat/ChatHistory.vue
index 070df7e93..8ce132274 100644
--- a/packages/stage-ui/src/components/scenarios/chat/ChatHistory.vue
+++ b/packages/stage-ui/src/components/scenarios/chat/ChatHistory.vue
@@ -11,7 +11,7 @@ import ChatUserItem from './ChatUserItem.vue'
const props = withDefaults(defineProps<{
messages: ChatHistoryMessage[]
- streamingMessage?: ChatAssistantMessage
+ streamingMessage?: ChatAssistantMessage & { context?: { ts?: number } }
sending?: boolean
assistantLabel?: string
userLabel?: string
@@ -46,18 +46,33 @@ watch([() => props.messages, () => props.streamingMessage], scrollToBottom, { de
watch(() => props.sending, scrollToBottom, { flush: 'post' })
onMounted(scrollToBottom)
-const streaming = computed(() => props.streamingMessage ?? { role: 'assistant', content: '', slices: [], tool_results: [] })
+const streaming = computed(() => props.streamingMessage ?? { role: 'assistant', content: '', slices: [], tool_results: [] })
const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length ?? 0) === 0 && !streaming.value.content)
+const streamingTs = computed(() => streaming.value.context?.ts)
+const renderMessages = computed(() => {
+ if (!props.sending)
+ return props.messages
+
+ const streamTs = streamingTs.value
+ if (!streamTs)
+ return props.messages
+
+ const hasStreamAlready = streamTs && props.messages.some(msg => msg.context?.ts === streamTs)
+ if (hasStreamAlready)
+ return props.messages
+
+ return [...props.messages, streaming.value]
+})
-
+
@@ -66,6 +81,7 @@ const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length
@@ -78,14 +94,5 @@ const showStreamingPlaceholder = computed(() => (streaming.value.slices?.length
/>
-
-
-
-
diff --git a/packages/stage-ui/src/components/scenarios/chat/types.ts b/packages/stage-ui/src/components/scenarios/chat/types.ts
index 0b9169beb..22fb23308 100644
--- a/packages/stage-ui/src/components/scenarios/chat/types.ts
+++ b/packages/stage-ui/src/components/scenarios/chat/types.ts
@@ -7,4 +7,8 @@ export interface ChatErrorMessage {
export type ChatHistoryMessage = (ChatMessage | ChatErrorMessage) & {
slices?: ChatSlices[]
+ context?: {
+ ts?: number
+ [key: string]: unknown
+ }
}
diff --git a/packages/stage-ui/src/stores/chat.ts b/packages/stage-ui/src/stores/chat.ts
index 76028d850..4f1a62468 100644
--- a/packages/stage-ui/src/stores/chat.ts
+++ b/packages/stage-ui/src/stores/chat.ts
@@ -27,8 +27,7 @@ interface MessageContext {
meta?: Record
}
-type ChatEntry = (ChatMessage | ErrorMessage) & { context?: MessageContext }
-export type { ChatEntry }
+export type ChatEntry = (ChatMessage | ErrorMessage) & { context?: MessageContext }
export interface ContextPayload {
content?: unknown
@@ -52,6 +51,8 @@ const ACTIVE_SESSION_STORAGE_KEY = 'chat/active-session'
export const CONTEXT_CHANNEL_NAME = 'airi-context-update'
export const CHAT_STREAM_CHANNEL_NAME = 'airi-chat-stream'
+type StreamingAssistantMessage = ChatAssistantMessage & { context?: MessageContext }
+
export const useChatStore = defineStore('chat', () => {
const { stream, discoverToolsCompatibility } = useLLM()
const { systemPrompt } = storeToRefs(useAiriCardStore())
@@ -60,6 +61,62 @@ export const useChatStore = defineStore('chat', () => {
const sessionMessages = useLocalStorage>(CHAT_STORAGE_KEY, {})
const sending = ref(false)
+ const streamingMessage = ref({ role: 'assistant', content: '', slices: [], tool_results: [] })
+ const sessionGenerations = ref>({})
+
+ interface SendOptions {
+ model: string
+ chatProvider: ChatProvider
+ providerConfig?: Record
+ attachments?: { type: 'image', data: string, mimeType: string }[]
+ tools?: StreamOptions['tools']
+ }
+
+ interface QueuedSend {
+ sendingMessage: string
+ options: SendOptions
+ generation: number
+ sessionId: string
+ cancelled?: boolean
+ deferred: {
+ resolve: () => void
+ reject: (error: unknown) => void
+ }
+ }
+
+ const pendingQueuedSends = ref([])
+
+ const sendQueue = createQueue({
+ handlers: [
+ async ({ data }) => {
+ const { sendingMessage, options, generation, deferred, sessionId, cancelled } = data
+
+ if (cancelled)
+ return
+
+ if (getSessionGeneration(sessionId) !== generation) {
+ deferred.reject(new Error('Chat session was reset before send could start'))
+ return
+ }
+
+ try {
+ await performSend(sendingMessage, options, generation, sessionId)
+ deferred.resolve()
+ }
+ catch (error) {
+ deferred.reject(error)
+ }
+ },
+ ],
+ })
+
+ sendQueue.on('enqueue', (queuedSend) => {
+ pendingQueuedSends.value = [...pendingQueuedSends.value, queuedSend]
+ })
+
+ sendQueue.on('dequeue', (queuedSend) => {
+ pendingQueuedSends.value = pendingQueuedSends.value.filter(item => item !== queuedSend)
+ })
// ----- Hooks (UI callbacks) -----
const onBeforeMessageComposedHooks = ref Promise>>([])
@@ -177,6 +234,22 @@ export const useChatStore = defineStore('chat', () => {
const codeBlockSystemPrompt = '- For any programming code block, always specify the programming language that supported on @shikijs/rehype on the rendered markdown, eg. ```python ... ```\n'
const mathSyntaxSystemPrompt = '- For any math equation, use LaTeX format, eg: $ x^3 $, always escape dollar sign outside math equation\n'
+ function ensureSessionGeneration(sessionId: string) {
+ if (sessionGenerations.value[sessionId] === undefined)
+ sessionGenerations.value = { ...sessionGenerations.value, [sessionId]: 0 }
+ }
+
+ function getSessionGeneration(sessionId: string) {
+ ensureSessionGeneration(sessionId)
+ return sessionGenerations.value[sessionId] ?? 0
+ }
+
+ function bumpSessionGeneration(sessionId: string) {
+ const nextGeneration = getSessionGeneration(sessionId) + 1
+ sessionGenerations.value = { ...sessionGenerations.value, [sessionId]: nextGeneration }
+ return nextGeneration
+ }
+
function generateInitialMessage() {
// TODO: compose, replace {{ user }} tag, etc
return {
@@ -186,6 +259,7 @@ export const useChatStore = defineStore('chat', () => {
}
function ensureSession(sessionId: string) {
+ ensureSessionGeneration(sessionId)
if (!sessionMessages.value[sessionId] || sessionMessages.value[sessionId].length === 0) {
sessionMessages.value[sessionId] = [{
...generateInitialMessage(),
@@ -200,6 +274,11 @@ export const useChatStore = defineStore('chat', () => {
ensureSession(activeSessionId.value)
+ function getSessionMessagesById(sessionId: string) {
+ ensureSession(sessionId)
+ return sessionMessages.value[sessionId]!
+ }
+
const messages = computed({
get: () => {
ensureSession(activeSessionId.value)
@@ -216,6 +295,7 @@ export const useChatStore = defineStore('chat', () => {
}
function cleanupMessages(sessionId = activeSessionId.value) {
+ bumpSessionGeneration(sessionId)
sessionMessages.value[sessionId] = [{
...generateInitialMessage(),
context: {
@@ -224,6 +304,17 @@ export const useChatStore = defineStore('chat', () => {
ts: Date.now(),
},
}]
+ // Reject pending sends for this session so callers don't hang after cleanup
+ for (const queued of pendingQueuedSends.value) {
+ if (queued.sessionId !== sessionId)
+ continue
+
+ queued.cancelled = true
+ queued.deferred.reject(new Error('Chat session was reset before send could start'))
+ }
+ pendingQueuedSends.value = pendingQueuedSends.value.filter(item => item.sessionId !== sessionId)
+ sending.value = false
+ streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
}
function getAllSessions() {
@@ -232,6 +323,7 @@ export const useChatStore = defineStore('chat', () => {
function replaceSessions(sessions: Record) {
sessionMessages.value = sessions
+ sessionGenerations.value = Object.fromEntries(Object.keys(sessions).map(sessionId => [sessionId, 0]))
const [firstSessionId] = Object.keys(sessions)
if (!sessionMessages.value[activeSessionId.value] && firstSessionId)
activeSessionId.value = firstSessionId
@@ -241,6 +333,7 @@ export const useChatStore = defineStore('chat', () => {
function resetAllSessions() {
sessionMessages.value = {}
+ sessionGenerations.value = {}
activeSessionId.value = 'default'
ensureSession(activeSessionId.value)
}
@@ -319,21 +412,28 @@ export const useChatStore = defineStore('chat', () => {
}
// ----- Send flow (user -> LLM -> assistant) -----
- const streamingMessage = ref({ role: 'assistant', content: '', slices: [], tool_results: [] })
-
- async function send(
+ async function performSend(
sendingMessage: string,
- options: {
- model: string
- chatProvider: ChatProvider
- providerConfig?: Record
- attachments?: { type: 'image', data: string, mimeType: string }[]
- tools?: StreamOptions['tools']
- },
+ options: SendOptions,
+ generation: number,
+ sessionId: string,
) {
if (!sendingMessage && !options.attachments?.length)
return
+
+ ensureSession(sessionId)
+
+ const isStaleGeneration = () => getSessionGeneration(sessionId) !== generation
+ const shouldAbort = () => isStaleGeneration()
+ if (shouldAbort())
+ return
sending.value = true
+ const assistantContext: MessageContext = {
+ sessionId,
+ source: 'llm',
+ ts: Date.now(),
+ }
+ streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [], context: assistantContext }
try {
await emitBeforeMessageComposedHooks(sendingMessage)
@@ -355,8 +455,12 @@ export const useChatStore = defineStore('chat', () => {
const finalContent = contentParts.length > 1 ? contentParts : sendingMessage
- const userContext: MessageContext = { sessionId: activeSessionId.value, source: 'text', ts: Date.now() }
- messages.value.push({ role: 'user', content: finalContent, context: userContext })
+ if (shouldAbort())
+ return
+
+ const sessionMessagesForSend = getSessionMessagesById(sessionId)
+ const userContext: MessageContext = { sessionId, source: 'text', ts: Date.now() }
+ sessionMessagesForSend.push({ role: 'user', content: finalContent, context: userContext })
publishContextMessage({
sessionId: userContext.sessionId,
@@ -368,6 +472,8 @@ export const useChatStore = defineStore('chat', () => {
const parser = useLlmmarkerParser({
onLiteral: async (literal) => {
+ if (shouldAbort())
+ return
await emitTokenLiteralHooks(literal)
streamingMessage.value.content += literal
@@ -385,6 +491,8 @@ export const useChatStore = defineStore('chat', () => {
})
},
onSpecial: async (special) => {
+ if (shouldAbort())
+ return
await emitTokenSpecialHooks(special)
},
minLiteralEmitLength: 24, // Avoid emitting literals too fast. This is a magic number and can be changed later.
@@ -393,6 +501,8 @@ export const useChatStore = defineStore('chat', () => {
const toolCallQueue = createQueue({
handlers: [
async (ctx) => {
+ if (shouldAbort())
+ return
if (ctx.data.type === 'tool-call') {
streamingMessage.value.slices.push(ctx.data)
return
@@ -405,9 +515,7 @@ export const useChatStore = defineStore('chat', () => {
],
})
- streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
-
- const newMessages = messages.value.map((msg) => {
+ const newMessages = sessionMessagesForSend.map((msg) => {
const { context: _context, ...withoutContext } = msg
const rawMessage = toRaw(withoutContext)
if (rawMessage.role === 'assistant') {
@@ -427,6 +535,9 @@ export const useChatStore = defineStore('chat', () => {
let fullText = ''
const headers = (options.providerConfig?.headers || {}) as Record
+ if (shouldAbort())
+ return
+
await stream(options.model, options.chatProvider, newMessages as Message[], {
headers,
tools: options.tools,
@@ -461,19 +572,13 @@ export const useChatStore = defineStore('chat', () => {
await parser.end()
// Add the completed message to the history only if it has content
- if (streamingMessage.value.slices.length > 0) {
- const assistantContext: MessageContext = {
- sessionId: activeSessionId.value,
- source: 'llm',
- ts: Date.now(),
- }
-
+ if (!isStaleGeneration() && streamingMessage.value.slices.length > 0) {
const assistantMessage: ChatEntry = {
...(toRaw(streamingMessage.value) as ChatAssistantMessage),
context: assistantContext,
}
- messages.value.push(assistantMessage)
+ sessionMessagesForSend.push(assistantMessage)
publishContextMessage({
sessionId: assistantContext.sessionId,
@@ -515,6 +620,24 @@ export const useChatStore = defineStore('chat', () => {
}
}
+ async function send(
+ sendingMessage: string,
+ options: SendOptions,
+ ) {
+ const sessionId = activeSessionId.value
+ const generation = getSessionGeneration(sessionId)
+
+ return new Promise((resolve, reject) => {
+ sendQueue.enqueue({
+ sendingMessage,
+ options,
+ generation,
+ sessionId,
+ deferred: { resolve, reject },
+ })
+ })
+ }
+
return {
sending,
activeSessionId,