dieya/dieya_risk_prediction_frontend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

支持多会话并发流式处理

Browse Source
This commit is contained in:
kx_guo 2026-09-11 17:06:31 +08:00
parent 8ccc9755cb
commit 73f574add4
2 changed files with 164 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/pages/RiskAssistantPage.vue
View File

@ -390,7 +390,7 @@ async function submit(
const value = prompt.value.trim(); const value = prompt.value.trim();
if (!value || loading.value) return; if (!value || loading.value) return;
prompt.value = ""; prompt.value = "";
await conversationStore.streamSend( const sentConversation = await conversationStore.streamSend(
value, value,
(delta) => { (delta) => {
void delta; void delta;
@ -398,15 +398,16 @@ async function submit(
messageType, messageType,
); );
if ( if (
conversation.value && sentConversation &&
route.query.conversation !== String(conversation.value.id) conversation.value?.id === sentConversation.id &&
route.query.conversation !== String(sentConversation.id)
) { ) {
await router.replace({ await router.replace({
name: "assessment", name: "assessment",
query: { conversation: String(conversation.value.id) }, query: { conversation: String(sentConversation.id) },
}); });
} }
await scrollToBottom(); if (conversation.value?.id === sentConversation?.id) await scrollToBottom();
} }
async function scrollToBottom(behavior: "smooth" | "auto" = "smooth") { async function scrollToBottom(behavior: "smooth" | "auto" = "smooth") {

244
src/stores/conversation.store.ts
View File

@ -1,5 +1,5 @@
import { defineStore } from "pinia"; import { defineStore } from "pinia";
import { ref } from "vue"; import { computed, reactive, ref } from "vue";
import { import {
createConversation, createConversation,
@ -17,12 +17,19 @@ export const useConversationStore = defineStore("conversation", () => {
const items = ref<ConversationSummary[]>([]); const items = ref<ConversationSummary[]>([]);
const current = ref<Conversation | null>(null); const current = ref<Conversation | null>(null);
const loading = ref(false); const loading = ref(false);
const sending = ref(false); const plainSending = ref(false);
const loadingOlder = ref(false); const loadingOlder = ref(false);
const error = ref<string | null>(null); const error = ref<string | null>(null);
const generating = ref(false); type StreamState = {
let activeController: AbortController | null = null; controller: AbortController;
let activeUserMessageId: number | null = null; userMessageId: number;
assistantMessageId: number;
serverUserMessageId: number | null;
};
const streamStates = reactive(new Map<number, StreamState>());
const conversationCache = reactive(new Map<number, Conversation>());
const sending = computed(() => plainSending.value || Boolean(current.value && streamStates.has(current.value.id)));
const generating = sending;
let listRequest: Promise<void> | null = null; let listRequest: Promise<void> | null = null;
let listRequestSearch: string | undefined; let listRequestSearch: string | undefined;
let listLoadedAt = 0; let listLoadedAt = 0;
@ -62,7 +69,19 @@ export const useConversationStore = defineStore("conversation", () => {
try { try {
const response = await fetchConversation(id); const response = await fetchConversation(id);
if (!response.data.data) throw new Error("对话响应为空"); if (!response.data.data) throw new Error("对话响应为空");
current.value = response.data.data; const remote = response.data.data;
const cached = conversationCache.get(id);
const stream = streamStates.get(id);
const merged = stream && cached
? {
...remote,
messages: mergePendingMessages(remote.messages, cached.messages, stream),
messageCount: Math.max(remote.messageCount, cached.messageCount),
messagesTotal: Math.max(remote.messagesTotal, cached.messagesTotal),
}
: remote;
conversationCache.set(id, merged);
current.value = merged;
} catch (reason) { } catch (reason) {
error.value = reason instanceof Error ? reason.message : "对话加载失败"; error.value = reason instanceof Error ? reason.message : "对话加载失败";
throw reason; throw reason;
@ -80,20 +99,24 @@ export const useConversationStore = defineStore("conversation", () => {
) { ) {
return false; return false;
} }
const offset = current.value.messagesOffset + current.value.messages.length; const conversationId = current.value.id;
const base = conversationCache.get(conversationId) ?? current.value;
const offset = base.messagesOffset + base.messages.length;
if (!current.value.hasMoreMessages) return false; if (!current.value.hasMoreMessages) return false;
loadingOlder.value = true; loadingOlder.value = true;
try { try {
const response = await fetchConversation(current.value.id, offset); const response = await fetchConversation(conversationId, offset);
if (!response.data.data) throw new Error("历史消息响应为空"); if (!response.data.data) throw new Error("历史消息响应为空");
const older = response.data.data; const older = response.data.data;
current.value = { const updated = {
...current.value, ...base,
messages: [...older.messages, ...current.value.messages], messages: [...older.messages, ...base.messages],
messagesOffset: 0, messagesOffset: 0,
hasMoreMessages: older.hasMoreMessages, hasMoreMessages: older.hasMoreMessages,
messagesTotal: older.messagesTotal, messagesTotal: older.messagesTotal,
}; };
conversationCache.set(conversationId, updated);
if (current.value?.id === conversationId) current.value = updated;
return true; return true;
} finally { } finally {
loadingOlder.value = false; loadingOlder.value = false;
@ -103,6 +126,7 @@ export const useConversationStore = defineStore("conversation", () => {
async function create(title?: string) { async function create(title?: string) {
const response = await createConversation(title); const response = await createConversation(title);
if (!response.data.data) throw new Error("新建对话响应为空"); if (!response.data.data) throw new Error("新建对话响应为空");
conversationCache.set(response.data.data.id, response.data.data);
current.value = response.data.data; current.value = response.data.data;
items.value = [toSummary(response.data.data), ...items.value]; items.value = [toSummary(response.data.data), ...items.value];
return current.value; return current.value;
@ -114,29 +138,30 @@ export const useConversationStore = defineStore("conversation", () => {
) { ) {
if (!current.value) await create(); if (!current.value) await create();
if (!current.value) throw new Error("当前对话不存在"); if (!current.value) throw new Error("当前对话不存在");
sending.value = true; const conversationId = current.value.id;
plainSending.value = true;
error.value = null; error.value = null;
try { try {
const response = await sendConversationMessage( const response = await sendConversationMessage(
current.value.id, conversationId,
content, content,
messageType, messageType,
); );
if (!response.data.data) throw new Error("消息响应为空"); if (!response.data.data) throw new Error("消息响应为空");
current.value = response.data.data; setConversation(response.data.data);
const index = items.value.findIndex( const index = items.value.findIndex(
(item) => item.id === current.value?.id, (item) => item.id === conversationId,
); );
const summary = toSummary(current.value); const summary = toSummary(response.data.data);
if (index >= 0) items.value.splice(index, 1, summary); if (index >= 0) items.value.splice(index, 1, summary);
else items.value.unshift(summary); else items.value.unshift(summary);
items.value.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); items.value.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
return current.value; return response.data.data;
} catch (reason) { } catch (reason) {
error.value = reason instanceof Error ? reason.message : "消息发送失败"; error.value = reason instanceof Error ? reason.message : "消息发送失败";
throw reason; throw reason;
} finally { } finally {
sending.value = false; plainSending.value = false;
} }
} }
@ -147,13 +172,19 @@ export const useConversationStore = defineStore("conversation", () => {
) { ) {
if (!current.value) await create(); if (!current.value) await create();
if (!current.value) throw new Error("当前对话不存在"); if (!current.value) throw new Error("当前对话不存在");
sending.value = true; const conversationId = current.value.id;
generating.value = true; if (streamStates.has(conversationId)) throw new Error("当前会话正在生成，请稍候");
activeController = new AbortController(); const state: StreamState = {
controller: new AbortController(),
userMessageId: -Date.now(),
assistantMessageId: -Date.now() - 1,
serverUserMessageId: null,
};
streamStates.set(conversationId, state);
error.value = null; error.value = null;
const startedAt = new Date().toISOString(); const startedAt = new Date().toISOString();
const userMessage = { const userMessage = {
id: -Date.now(), id: state.userMessageId,
role: "user" as const, role: "user" as const,
messageType, messageType,
content, content,
@ -162,9 +193,8 @@ export const useConversationStore = defineStore("conversation", () => {
semantic: null, semantic: null,
createdAt: startedAt, createdAt: startedAt,
}; };
activeUserMessageId = userMessage.id;
const assistantMessage = { const assistantMessage = {
id: userMessage.id - 1, id: state.assistantMessageId,
role: "assistant" as const, role: "assistant" as const,
messageType: "text" as const, messageType: "text" as const,
content: "", content: "",
@ -173,14 +203,15 @@ export const useConversationStore = defineStore("conversation", () => {
semantic: null as Record<string, unknown> | null, semantic: null as Record<string, unknown> | null,
createdAt: startedAt, createdAt: startedAt,
}; };
current.value = { const optimistic = {
...current.value, ...current.value,
messages: [...current.value.messages, userMessage, assistantMessage], messages: [...current.value.messages, userMessage, assistantMessage],
messageCount: current.value.messageCount + 2, messageCount: current.value.messageCount + 2,
messagesTotal: current.value.messagesTotal + 2, messagesTotal: current.value.messagesTotal + 2,
}; };
conversationCache.set(conversationId, optimistic);
current.value = optimistic;
try { try {
const conversationId = current.value.id;
await streamConversationMessage( await streamConversationMessage(
conversationId, conversationId,
content, content,
@ -191,132 +222,156 @@ export const useConversationStore = defineStore("conversation", () => {
typeof event.data === "object" && typeof event.data === "object" &&
"messageId" in event.data && "messageId" in event.data &&
typeof event.data.messageId === "number" && typeof event.data.messageId === "number" &&
current.value conversationCache.get(conversationId)
) { ) {
const serverMessageId = event.data.messageId; const serverMessageId = event.data.messageId;
const messages = current.value.messages.map((message) => state.serverUserMessageId = serverMessageId;
message.id === userMessage.id const cached = conversationCache.get(conversationId);
if (!cached) return;
const messages = cached.messages.map((message) =>
message.id === state.userMessageId
? { ...message, id: serverMessageId } ? { ...message, id: serverMessageId }
: message, : message,
); );
activeUserMessageId = serverMessageId; setConversation({ ...cached, messages });
current.value = { ...current.value, messages };
} }
if (event.type === "delta" && event.content) onDelta(event.content); if (event.type === "delta" && event.content) onDelta(event.content);
if (event.type === "semantic" && event.data && current.value) { if (event.type === "semantic" && event.data) {
const messages = [...current.value.messages]; const cached = conversationCache.get(conversationId);
const last = messages[messages.length - 1]; if (!cached) return;
if (last?.role === "assistant") { const messages = cached.messages.map((message) =>
messages[messages.length - 1] = { message.id === state.assistantMessageId
...last, ? {
...message,
semantic: event.data as Record<string, unknown>, semantic: event.data as Record<string, unknown>,
};
current.value = { ...current.value, messages };
} }
: message,
);
setConversation({ ...cached, messages });
} }
if (event.type === "delta" && event.content && current.value) { if (event.type === "delta" && event.content) {
const messages = [...current.value.messages]; const cached = conversationCache.get(conversationId);
const last = messages[messages.length - 1]; if (!cached) return;
if (last?.role === "assistant") { const messages = cached.messages.map((message) =>
messages[messages.length - 1] = { message.id === state.assistantMessageId
...last, ? { ...message, content: message.content + event.content }
content: last.content + event.content, : message,
}; );
current.value = { ...current.value, messages }; setConversation({ ...cached, messages });
}
} }
if (event.type === "error") if (event.type === "error")
throw new Error(event.message || "AI 回复失败"); throw new Error(event.message || "AI 回复失败");
if (event.type === "done" && event.data) if (event.type === "done" && event.data)
current.value = event.data as Conversation; setConversation(event.data as Conversation);
}, },
messageType, messageType,
activeController.signal, state.controller.signal,
); );
const completed = conversationCache.get(conversationId);
if ( if (
!current.value || !completed ||
!current.value.messages.some( !completed.messages.some(
(message) => message.role === "assistant" && message.content.trim(), (message) => message.role === "assistant" && message.content.trim(),
) )
) { ) {
throw new Error("AI 回复为空"); throw new Error("AI 回复为空");
} }
const summary = toSummary(current.value); const summary = toSummary(completed);
const index = items.value.findIndex((item) => item.id === summary.id); const index = items.value.findIndex((item) => item.id === summary.id);
if (index >= 0) items.value.splice(index, 1, summary); if (index >= 0) items.value.splice(index, 1, summary);
else items.value.unshift(summary); else items.value.unshift(summary);
return current.value; return completed;
} catch (reason) { } catch (reason) {
const conversationId = current.value?.id; if (state.controller.signal.aborted) {
if (activeController?.signal.aborted) {
if (conversationId) {
try { try {
const response = await fetchConversation(conversationId); const response = await fetchConversation(conversationId);
if (response.data.data) current.value = response.data.data; if (response.data.data) setConversation(response.data.data);
} catch { } catch {
// If the request was cancelled before persistence, remove both // If the request was cancelled before persistence, remove both
// temporary messages from the local conversation. // temporary messages from the local conversation.
if (current.value) { const cached = conversationCache.get(conversationId);
const messages = current.value.messages.filter( if (cached) {
const messages = cached.messages.filter(
(message) => (message) =>
message.id !== assistantMessage.id && message.id !== assistantMessage.id &&
message.id !== userMessage.id, message.id !== userMessage.id,
); );
current.value = { setConversation({
...current.value, ...cached,
messages, messages,
messageCount: Math.max(0, current.value.messageCount - 2), messageCount: Math.max(0, cached.messageCount - 2),
messagesTotal: Math.max(0, current.value.messagesTotal - 2), messagesTotal: Math.max(0, cached.messagesTotal - 2),
}; });
} }
} }
const summary = current.value ? toSummary(current.value) : null; const completed = conversationCache.get(conversationId);
if (summary) { if (completed) upsertSummary(completed);
const index = items.value.findIndex( return completed ?? null;
(item) => item.id === summary.id,
);
if (index >= 0) items.value.splice(index, 1, summary);
else items.value.unshift(summary);
} }
} const cached = conversationCache.get(conversationId);
return current.value; if (cached) {
} const messages = cached.messages.filter(
if (current.value) {
const messages = current.value.messages.filter(
(message) => (message) =>
message.id !== assistantMessage.id && message.id !== userMessage.id, message.id !== assistantMessage.id && message.id !== userMessage.id,
); );
current.value = { setConversation({
...current.value, ...cached,
messages, messages,
messageCount: Math.max(0, current.value.messageCount - 2), messageCount: Math.max(0, cached.messageCount - 2),
messagesTotal: Math.max(0, current.value.messagesTotal - 2), messagesTotal: Math.max(0, cached.messagesTotal - 2),
}; });
} }
if (conversationId) {
try { try {
const response = await fetchConversation(conversationId); const response = await fetchConversation(conversationId);
if (response.data.data) current.value = response.data.data; if (response.data.data) setConversation(response.data.data);
} catch { } catch {
// Keep the local state when recovery is unavailable. // Keep the cached state when recovery is unavailable.
}
} }
if (current.value?.id === conversationId)
error.value = reason instanceof Error ? reason.message : "AI 回复失败"; error.value = reason instanceof Error ? reason.message : "AI 回复失败";
throw reason; throw reason;
} finally { } finally {
sending.value = false; streamStates.delete(conversationId);
generating.value = false;
activeController = null;
activeUserMessageId = null;
} }
} }
function stopGeneration() { function stopGeneration(conversationId = current.value?.id) {
activeController?.abort(); if (conversationId) streamStates.get(conversationId)?.controller.abort();
} }
function isGeneratingMessage(messageId: number) { function isGeneratingMessage(messageId: number) {
return activeUserMessageId === messageId && generating.value; const state = current.value ? streamStates.get(current.value.id) : undefined;
return Boolean(state && (state.userMessageId === messageId || state.serverUserMessageId === messageId));
}
function setConversation(conversation: Conversation) {
conversationCache.set(conversation.id, conversation);
if (current.value?.id === conversation.id) current.value = conversation;
upsertSummary(conversation);
}
function mergePendingMessages(
remoteMessages: Conversation["messages"],
cachedMessages: Conversation["messages"],
state: StreamState,
) {
const known = new Set(remoteMessages.map((message) => message.id));
const pending = cachedMessages.filter(
(message) =>
(message.id === state.assistantMessageId ||
message.id === state.userMessageId ||
message.id === state.serverUserMessageId) &&
!known.has(message.id),
);
return [...remoteMessages, ...pending];
}
function upsertSummary(conversation: Conversation) {
const summary = toSummary(conversation);
const index = items.value.findIndex((item) => item.id === summary.id);
if (index >= 0) items.value.splice(index, 1, summary);
else items.value.unshift(summary);
items.value.sort((a, b) => b.updatedAt.localeCompare(a.updatedAt));
} }
async function withdraw(messageId: number) { async function withdraw(messageId: number) {
@ -330,6 +385,7 @@ export const useConversationStore = defineStore("conversation", () => {
); );
if (!response.data.data) throw new Error("撤回响应为空"); if (!response.data.data) throw new Error("撤回响应为空");
current.value = response.data.data; current.value = response.data.data;
conversationCache.set(current.value.id, current.value);
const index = items.value.findIndex( const index = items.value.findIndex(
(item) => item.id === current.value?.id, (item) => item.id === current.value?.id,
); );