client: 增加ACK支持

Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md

@@ -0,0 +1,124 @@
+# ACK机制支持指南
+
+## 概述
+
+客户端已实现基于emit机制的ACK确认功能，允许客户端在发送消息后等待服务器确认，从而支持同步和异步两种通信模式。
+
+## 客户端改动
+
+### 1. 新增参数
+
+客户端的三个主要发送方法 `send_text`、`send_voice` 和 `send_image` 都新增了以下参数：
+
+- `wait_for_ack: bool = False` - 是否等待服务器ACK确认
+- `timeout: int = 5` - 等待ACK的超时时间（秒）
+
+### 2. 消息格式变化
+
+客户端发送的消息中新增了 `ack_id` 字段，格式为 `{id}_{timestamp}`，其中：
+- `id` 是客户端ID
+- `timestamp` 是当前时间的格式化字符串（精确到微秒）
+
+### 3. ACK事件
+
+客户端监听 `ack` 事件，用于接收服务器的确认消息。
+
+## 后端WebSocket服务器需要的支持
+
+### 1. 接收消息处理
+
+当服务器收到客户端发送的消息时，需要：
+
+1. 检查消息中是否包含 `ack_id` 字段
+2. 如果包含 `ack_id`，立即发送ACK确认消息，不需要等待后续业务逻辑执行
+3. 继续执行后续的业务逻辑
+
+### 2. ACK消息格式
+
+服务器需要向客户端发送 `ack` 事件，消息格式如下：
+
+```json
+{
+  "ack_id": "客户端发送的ack_id值"
+}
+```
+
+### 3. 发送ACK的时机
+
+**非常重要**：服务器必须在收到客户端消息后**立即**发送ACK确认，而不是在业务逻辑执行完成后发送。这样可以确保客户端能够快速收到确认，继续执行后续操作。
+
+### 4. 实现示例（Python Socket.IO）
+
+```python
+# 服务器端示例代码
+@sio.on('text_message')  # 根据实际事件名调整
+def handle_text_message(sid, data):
+    # 1. 立即发送ACK确认
+    if 'ack_id' in data:
+        sio.emit('ack', {'ack_id': data['ack_id']}, room=sid)
+    
+    # 2. 执行后续业务逻辑（可能耗时较长）
+    # ... 业务逻辑代码 ...
+    
+@sio.on('voice_message')  # 根据实际事件名调整
+def handle_voice_message(sid, data):
+    # 1. 立即发送ACK确认
+    if 'ack_id' in data:
+        sio.emit('ack', {'ack_id': data['ack_id']}, room=sid)
+    
+    # 2. 执行后续业务逻辑（可能耗时较长）
+    # ... 业务逻辑代码 ...
+    
+@sio.on('image_message')  # 根据实际事件名调整
+def handle_image_message(sid, data):
+    # 1. 立即发送ACK确认
+    if 'ack_id' in data:
+        sio.emit('ack', {'ack_id': data['ack_id']}, room=sid)
+    
+    # 2. 执行后续业务逻辑（可能耗时较长）
+    # ... 业务逻辑代码 ...
+```
+
+## 使用示例
+
+### 客户端同步调用（等待ACK）
+
+```python
+client = ASEngineClient(server_url, namespace, id)
+client.connect()
+
+# 同步调用：等待ACK确认后再继续执行
+result = client.send_text('text_message', 'Hello', wait_for_ack=True, timeout=3)
+if result:
+    print("消息已被服务器确认收到")
+    # 执行后续操作
+else:
+    print("等待ACK超时")
+```
+
+### 客户端异步调用（不等待ACK）
+
+```python
+client = ASEngineClient(server_url, namespace, id)
+client.connect()
+
+# 异步调用：发送后立即返回，不等待ACK
+result = client.send_text('text_message', 'Hello', wait_for_ack=False)
+if result:
+    print("消息已发送")
+    # 执行后续操作，不需要等待ACK
+```
+
+## 注意事项
+
+1. **向后兼容**：客户端的改动保持了向后兼容性，现有代码无需修改即可继续使用。
+2. **超时处理**：客户端会在指定的超时时间后停止等待ACK，并返回False。
+3. **资源清理**：客户端会在收到ACK或超时时清理相关的ACK状态，避免内存泄漏。
+4. **唯一标识**：`ack_id` 确保了每个消息的唯一性，避免ACK混淆。
+5. **性能考虑**：等待ACK会阻塞当前线程，建议仅在必要时使用同步模式。
+
+## 总结
+
+通过实现基于emit机制的ACK确认功能，客户端和服务器之间的通信可以灵活支持同步和异步两种模式。服务器需要做的改动非常简单，只需要在收到消息后立即发送ACK确认即可。
+
+这种设计既保证了通信的可靠性，又不会影响业务逻辑的执行效率，是一种高效且可靠的通信方式。

