refactor: web chat queue management and streamline chat route handling
This commit is contained in:
Soulter
@@ -28,5 +28,3 @@ pip_installer = PipInstaller(
|
||||
astrbot_config.get("pip_install_arg", ""),
|
||||
astrbot_config.get("pypi_index_url", None),
|
||||
)
|
||||
web_chat_queue = asyncio.Queue(maxsize=32)
|
||||
web_chat_back_queue = asyncio.Queue(maxsize=32)
|
||||
|
||||
@@ -23,7 +23,6 @@ from astrbot.core.provider.entities import (
|
||||
LLMResponse,
|
||||
)
|
||||
from astrbot.core.star.star_handler import EventType
|
||||
from astrbot.core import web_chat_back_queue
|
||||
from ..agent_runner.tool_loop_agent import ToolLoopAgent
|
||||
|
||||
|
||||
@@ -283,13 +282,6 @@ class LLMRequestSubStage(Stage):
|
||||
cid=cid,
|
||||
title=title,
|
||||
)
|
||||
web_chat_back_queue.put_nowait(
|
||||
{
|
||||
"type": "update_title",
|
||||
"cid": cid,
|
||||
"data": title,
|
||||
}
|
||||
)
|
||||
|
||||
async def _save_to_history(
|
||||
self,
|
||||
|
||||
@@ -2,7 +2,7 @@ import time
|
||||
import asyncio
|
||||
import uuid
|
||||
import os
|
||||
from typing import Awaitable, Any
|
||||
from typing import Awaitable, Any, Callable
|
||||
from astrbot.core.platform import (
|
||||
Platform,
|
||||
AstrBotMessage,
|
||||
@@ -13,7 +13,7 @@ from astrbot.core.platform import (
|
||||
from astrbot.core.message.message_event_result import MessageChain
|
||||
from astrbot.core.message.components import Plain, Image, Record # noqa: F403
|
||||
from astrbot import logger
|
||||
from astrbot.core import web_chat_queue
|
||||
from .webchat_queue_mgr import webchat_queue_mgr, WebChatQueueMgr
|
||||
from .webchat_event import WebChatMessageEvent
|
||||
from astrbot.core.platform.astr_message_event import MessageSesion
|
||||
from ...register import register_platform_adapter
|
||||
@@ -21,14 +21,46 @@ from astrbot.core.utils.astrbot_path import get_astrbot_data_path
|
||||
|
||||
|
||||
class QueueListener:
|
||||
def __init__(self, queue: asyncio.Queue, callback: callable) -> None:
|
||||
self.queue = queue
|
||||
def __init__(self, webchat_queue_mgr: WebChatQueueMgr, callback: Callable) -> None:
|
||||
self.webchat_queue_mgr = webchat_queue_mgr
|
||||
self.callback = callback
|
||||
self.running_tasks = set()
|
||||
|
||||
async def listen_to_queue(self, conversation_id: str):
|
||||
"""Listen to a specific conversation queue"""
|
||||
queue = self.webchat_queue_mgr.get_or_create_queue(conversation_id)
|
||||
while True:
|
||||
try:
|
||||
data = await queue.get()
|
||||
await self.callback(data)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error processing message from conversation {conversation_id}: {e}"
|
||||
)
|
||||
break
|
||||
|
||||
async def run(self):
|
||||
"""Monitor for new conversation queues and start listeners"""
|
||||
monitored_conversations = set()
|
||||
|
||||
while True:
|
||||
data = await self.queue.get()
|
||||
await self.callback(data)
|
||||
# Check for new conversations
|
||||
current_conversations = set(self.webchat_queue_mgr.queues.keys())
|
||||
new_conversations = current_conversations - monitored_conversations
|
||||
|
||||
# Start listeners for new conversations
|
||||
for conversation_id in new_conversations:
|
||||
task = asyncio.create_task(self.listen_to_queue(conversation_id))
|
||||
self.running_tasks.add(task)
|
||||
task.add_done_callback(self.running_tasks.discard)
|
||||
monitored_conversations.add(conversation_id)
|
||||
logger.debug(f"Started listener for conversation: {conversation_id}")
|
||||
|
||||
# Clean up monitored conversations that no longer exist
|
||||
removed_conversations = monitored_conversations - current_conversations
|
||||
monitored_conversations -= removed_conversations
|
||||
|
||||
await asyncio.sleep(1) # Check for new conversations every second
|
||||
|
||||
|
||||
@register_platform_adapter("webchat", "webchat")
|
||||
@@ -45,7 +77,7 @@ class WebChatAdapter(Platform):
|
||||
os.makedirs(self.imgs_dir, exist_ok=True)
|
||||
|
||||
self.metadata = PlatformMetadata(
|
||||
name="webchat", description="webchat", id=self.config.get("id")
|
||||
name="webchat", description="webchat", id=self.config.get("id", "")
|
||||
)
|
||||
|
||||
async def send_by_session(
|
||||
@@ -105,7 +137,7 @@ class WebChatAdapter(Platform):
|
||||
abm = await self.convert_message(data)
|
||||
await self.handle_msg(abm)
|
||||
|
||||
bot = QueueListener(web_chat_queue, callback)
|
||||
bot = QueueListener(webchat_queue_mgr, callback)
|
||||
return bot.run()
|
||||
|
||||
def meta(self) -> PlatformMetadata:
|
||||
|
||||
@@ -5,8 +5,8 @@ from astrbot.api import logger
|
||||
from astrbot.api.event import AstrMessageEvent, MessageChain
|
||||
from astrbot.api.message_components import Plain, Image, Record
|
||||
from astrbot.core.utils.io import download_image_by_url
|
||||
from astrbot.core import web_chat_back_queue
|
||||
from astrbot.core.utils.astrbot_path import get_astrbot_data_path
|
||||
from .webchat_queue_mgr import webchat_queue_mgr
|
||||
|
||||
imgs_dir = os.path.join(get_astrbot_data_path(), "webchat", "imgs")
|
||||
|
||||
@@ -18,13 +18,14 @@ class WebChatMessageEvent(AstrMessageEvent):
|
||||
|
||||
@staticmethod
|
||||
async def _send(message: MessageChain, session_id: str, streaming: bool = False):
|
||||
cid = session_id.split("!")[-1]
|
||||
web_chat_back_queue = webchat_queue_mgr.get_or_create_back_queue(cid)
|
||||
if not message:
|
||||
await web_chat_back_queue.put(
|
||||
{"type": "end", "data": "", "streaming": False}
|
||||
)
|
||||
return ""
|
||||
|
||||
cid = session_id.split("!")[-1]
|
||||
data = ""
|
||||
for comp in message.chain:
|
||||
if isinstance(comp, Plain):
|
||||
@@ -98,18 +99,22 @@ class WebChatMessageEvent(AstrMessageEvent):
|
||||
|
||||
async def send(self, message: MessageChain):
|
||||
await WebChatMessageEvent._send(message, session_id=self.session_id)
|
||||
cid = self.session_id.split("!")[-1]
|
||||
web_chat_back_queue = webchat_queue_mgr.get_or_create_back_queue(cid)
|
||||
await web_chat_back_queue.put(
|
||||
{
|
||||
"type": "end",
|
||||
"data": "",
|
||||
"streaming": False,
|
||||
"cid": self.session_id.split("!")[-1],
|
||||
"cid": cid,
|
||||
}
|
||||
)
|
||||
await super().send(message)
|
||||
|
||||
async def send_streaming(self, generator, use_fallback: bool = False):
|
||||
final_data = ""
|
||||
cid = self.session_id.split("!")[-1]
|
||||
web_chat_back_queue = webchat_queue_mgr.get_or_create_back_queue(cid)
|
||||
async for chain in generator:
|
||||
if chain.type == "break" and final_data:
|
||||
# 分割符
|
||||
@@ -118,7 +123,7 @@ class WebChatMessageEvent(AstrMessageEvent):
|
||||
"type": "end",
|
||||
"data": final_data,
|
||||
"streaming": True,
|
||||
"cid": self.session_id.split("!")[-1],
|
||||
"cid": cid,
|
||||
}
|
||||
)
|
||||
final_data = ""
|
||||
@@ -132,7 +137,7 @@ class WebChatMessageEvent(AstrMessageEvent):
|
||||
"type": "end",
|
||||
"data": final_data,
|
||||
"streaming": True,
|
||||
"cid": self.session_id.split("!")[-1],
|
||||
"cid": cid,
|
||||
}
|
||||
)
|
||||
await super().send_streaming(generator, use_fallback)
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
import asyncio
|
||||
|
||||
class WebChatQueueMgr:
|
||||
def __init__(self) -> None:
|
||||
self.queues = {}
|
||||
"""Conversation ID to asyncio.Queue mapping"""
|
||||
self.back_queues = {}
|
||||
"""Conversation ID to asyncio.Queue mapping for responses"""
|
||||
|
||||
def get_or_create_queue(self, conversation_id: str) -> asyncio.Queue:
|
||||
"""Get or create a queue for the given conversation ID"""
|
||||
if conversation_id not in self.queues:
|
||||
self.queues[conversation_id] = asyncio.Queue()
|
||||
return self.queues[conversation_id]
|
||||
|
||||
def get_or_create_back_queue(self, conversation_id: str) -> asyncio.Queue:
|
||||
"""Get or create a back queue for the given conversation ID"""
|
||||
if conversation_id not in self.back_queues:
|
||||
self.back_queues[conversation_id] = asyncio.Queue()
|
||||
return self.back_queues[conversation_id]
|
||||
|
||||
def remove_queues(self, conversation_id: str):
|
||||
"""Remove queues for the given conversation ID"""
|
||||
if conversation_id in self.queues:
|
||||
del self.queues[conversation_id]
|
||||
if conversation_id in self.back_queues:
|
||||
del self.back_queues[conversation_id]
|
||||
|
||||
def has_queue(self, conversation_id: str) -> bool:
|
||||
"""Check if a queue exists for the given conversation ID"""
|
||||
return conversation_id in self.queues
|
||||
|
||||
webchat_queue_mgr = WebChatQueueMgr()
|
||||
@@ -2,7 +2,7 @@ import uuid
|
||||
import json
|
||||
import os
|
||||
from .route import Route, Response, RouteContext
|
||||
from astrbot.core import web_chat_queue, web_chat_back_queue
|
||||
from astrbot.core.platform.sources.webchat.webchat_queue_mgr import webchat_queue_mgr
|
||||
from quart import request, Response as QuartResponse, g, make_response
|
||||
from astrbot.core.db import BaseDatabase
|
||||
import asyncio
|
||||
@@ -21,7 +21,6 @@ class ChatRoute(Route):
|
||||
super().__init__(context)
|
||||
self.routes = {
|
||||
"/chat/send": ("POST", self.chat),
|
||||
"/chat/listen": ("GET", self.listener),
|
||||
"/chat/new_conversation": ("GET", self.new_conversation),
|
||||
"/chat/conversations": ("GET", self.get_conversations),
|
||||
"/chat/get_conversation": ("GET", self.get_conversation),
|
||||
@@ -40,9 +39,6 @@ class ChatRoute(Route):
|
||||
|
||||
self.supported_imgs = ["jpg", "jpeg", "png", "gif", "webp"]
|
||||
|
||||
self.curr_user_cid = {}
|
||||
self.curr_chat_sse = {}
|
||||
|
||||
async def status(self):
|
||||
has_llm_enabled = (
|
||||
self.core_lifecycle.provider_manager.curr_provider_inst is not None
|
||||
@@ -133,21 +129,10 @@ class ChatRoute(Route):
|
||||
if not conversation_id:
|
||||
return Response().error("conversation_id is empty").__dict__
|
||||
|
||||
self.curr_user_cid[username] = conversation_id
|
||||
# Get conversation-specific queues
|
||||
back_queue = webchat_queue_mgr.get_or_create_back_queue(conversation_id)
|
||||
|
||||
await web_chat_queue.put(
|
||||
(
|
||||
username,
|
||||
conversation_id,
|
||||
{
|
||||
"message": message,
|
||||
"image_url": image_url, # list
|
||||
"audio_url": audio_url,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
# 持久化
|
||||
# append user message
|
||||
conversation = self.db.get_conversation_by_user_id(username, conversation_id)
|
||||
try:
|
||||
history = json.loads(conversation.history)
|
||||
@@ -164,30 +149,12 @@ class ChatRoute(Route):
|
||||
username, conversation_id, history=json.dumps(history)
|
||||
)
|
||||
|
||||
return Response().ok().__dict__
|
||||
|
||||
async def listener(self):
|
||||
"""一直保持长连接"""
|
||||
|
||||
username = g.get("username", "guest")
|
||||
|
||||
if username in self.curr_chat_sse:
|
||||
return Response().error("Already connected").__dict__
|
||||
|
||||
self.curr_chat_sse[username] = None
|
||||
|
||||
heartbeat = json.dumps({"type": "heartbeat", "data": "ping"})
|
||||
|
||||
async def stream():
|
||||
try:
|
||||
yield f"data: {heartbeat}\n\n" # 心跳包
|
||||
while True:
|
||||
try:
|
||||
result = await asyncio.wait_for(
|
||||
web_chat_back_queue.get(), timeout=10
|
||||
) # 设置超时时间为5秒
|
||||
result = await asyncio.wait_for(back_queue.get(), timeout=10)
|
||||
except asyncio.TimeoutError:
|
||||
yield f"data: {heartbeat}\n\n" # 心跳包
|
||||
continue
|
||||
|
||||
if not result:
|
||||
@@ -197,9 +164,6 @@ class ChatRoute(Route):
|
||||
type = result.get("type")
|
||||
cid = result.get("cid")
|
||||
streaming = result.get("streaming", False)
|
||||
if cid != self.curr_user_cid.get(username):
|
||||
# 丢弃
|
||||
continue
|
||||
yield f"data: {json.dumps(result, ensure_ascii=False)}\n\n"
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
@@ -210,6 +174,7 @@ class ChatRoute(Route):
|
||||
continue
|
||||
|
||||
if result_text:
|
||||
# append bot message
|
||||
conversation = self.db.get_conversation_by_user_id(
|
||||
username, cid
|
||||
)
|
||||
@@ -222,11 +187,25 @@ class ChatRoute(Route):
|
||||
self.db.update_conversation(
|
||||
username, cid, history=json.dumps(history)
|
||||
)
|
||||
break
|
||||
except BaseException as _:
|
||||
logger.debug(f"用户 {username} 断开聊天长连接。")
|
||||
self.curr_chat_sse.pop(username)
|
||||
return
|
||||
|
||||
# Put message to conversation-specific queue
|
||||
chat_queue = webchat_queue_mgr.get_or_create_queue(conversation_id)
|
||||
await chat_queue.put(
|
||||
(
|
||||
username,
|
||||
conversation_id,
|
||||
{
|
||||
"message": message,
|
||||
"image_url": image_url, # list
|
||||
"audio_url": audio_url,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
response = await make_response(
|
||||
stream(),
|
||||
{
|
||||
@@ -236,7 +215,6 @@ class ChatRoute(Route):
|
||||
"Connection": "keep-alive",
|
||||
},
|
||||
)
|
||||
response.timeout = None
|
||||
return response
|
||||
|
||||
async def delete_conversation(self):
|
||||
@@ -245,6 +223,8 @@ class ChatRoute(Route):
|
||||
if not conversation_id:
|
||||
return Response().error("Missing key: conversation_id").__dict__
|
||||
|
||||
# Clean up queues when deleting conversation
|
||||
webchat_queue_mgr.remove_queues(conversation_id)
|
||||
self.db.delete_conversation(username, conversation_id)
|
||||
return Response().ok().__dict__
|
||||
|
||||
@@ -279,6 +259,4 @@ class ChatRoute(Route):
|
||||
|
||||
conversation = self.db.get_conversation_by_user_id(username, conversation_id)
|
||||
|
||||
self.curr_user_cid[username] = conversation_id
|
||||
|
||||
return Response().ok(data=conversation).__dict__
|
||||
|
||||
+139
-272
@@ -187,8 +187,7 @@
|
||||
style="width: 85%; max-width: 900px; margin: 0 auto; border: 1px solid #e0e0e0; border-radius: 24px; padding: 4px;">
|
||||
<textarea id="input-field" v-model="prompt" @keydown="handleInputKeyDown"
|
||||
@click:clear="clearMessage" placeholder="Ask AstrBot..."
|
||||
style="width: 100%; resize: none; outline: none; border: 1px solid var(--v-theme-border); border-radius: 12px; padding: 12px 16px; min-height: 40px; font-family: inherit; font-size: 16px; background-color: var(--v-theme-surface);"
|
||||
:disabled="loadingChat"></textarea>
|
||||
style="width: 100%; resize: none; outline: none; border: 1px solid var(--v-theme-border); border-radius: 12px; padding: 12px 16px; min-height: 40px; font-family: inherit; font-size: 16px; background-color: var(--v-theme-surface);"></textarea>
|
||||
<div style="display: flex; justify-content: flex-end; margin-top: 8px;">
|
||||
<v-btn @click="sendMessage" icon="mdi-send" variant="text" color="deep-purple"
|
||||
:disabled="!prompt && stagedImagesName.length === 0 && !stagedAudioUrl"
|
||||
@@ -308,7 +307,6 @@ export default {
|
||||
|
||||
eventSource: null,
|
||||
eventSourceReader: null,
|
||||
sseReconnecting: false, // 添加重连状态标志
|
||||
|
||||
// // Ctrl键长按相关变量
|
||||
ctrlKeyDown: false,
|
||||
@@ -352,8 +350,6 @@ export default {
|
||||
if (from &&
|
||||
((from.path.startsWith('/chat') && to.path.startsWith('/chatbox')) ||
|
||||
(from.path.startsWith('/chatbox') && to.path.startsWith('/chat')))) {
|
||||
console.log('Route mode changed, reconnecting SSE...');
|
||||
this.reconnectSSE();
|
||||
}
|
||||
|
||||
// Check if the route matches /chat/<cid> or /chatbox/<cid> pattern
|
||||
@@ -394,7 +390,6 @@ export default {
|
||||
// Theme is now handled globally by the customizer store.
|
||||
// 设置输入框标签
|
||||
this.inputFieldLabel = this.tm('input.chatPrompt');
|
||||
this.startListeningEvent();
|
||||
this.checkStatus();
|
||||
this.getConversations();
|
||||
let inputField = document.getElementById('input-field');
|
||||
@@ -420,8 +415,6 @@ export default {
|
||||
},
|
||||
|
||||
beforeUnmount() {
|
||||
this.disconnectSSE();
|
||||
|
||||
// 移除keyup事件监听
|
||||
document.removeEventListener('keyup', this.handleInputKeyUp);
|
||||
|
||||
@@ -529,246 +522,10 @@ export default {
|
||||
}
|
||||
},
|
||||
|
||||
// 断开SSE连接
|
||||
disconnectSSE() {
|
||||
if (this.eventSourceReader) {
|
||||
try {
|
||||
this.eventSourceReader.cancel();
|
||||
console.log('SSE Reader cancelled');
|
||||
} catch (error) {
|
||||
console.warn('Error cancelling SSE reader:', error);
|
||||
}
|
||||
this.eventSourceReader = null;
|
||||
}
|
||||
|
||||
if (this.eventSource) {
|
||||
try {
|
||||
this.eventSource.cancel();
|
||||
console.log('SSE连接已断开');
|
||||
} catch (error) {
|
||||
console.warn('Error cancelling SSE:', error);
|
||||
}
|
||||
this.eventSource = null;
|
||||
}
|
||||
},
|
||||
|
||||
// 重新连接SSE
|
||||
async reconnectSSE() {
|
||||
if (this.sseReconnecting) {
|
||||
console.log('SSE reconnection already in progress');
|
||||
return;
|
||||
}
|
||||
|
||||
this.sseReconnecting = true;
|
||||
console.log('Reconnecting SSE...');
|
||||
this.disconnectSSE();
|
||||
|
||||
// 等待更长时间确保后端连接完全清理
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
|
||||
this.startListeningEvent();
|
||||
},
|
||||
|
||||
async startListeningEvent() {
|
||||
// 确保之前的连接已断开
|
||||
this.disconnectSSE();
|
||||
|
||||
// 如果正在重连过程中，等待一下
|
||||
if (this.sseReconnecting) {
|
||||
await new Promise(resolve => setTimeout(resolve, 500));
|
||||
}
|
||||
|
||||
let retryCount = 0;
|
||||
const maxRetries = 3;
|
||||
|
||||
while (retryCount < maxRetries) {
|
||||
try {
|
||||
console.log(`尝试建立SSE连接 (${retryCount + 1}/${maxRetries})`);
|
||||
|
||||
const response = await fetch('/api/chat/listen', {
|
||||
method: 'GET',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': 'Bearer ' + localStorage.getItem('token')
|
||||
}
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`SSE连接失败: ${response.statusText}`);
|
||||
}
|
||||
|
||||
const reader = response.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
this.eventSource = reader;
|
||||
this.eventSourceReader = reader;
|
||||
this.sseReconnecting = false;
|
||||
|
||||
let in_streaming = false;
|
||||
let message_obj = null;
|
||||
console.log('SSE连接已建立');
|
||||
// 显示连接成功状态
|
||||
if (retryCount > 0) {
|
||||
this.showConnectionStatus(this.tm('connection.status.reconnected'), 'success');
|
||||
}
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
console.log('SSE连接正常关闭');
|
||||
break;
|
||||
}
|
||||
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
|
||||
// 可能有多行
|
||||
let lines = chunk.split('\n\n');
|
||||
|
||||
console.log('SSE数据:', lines);
|
||||
|
||||
for (let i = 0; i < lines.length; i++) {
|
||||
let line = lines[i].trim();
|
||||
|
||||
if (!line) {
|
||||
continue;
|
||||
}
|
||||
|
||||
console.log(line); // 处理后端错误响应格式
|
||||
if (line.startsWith('{"status":"error"')) {
|
||||
try {
|
||||
const errorObj = JSON.parse(line);
|
||||
if (errorObj.message === 'Already connected') {
|
||||
throw new Error('CONNECTION_CONFLICT');
|
||||
}
|
||||
console.error('后端错误:', errorObj.message);
|
||||
continue;
|
||||
} catch (parseError) {
|
||||
if (parseError.message === 'CONNECTION_CONFLICT') {
|
||||
throw parseError;
|
||||
}
|
||||
console.warn('解析错误响应失败:', line);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// data: {"type": "plain", "data": "helloworld"}
|
||||
let chunk_json;
|
||||
try {
|
||||
chunk_json = JSON.parse(line.replace('data: ', ''));
|
||||
} catch (parseError) {
|
||||
console.warn('JSON解析失败:', line, parseError);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 检查解析后的数据是否有效
|
||||
if (!chunk_json || typeof chunk_json !== 'object') {
|
||||
console.warn('无效的数据对象:', chunk_json);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 检查是否有type字段
|
||||
if (!chunk_json.hasOwnProperty('type')) {
|
||||
console.warn('数据缺少type字段:', chunk_json);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (chunk_json.type === 'heartbeat') {
|
||||
continue; // 心跳包
|
||||
}
|
||||
if (chunk_json.type === 'error') {
|
||||
console.error('Error received:', chunk_json.data);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (chunk_json.type === 'image') {
|
||||
let img = chunk_json.data.replace('[IMAGE]', '');
|
||||
const imageUrl = await this.getMediaFile(img);
|
||||
let bot_resp = {
|
||||
type: 'bot',
|
||||
message: `<img src="${imageUrl}" style="max-width: 80%; border-radius: 8px; box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);"/>`
|
||||
}
|
||||
this.messages.push(bot_resp);
|
||||
} else if (chunk_json.type === 'record') {
|
||||
let audio = chunk_json.data.replace('[RECORD]', '');
|
||||
const audioUrl = await this.getMediaFile(audio);
|
||||
let bot_resp = {
|
||||
type: 'bot',
|
||||
message: `<audio controls class="audio-player">
|
||||
<source src="${audioUrl}" type="audio/wav">
|
||||
${this.t('messages.errors.browser.audioNotSupported')}
|
||||
</audio>`
|
||||
}
|
||||
this.messages.push(bot_resp);
|
||||
} else if (chunk_json.type === 'plain') {
|
||||
if (!in_streaming) {
|
||||
message_obj = {
|
||||
type: 'bot',
|
||||
message: this.ref(chunk_json.data),
|
||||
}
|
||||
this.messages.push(message_obj);
|
||||
in_streaming = true;
|
||||
} else {
|
||||
message_obj.message.value += chunk_json.data;
|
||||
}
|
||||
} else if (chunk_json.type === 'end') {
|
||||
in_streaming = false;
|
||||
// 在消息流结束后初始化代码复制按钮
|
||||
this.initCodeCopyButtons();
|
||||
continue;
|
||||
} else if (chunk_json.type === 'update_title') {
|
||||
// 更新对话标题
|
||||
const conversation = this.conversations.find(c => c.cid === chunk_json.cid);
|
||||
if (conversation) {
|
||||
conversation.title = chunk_json.data;
|
||||
}
|
||||
} else {
|
||||
console.warn('未知数据类型:', chunk_json.type);
|
||||
}
|
||||
this.scrollToBottom();
|
||||
}
|
||||
} catch (readError) {
|
||||
if (readError.name === 'AbortError') {
|
||||
console.log('SSE连接被取消');
|
||||
break;
|
||||
}
|
||||
if (readError.message === 'CONNECTION_CONFLICT') {
|
||||
throw readError;
|
||||
}
|
||||
console.error('SSE读取错误:', readError);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// 如果成功连接并正常结束，跳出重试循环
|
||||
break;
|
||||
|
||||
} catch (error) {
|
||||
console.error(`SSE连接错误 (尝试 ${retryCount + 1}):`, error);
|
||||
|
||||
retryCount++;
|
||||
if (error.message === 'CONNECTION_CONFLICT' && retryCount < maxRetries) {
|
||||
console.log(`连接冲突，等待 ${2000 * retryCount}ms 后重试...`);
|
||||
this.showConnectionStatus(`${this.tm('connection.status.reconnecting')} (${retryCount}/${maxRetries})`, 'warning');
|
||||
await new Promise(resolve => setTimeout(resolve, 2000 * retryCount));
|
||||
continue;
|
||||
}
|
||||
|
||||
if (retryCount >= maxRetries) {
|
||||
console.error('SSE连接重试次数已达上限');
|
||||
this.showConnectionStatus(this.tm('connection.status.failed'), 'error');
|
||||
this.sseReconnecting = false;
|
||||
break;
|
||||
}
|
||||
|
||||
// 等待一段时间后重试
|
||||
await new Promise(resolve => setTimeout(resolve, 1000 * retryCount));
|
||||
} finally {
|
||||
this.eventSource = null;
|
||||
this.eventSourceReader = null;
|
||||
}
|
||||
}
|
||||
|
||||
this.sseReconnecting = false;
|
||||
showConnectionStatus(message, type) {
|
||||
// You can implement a toast notification here or update UI status
|
||||
console.log(`Connection status: ${message} (${type})`);
|
||||
},
|
||||
|
||||
removeAudio() {
|
||||
@@ -920,7 +677,6 @@ export default {
|
||||
}
|
||||
}
|
||||
this.messages = message;
|
||||
// 初始化代码复制按钮
|
||||
this.initCodeCopyButtons();
|
||||
}).catch(err => {
|
||||
console.error(err);
|
||||
@@ -1032,33 +788,144 @@ export default {
|
||||
this.messages.push(userMessage);
|
||||
this.scrollToBottom();
|
||||
|
||||
this.loadingChat = true;
|
||||
this.loadingChat = true
|
||||
|
||||
fetch('/api/chat/send', {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': 'Bearer ' + localStorage.getItem('token')
|
||||
},
|
||||
body: JSON.stringify({
|
||||
message: this.prompt.trim(), // 确保发送的消息已去除前后空格
|
||||
conversation_id: this.currCid,
|
||||
image_url: this.stagedImagesName,
|
||||
audio_url: this.stagedAudioUrl ? [this.stagedAudioUrl] : []
|
||||
})
|
||||
})
|
||||
.then(response => {
|
||||
this.prompt = '';
|
||||
this.stagedImagesName = [];
|
||||
this.stagedImagesUrl = [];
|
||||
this.stagedAudioUrl = "";
|
||||
this.loadingChat = false;
|
||||
})
|
||||
.catch(err => {
|
||||
console.error(err);
|
||||
this.loadingChat = false;
|
||||
try {
|
||||
const response = await fetch('/api/chat/send', {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': 'Bearer ' + localStorage.getItem('token')
|
||||
},
|
||||
body: JSON.stringify({
|
||||
message: this.prompt.trim(), // 确保发送的消息已去除前后空格
|
||||
conversation_id: this.currCid,
|
||||
image_url: this.stagedImagesName,
|
||||
audio_url: this.stagedAudioUrl ? [this.stagedAudioUrl] : []
|
||||
})
|
||||
});
|
||||
|
||||
this.prompt = ''; // 清空输入框;
|
||||
|
||||
if (!response.ok) {
|
||||
throw new Error(`HTTP error! status: ${response.status}`);
|
||||
}
|
||||
|
||||
const reader = response.body.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let in_streaming = false;
|
||||
let message_obj = null;
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
console.log('SSE stream completed');
|
||||
break;
|
||||
}
|
||||
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
const lines = chunk.split('\n\n');
|
||||
|
||||
for (let i = 0; i < lines.length; i++) {
|
||||
let line = lines[i].trim();
|
||||
|
||||
if (!line) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Parse SSE data
|
||||
let chunk_json;
|
||||
try {
|
||||
chunk_json = JSON.parse(line.replace('data: ', ''));
|
||||
} catch (parseError) {
|
||||
console.warn('JSON解析失败:', line, parseError);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 检查解析后的数据是否有效
|
||||
if (!chunk_json || typeof chunk_json !== 'object' || !chunk_json.hasOwnProperty('type')) {
|
||||
console.warn('无效的数据对象:', chunk_json);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (chunk_json.type === 'heartbeat') {
|
||||
continue; // 心跳包
|
||||
}
|
||||
if (chunk_json.type === 'error') {
|
||||
console.error('Error received:', chunk_json.data);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (chunk_json.type === 'image') {
|
||||
let img = chunk_json.data.replace('[IMAGE]', '');
|
||||
const imageUrl = await this.getMediaFile(img);
|
||||
let bot_resp = {
|
||||
type: 'bot',
|
||||
message: `<img src="${imageUrl}" style="max-width: 80%; border-radius: 8px; box-shadow: 0 0 10px rgba(0, 0, 0, 0.1);"/>`
|
||||
}
|
||||
this.messages.push(bot_resp);
|
||||
} else if (chunk_json.type === 'record') {
|
||||
let audio = chunk_json.data.replace('[RECORD]', '');
|
||||
const audioUrl = await this.getMediaFile(audio);
|
||||
let bot_resp = {
|
||||
type: 'bot',
|
||||
message: `<audio controls class="audio-player">
|
||||
<source src="${audioUrl}" type="audio/wav">
|
||||
${this.t('messages.errors.browser.audioNotSupported')}
|
||||
</audio>`
|
||||
}
|
||||
this.messages.push(bot_resp);
|
||||
} else if (chunk_json.type === 'plain') {
|
||||
if (!in_streaming) {
|
||||
message_obj = {
|
||||
type: 'bot',
|
||||
message: this.ref(chunk_json.data),
|
||||
}
|
||||
this.messages.push(message_obj);
|
||||
in_streaming = true;
|
||||
} else {
|
||||
message_obj.message.value += chunk_json.data;
|
||||
}
|
||||
} else if (chunk_json.type === 'end') {
|
||||
in_streaming = false;
|
||||
// 在消息流结束后初始化代码复制按钮
|
||||
this.initCodeCopyButtons();
|
||||
continue;
|
||||
} else if (chunk_json.type === 'update_title') {
|
||||
// 更新对话标题
|
||||
const conversation = this.conversations.find(c => c.cid === chunk_json.cid);
|
||||
if (conversation) {
|
||||
conversation.title = chunk_json.data;
|
||||
}
|
||||
} else {
|
||||
console.warn('未知数据类型:', chunk_json.type);
|
||||
}
|
||||
this.scrollToBottom();
|
||||
}
|
||||
} catch (readError) {
|
||||
console.error('SSE读取错误:', readError);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Clear input after successful send
|
||||
this.prompt = '';
|
||||
this.stagedImagesName = [];
|
||||
this.stagedImagesUrl = [];
|
||||
this.stagedAudioUrl = "";
|
||||
this.loadingChat = false;
|
||||
|
||||
// get the latest conversations
|
||||
this.getConversations();
|
||||
|
||||
} catch (err) {
|
||||
console.error('发送消息失败:', err);
|
||||
this.loadingChat = false;
|
||||
this.showConnectionStatus(this.tm('connection.status.failed'), 'error');
|
||||
}
|
||||
},
|
||||
|
||||
scrollToBottom() {
|
||||
this.$nextTick(() => {
|
||||
const container = this.$refs.messageContainer;
|
||||
|
||||
Reference in New Issue
Block a user