diff --git a/web/src/composables/useChatStream.ts b/web/src/composables/useChatStream.ts new file mode 100644 index 0000000..cf579ba --- /dev/null +++ b/web/src/composables/useChatStream.ts @@ -0,0 +1,107 @@ +import { ref } from 'vue' +import { useAuthStore } from '@/stores/auth' + +export type ChatStreamEventType = + | 'message_meta' + | 'text_delta' + | 'thinking_delta' + | 'usage' + | 'done' + | 'error' + +export interface ChatStreamEvent { + id: number + event: ChatStreamEventType + data: unknown +} + +export interface ChatStreamHandlers { + onEvent: (e: ChatStreamEvent) => void + onClose?: () => void + onError?: (err: Error) => void +} + +export interface ChatStreamHandle { + stop: () => void +} + +/** + * useChatStream 通过 fetch + ReadableStream 自实现 SSE 解析, + * 以支持 Authorization Bearer 头(EventSource 不支持自定义 header)。 + * start() 立即返回 stop 句柄,读流在后台 fire-and-forget。 + */ +export function useChatStream() { + const auth = useAuthStore() + const reading = ref(false) + + function start( + streamUrl: string, + since: number, + h: ChatStreamHandlers, + ): ChatStreamHandle { + const url = new URL(streamUrl, window.location.origin) + url.searchParams.set('since', String(since)) + const ctrl = new AbortController() + reading.value = true + + void (async () => { + try { + const headers: Record = { Accept: 'text/event-stream' } + if (auth.token) headers['Authorization'] = `Bearer ${auth.token}` + const resp = await fetch(url.toString(), { + headers, + signal: ctrl.signal, + }) + if (!resp.ok || !resp.body) { + throw new Error(`SSE HTTP ${resp.status}`) + } + const reader = resp.body.getReader() + const decoder = new TextDecoder() + let buf = '' + for (;;) { + const { value, done } = await reader.read() + if (done) break + buf += decoder.decode(value, { stream: true }) + let idx = buf.indexOf('\n\n') + while (idx >= 0) { + const raw = buf.slice(0, idx) + buf = buf.slice(idx + 2) + const ev = parseSSE(raw) + if (ev) h.onEvent(ev) + idx = buf.indexOf('\n\n') + } + } + h.onClose?.() + } catch (err) { + const e = err as { name?: string } + if (e?.name === 'AbortError') return + h.onError?.(err instanceof Error ? err : new Error(String(err))) + } finally { + reading.value = false + } + })() + + return { stop: () => ctrl.abort() } + } + + return { start, reading } +} + +function parseSSE(raw: string): ChatStreamEvent | null { + let id = 0 + let event = '' + const dataLines: string[] = [] + for (const line of raw.split('\n')) { + if (line.startsWith('id:')) id = Number(line.slice(3).trim()) + else if (line.startsWith('event:')) event = line.slice(6).trim() + else if (line.startsWith('data:')) dataLines.push(line.slice(5).trim()) + } + if (!event) return null + let data: unknown + try { + data = JSON.parse(dataLines.join('\n')) + } catch { + data = dataLines.join('\n') + } + return { id, event: event as ChatStreamEventType, data } +} diff --git a/web/src/stores/chat.ts b/web/src/stores/chat.ts new file mode 100644 index 0000000..4311873 --- /dev/null +++ b/web/src/stores/chat.ts @@ -0,0 +1,212 @@ +import { defineStore } from 'pinia' +import { computed, ref } from 'vue' + +import { + chatApi, + type ChatMessage, + type Conversation, + type ListConversationsParams, +} from '@/api/chat' +import { + useChatStream, + type ChatStreamEvent, + type ChatStreamHandle, +} from '@/composables/useChatStream' + +interface UsagePayload { + PromptTokens?: number + CompletionTokens?: number + ThinkingTokens?: number +} + +interface TextDeltaPayload { + text?: string +} + +interface ErrorPayload { + code?: string + message?: string +} + +interface MetaPayload { + id?: number + role?: string + status?: ChatMessage['status'] +} + +export const useChatStore = defineStore('chat', () => { + const conversations = ref([]) + const currentConversation = ref(null) + const messages = ref([]) + const streamingMessageId = ref(null) + const streamHandle = ref(null) + + const isStreaming = computed(() => streamingMessageId.value !== null) + + async function loadConversations(filter: ListConversationsParams = {}) { + conversations.value = await chatApi.listConversations(filter) + } + + async function openConversation(id: string) { + detachStream() + const detail = await chatApi.getConversation(id) + currentConversation.value = detail.conversation + messages.value = detail.messages + const pending = detail.messages.find((m) => m.status === 'pending') + if (pending) { + streamingMessageId.value = pending.id + attachStream(`/api/v1/conversations/${id}/stream`, pending.id - 1) + } + } + + async function sendMessage(content: string, attachmentIds: string[] = []) { + if (!currentConversation.value) return + const conv = currentConversation.value + const resp = await chatApi.sendMessage(conv.id, { + content, + attachment_ids: attachmentIds, + }) + const now = new Date().toISOString() + messages.value.push({ + id: resp.user_message_id, + conversation_id: conv.id, + role: 'user', + content, + status: 'ok', + created_at: now, + updated_at: now, + }) + messages.value.push({ + id: resp.assistant_message_id, + conversation_id: conv.id, + role: 'assistant', + content: '', + status: 'pending', + created_at: now, + updated_at: now, + }) + streamingMessageId.value = resp.assistant_message_id + attachStream(resp.stream_url, resp.user_message_id) + } + + async function cancelMessage() { + const id = streamingMessageId.value + if (id == null) return + await chatApi.cancelMessage(id) + detachStream() + } + + async function retryMessage(messageId: number) { + if (!currentConversation.value) return + const resp = await chatApi.retryMessage(messageId) + const idx = messages.value.findIndex((m) => m.id === messageId) + if (idx >= 0) { + const old = messages.value[idx] + messages.value[idx] = { + ...old, + id: resp.assistant_message_id, + content: '', + thinking: null, + status: 'pending', + error_message: null, + } + } + streamingMessageId.value = resp.assistant_message_id + attachStream(resp.stream_url, resp.assistant_message_id - 1) + } + + async function deleteConversation(id: string) { + await chatApi.deleteConversation(id) + conversations.value = conversations.value.filter((c) => c.id !== id) + if (currentConversation.value?.id === id) { + currentConversation.value = null + messages.value = [] + detachStream() + } + } + + function attachStream(streamUrl: string, sinceMessageID: number) { + detachStream() + const stream = useChatStream() + streamHandle.value = stream.start(streamUrl, sinceMessageID, { + onEvent: (ev) => applyStreamEvent(ev), + onClose: () => { + streamHandle.value = null + }, + onError: (err) => { + console.error('chat stream error', err) + const id = streamingMessageId.value + if (id != null) { + const msg = messages.value.find((m) => m.id === id) + if (msg && msg.status === 'pending') { + msg.status = 'error' + msg.error_message = err.message + } + } + streamingMessageId.value = null + streamHandle.value = null + }, + }) + } + + function detachStream() { + streamHandle.value?.stop() + streamHandle.value = null + } + + function applyStreamEvent(ev: ChatStreamEvent) { + const targetId = streamingMessageId.value + if (targetId == null) return + const msg = messages.value.find((m) => m.id === targetId) + if (!msg) return + switch (ev.event) { + case 'message_meta': { + const d = ev.data as MetaPayload + if (d.status) msg.status = d.status + break + } + case 'text_delta': { + const d = ev.data as TextDeltaPayload + if (d.text) msg.content += d.text + break + } + case 'thinking_delta': { + const d = ev.data as TextDeltaPayload + if (d.text) msg.thinking = (msg.thinking ?? '') + d.text + break + } + case 'usage': { + const u = ev.data as UsagePayload + msg.prompt_tokens = u.PromptTokens ?? msg.prompt_tokens + msg.completion_tokens = u.CompletionTokens ?? msg.completion_tokens + msg.thinking_tokens = u.ThinkingTokens ?? msg.thinking_tokens + break + } + case 'done': + msg.status = 'ok' + streamingMessageId.value = null + break + case 'error': { + const d = ev.data as ErrorPayload + msg.status = 'error' + msg.error_message = d.message ?? d.code ?? 'stream error' + streamingMessageId.value = null + break + } + } + } + + return { + conversations, + currentConversation, + messages, + streamingMessageId, + isStreaming, + loadConversations, + openConversation, + sendMessage, + cancelMessage, + retryMessage, + deleteConversation, + } +})