Html/apps/extension_ase/ase_client/client.py

@@ -5,6 +5,7 @@ import os
 from datetime import datetime
 from typing import Dict, Callable, Any, Optional
 from functools import partial
+from threading import Event
 
 class ASEngineClient:
     def __init__(self, server_url: str, namespace: str, id:str):
@@ -29,8 +30,9 @@ class ASEngineClient:
         self._on_connect_callback_entry=None
         self._on_connect_callback_initData=None
         
-
-
+        # ACK相关
+        self.ack_events: Dict[str, Event] = {}  # 存储ack_id和对应的Event对象
+        
         # 注册基础事件处理函数
         self._register_base_events()
         
@@ -62,6 +64,15 @@ class ASEngineClient:
         @self.sio.on('response', namespace=self.namespace)
         def on_response(data):
             print(f"Received response: {data}")
+        
+        # ACK事件处理
+        @self.sio.on('ack', namespace=self.namespace)
+        def on_ack(data):
+            ack_id = data.get('ack_id')
+            if ack_id and ack_id in self.ack_events:
+                print(f"Received ACK for message: {ack_id}")
+                event = self.ack_events[ack_id]
+                event.set()  # 触发事件，通知等待的线程
             
     def register_on_connect_entry(self, callback: Callable, entry: Any, init_data: Any)->bool:
         if not callable(callback):
@@ -169,12 +180,15 @@ class ASEngineClient:
             self.sid = None
             print("Disconnected from server")
         
-    def send_text(self, route: str, text: str, id:str="") -> bool:
+    def send_text(self, route: str, text: str, id:str="", wait_for_ack: bool = False, timeout: int = 5) -> bool:
         """
         发送文本消息
         :param route: 目标路由
         :param text: 要发送的文本
-        :return: 是否发送成功
+        :param id: 发送者ID
+        :param wait_for_ack: 是否等待服务器ACK确认
+        :param timeout: 等待ACK的超时时间（秒）
+        :return: 是否发送成功（若wait_for_ack为True，则返回是否收到ACK）
         """
         if not self.connected:
             print("Not connected to server")
@@ -183,19 +197,53 @@ class ASEngineClient:
             id = self.id
 
         try:
-            self.sio.emit(route, {'type': 'text', 'data': text, 'id': id}, namespace=self.namespace)
+            # 生成唯一ack_id
+            ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
+            
+            # 准备发送数据
+            data = {
+                'type': 'text', 
+                'data': text, 
+                'id': id,
+                'ack_id': ack_id
+            }
+            
+            # 发送数据
+            self.sio.emit(route, data, namespace=self.namespace)
             print(f"Sent text to route '{route}': {text}")
-            return True
+            
+            # 如果不需要等待ACK，直接返回True
+            if not wait_for_ack:
+                return True
+            
+            # 初始化ACK事件
+            event = Event()
+            self.ack_events[ack_id] = event
+            
+            # 等待ACK或超时
+            received_ack = event.wait(timeout=timeout)
+            
+            # 清理ACK事件
+            del self.ack_events[ack_id]
+            
+            if received_ack:
+                return True
+            else:
+                print(f"ACK timeout for message: {ack_id}")
+                return False
         except Exception as e:
             print(f"Failed to send text to route '{route}': {e}")
             return False
         
-    def send_voice(self, route: str, voice_data: bytes, id:str=None) -> bool:
+    def send_voice(self, route: str, voice_data: bytes, id:str=None, wait_for_ack: bool = False, timeout: int = 5) -> bool:
         """
         发送语音数据
         :param route: 目标路由
         :param voice_data: 语音数据字节流
-        :return: 是否发送成功
+        :param id: 发送者ID
+        :param wait_for_ack: 是否等待服务器ACK确认
+        :param timeout: 等待ACK的超时时间（秒）
+        :return: 是否发送成功（若wait_for_ack为True，则返回是否收到ACK）
         """
         if id is None:
             id = self.id
@@ -204,8 +252,9 @@ class ASEngineClient:
             return False
         
         try:
-            # 编码语音数据为base64
-            # voice_data = 'data:image/jpeg;base64,' +base64.b64encode(voice_data).decode('utf-8')
+            # 生成唯一ack_id
+            ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
+            
             # 发送数据
             self.sio.emit(
                 route,
@@ -214,24 +263,46 @@ class ASEngineClient:
                     'data': voice_data,
                     'timestamp': int(time.time() * 1000),
                     'sampleRate': 16000,
-                    'id': id
+                    'id': id,
+                    'ack_id': ack_id
                 },
                 namespace=self.namespace
             )
             
