Merge pull request #2082 from AstrBotDevs/fix-webchat-segment-reply

fix: 修复 WebChat 下可能消息错位的问题

astrbot/core/pipeline/scheduler.py

@@ -73,7 +73,7 @@ class PipelineScheduler:
         await self._process_stages(event)
 
         # 如果没有发送操作, 则发送一个空消息, 以便于后续的处理
-        if not event._has_send_oper and event.get_platform_name() == "webchat":
+        if event.get_platform_name() == "webchat":
             await event.send(None)
 
         logger.debug("pipeline 执行完毕。")

astrbot/core/platform/sources/webchat/webchat_event.py

@@ -22,7 +22,11 @@ class WebChatMessageEvent(AstrMessageEvent):
         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}
+                {
+                    "type": "end",
+                    "data": "",
+                    "streaming": False,
+                }  # end means this request is finished
             )
             return ""
 
@@ -99,16 +103,6 @@ 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": cid,
-            }
-        )
         await super().send(message)
 
     async def send_streaming(self, generator, use_fallback: bool = False):
@@ -120,7 +114,7 @@ class WebChatMessageEvent(AstrMessageEvent):
                 # 分割符
                 await web_chat_back_queue.put(
                     {
-                        "type": "end",
+                        "type": "break",  # break means a segment end
                         "data": final_data,
                         "streaming": True,
                         "cid": cid,
@@ -134,7 +128,7 @@ class WebChatMessageEvent(AstrMessageEvent):
 
         await web_chat_back_queue.put(
             {
-                "type": "end",
+                "type": "complete",  # complete means we return the final result
                 "data": final_data,
                 "streaming": True,
                 "cid": cid,

astrbot/dashboard/routes/chat.py

@@ -166,15 +166,12 @@ class ChatRoute(Route):
                     type = result.get("type")
                     cid = result.get("cid")
                     streaming = result.get("streaming", False)
-                    chain_type = result.get("chain_type")
                     yield f"data: {json.dumps(result, ensure_ascii=False)}\n\n"
                     await asyncio.sleep(0.05)
 
-                    if streaming and type != "end":
-                        # If the result is still streaming, we continue to wait for more data
-                        continue
-
-                    if result_text:
+                    if type == "end":
+                        break
+                    elif (streaming and type == "complete") or not streaming:
                         # append bot message
                         conversation = self.db.get_conversation_by_user_id(
                             username, cid
@@ -188,10 +185,6 @@ class ChatRoute(Route):
                         self.db.update_conversation(
                             username, cid, history=json.dumps(history)
                         )
-                        if chain_type not in ["tool_call", "tool_call_result"]:
-                            # If the result is not a tool call or tool call result,
-                            # we can break the loop and end the stream
-                            break
 
             except BaseException as _:
                 logger.debug(f"用户 {username} 断开聊天长连接。")

dashboard/src/views/ChatPage.vue

@@ -981,20 +981,19 @@ export default {
                                 } else {
                                     message_obj.message.value += chunk_json.data;
                                 }
-                            } else if (chunk_json.type === 'end') {
-                                in_streaming = false;
-                                // 在消息流结束后初始化代码复制按钮和图片点击事件
-                                this.initCodeCopyButtons();
-                                this.initImageClickEvents();
-                                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);
+                            }
+                            if ((chunk_json.type === 'break' && chunk_json.streaming) || !chunk_json.streaming) {
+                                // break means a segment end
+                                in_streaming = false;
+                                // 在消息流结束后初始化代码复制按钮和图片点击事件
+                                this.initCodeCopyButtons();
+                                this.initImageClickEvents();
                             }
                             this.scrollToBottom();
                         }