-
             print(f"Sent voice file to route '{route}'")
-            return True
+            
+            # 如果不需要等待ACK，直接返回True
+            if not wait_for_ack:
+                return True
+            
+            # 初始化ACK事件
+            event = Event()
+            self.ack_events[ack_id] = event
+            
+            # 等待ACK或超时
+            received_ack = event.wait(timeout=timeout)
+            
+            # 清理ACK事件
+            del self.ack_events[ack_id]
+            
+            if received_ack:
+                return True
+            else:
+                print(f"ACK timeout for message: {ack_id}")
+                return False
         except Exception as e:
             print(f"Failed to send voice to route '{route}': {e}")
             return False
         
-    def send_image(self, route: str, file_path: str, id:str) -> bool:
+    def send_image(self, route: str, file_path: str, id:str, wait_for_ack: bool = False, timeout: int = 5) -> bool:
         """
         发送图片文件
         :param route: 目标路由
         :param file_path: 图片文件路径
-        :return: 是否发送成功
+        :param id: 发送者ID
+        :param wait_for_ack: 是否等待服务器ACK确认
+        :param timeout: 等待ACK的超时时间（秒）
+        :return: 是否发送成功（若wait_for_ack为True，则返回是否收到ACK）
         """
         if not self.connected:
             print("Not connected to server")
@@ -251,18 +322,41 @@ class ASEngineClient:
             # 获取文件名
             filename = os.path.basename(file_path)
             
+            # 生成唯一ack_id
+            ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
+            
             # 发送数据
             self.sio.emit(
                 route,
                 {
                     'type': 'image',
                     'data': image_data,
-                    'id': id
+                    'id': id,
+                    'ack_id': ack_id
                 },
                 namespace=self.namespace
             )
             print(f"Sent image file to route '{route}': {filename}")
-            return True
+            
+            # 如果不需要等待ACK，直接返回True
+            if not wait_for_ack:
+                return True
+            
+            # 初始化ACK事件
+            event = Event()
+            self.ack_events[ack_id] = event
+            
+            # 等待ACK或超时
+            received_ack = event.wait(timeout=timeout)
+            
+            # 清理ACK事件
+            del self.ack_events[ack_id]
+            
+            if received_ack:
+                return True
+            else:
+                print(f"ACK timeout for message: {ack_id}")
+                return False
         except Exception as e:
             print(f"Failed to send image to route '{route}': {e}")
             return False

Html/apps/sockets/namespaces.py

@@ -106,7 +106,8 @@ class AgentNamespace(Namespace):
         chatmanager.init(user_uuid, user_uuid2ase_client[user_uuid], fmd, fmdp, fsp, course_id, chapter_name, lesson_name, max_iter=5, app=current_app, socketio=self)
         backboard_manager.add_backboard(user_uuid, user_id, course_id, lesson_name, root_path=f'/home/{user_id}/{course_id}/{chapter_name}/{lesson_name}')
         chatmanager.bb = backboard_manager.get_backboard(user_uuid)
-        chatmanager.next_chapter()
+        if not progress:
+            chatmanager.next_chapter()
         
         # 如果有保存的进度，需要跳过已经完成的章节
         for _ in range(need_skip_chapters):
@@ -212,25 +213,17 @@ class AgentNamespace(Namespace):
         namespace_entry, ase_client = namespace_entry
         namespace_entry.emit('open-iframe',"", room=init_data['room'], namespace='/agent')
         
-        # 将异步emit改为同步requests.post请求
         try:
-            # 构建请求URL，使用配置中的ASE_ENGINE_URL
             base_url = current_app.config['ASE_ENGINE_URL']
             clear_url = f"{base_url}/clear_user_session"
-            
-            # 准备请求参数
             payload = {
                 'namespace_url': current_app.config['ASE_ENGINE_URL_TOKEN'],  # WebSocket命名空间URL
                 'user_id': init_data['room'],  # 用户ID，对应session_info.id
                 'clear_type': 'all'  # 清理类型，默认为全部
             }
-            
-            # 添加token认证头
             headers = {
                 'Content-Type': 'application/json'
             }
-            
-            # 发送同步POST请求
             response = requests.post(clear_url, json=payload, headers=headers)
             response.raise_for_status()  # 检查请求是否成功
             print(f"Clear user session successful: {response.json()}")
@@ -278,7 +271,6 @@ class AgentNamespace(Namespace):
         else: chatmanager.scores.append(data.get('data', None))
         user.set_course_progress(chatmanager.chat_historys, chatmanager.scores, chatmanager.course_id, chatmanager.chapter_name, chatmanager.lesson_name)
         chatmanager.load_next_chapter()
-        time.sleep(0.5)
         chatmanager.ase_client.send_text("chapter-start", "")