Compare commits
10 Commits
save_load_
...
ack-client
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f8f7a9d104 | ||
|
|
a210f5bac3 | ||
|
|
dc9f246841 | ||
|
|
74f26c72fd | ||
|
|
aae10819ca | ||
|
|
42549bb32a | ||
|
|
93cf991779 | ||
|
|
788fbb07da | ||
|
|
6c48d79076 | ||
|
|
5e6e040753 |
124
Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md
Normal file
124
Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md
Normal file
@@ -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确认即可。
|
||||
|
||||
这种设计既保证了通信的可靠性,又不会影响业务逻辑的执行效率,是一种高效且可靠的通信方式。
|
||||
@@ -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)
|
||||
print(f"Sent text to route '{route}': {text}")
|
||||
return True
|
||||
# 生成唯一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[:20]}")
|
||||
|
||||
# 如果不需要等待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
|
||||
@@ -286,6 +380,6 @@ class HSAEngineClient(ASEngineClient):
|
||||
super().__init__(server_url, namespace, id)
|
||||
self.chatmanager = chatmanager
|
||||
def loadmarkdown(self, fmd, fmdp, fsp):
|
||||
self.send_text('markdown-in', fmd, self.id)
|
||||
self.send_text('markdown-prompt-in', fmdp, self.id)
|
||||
self.send_text('score-prompt-in', fsp, self.id)
|
||||
self.send_text('markdown-in', fmd, self.id, wait_for_ack=True)
|
||||
self.send_text('markdown-prompt-in', fmdp, self.id, wait_for_ack=True)
|
||||
self.send_text('score-prompt-in', fsp, self.id, wait_for_ack=True)
|
||||
|
||||
@@ -27,7 +27,7 @@ def get_text_file(file_name):
|
||||
response = requests.get(file_name)
|
||||
|
||||
if response.status_code == 200:
|
||||
print(file_name)
|
||||
# print(file_name)
|
||||
# print(f"get_text_file: {response.text}")
|
||||
|
||||
return response.text
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
# myapp/sockets/namespaces.py
|
||||
import json
|
||||
import time
|
||||
from flask import current_app, request, session
|
||||
import requests
|
||||
from flask import current_app, request, session, url_for
|
||||
from flask_socketio import Namespace, join_room, leave_room, emit
|
||||
from ..config import Config
|
||||
|
||||
from ..function_tools import judge
|
||||
from ..function_tools import next_chapter
|
||||
@@ -11,6 +13,7 @@ from ..services.markdown_service import load_full_markdown_file
|
||||
from ..services.course_service import save_learning_progress
|
||||
from ..extension_ase.ase_client import HSAEngineClient
|
||||
from ..services.user_service import get_or_load_current_user
|
||||
from ..services.course_service import load_learning_progress
|
||||
|
||||
|
||||
|
||||
@@ -74,7 +77,6 @@ class AgentNamespace(Namespace):
|
||||
chatmanager.function_manager.add_function(next_chapter, {'user_uuid': user_uuid})
|
||||
|
||||
# 尝试加载用户学习进度
|
||||
from ..services.course_service import load_learning_progress
|
||||
progress = load_learning_progress(user_uuid, course_id, chapter_name, lesson_name)
|
||||
need_skip_chapters = 0
|
||||
|
||||
@@ -104,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):
|
||||
@@ -209,10 +212,28 @@ class AgentNamespace(Namespace):
|
||||
print(f"on_connect_to_ase {client_entry}, {namespace_entry}, {init_data}")
|
||||
namespace_entry, ase_client = namespace_entry
|
||||
namespace_entry.emit('open-iframe',"", room=init_data['room'], namespace='/agent')
|
||||
namespace_entry.emit('message', "Agents服务已连接, 加载教案中", room=init_data['room'], namespace='/agent')
|
||||
|
||||
try:
|
||||
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' # 清理类型,默认为全部
|
||||
}
|
||||
headers = {
|
||||
'Content-Type': 'application/json'
|
||||
}
|
||||
response = requests.post(clear_url, json=payload, headers=headers)
|
||||
response.raise_for_status() # 检查请求是否成功
|
||||
print(f"Clear user session successful: {response.json()}")
|
||||
except Exception as e:
|
||||
print(f"Error clearing user session: {e}")
|
||||
ase_client.chatmanager.load_next_chapter()
|
||||
ase_client.send_text("chapter-start", "")
|
||||
ase_client.send_text("clear","")
|
||||
namespace_entry.emit('message', "Multi-Agents服务已连接", room=init_data['room'], namespace='/agent')
|
||||
namespace_entry.emit('message', "教案加载完毕", room=init_data['room'], namespace='/agent')
|
||||
|
||||
|
||||
#接受消息回来直接让message监听发送
|
||||
def receive_ase_paste_detected(client_entry: HSAEngineClient, namespace_entry: Namespace, data, init_data):
|
||||
@@ -252,7 +273,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", "")
|
||||
|
||||
|
||||
|
||||
@@ -156,7 +156,6 @@ def load_chapters(markdown, markdown_prompt, score_prompt):
|
||||
|
||||
chapter_chain = []
|
||||
No = 1
|
||||
print (chapter_dict)
|
||||
for chapter_name in chapter_sequence:
|
||||
chapter_chain.append(Chapter( No, CHAPTER_LATTER, chapter_name, chapter_dict[chapter_name]["markdown"], chapter_dict[chapter_name]["markdown_prompt"], chapter_dict[chapter_name]["score_prompt"],chapter_dict[chapter_name]["require_tools"]))
|
||||
No+=1
|
||||
|
||||
@@ -43,13 +43,22 @@ class Backboard:
|
||||
f"- File tree: {self.file_tree}{endl}")
|
||||
# f"- Activated file content (current editing 10 lines): {endl}"
|
||||
def add_history(self, realtime_action):
|
||||
if(realtime_action['type']=='config'):return
|
||||
if len(self.history)!=0:
|
||||
if (self.history[-1]['type']=='fileEdit'):
|
||||
if(self.history[-1]['filePath'] == realtime_action['filePath']):
|
||||
# 检查必要的键是否存在,避免KeyError
|
||||
if realtime_action.get('type') == 'config':
|
||||
return
|
||||
if len(self.history) > 0 and realtime_action.get('type') == 'fileEdit':
|
||||
# 安全地获取filePath,并进行比较
|
||||
current_file_path = realtime_action.get('filePath')
|
||||
# 检查最近的历史记录是否包含filePath
|
||||
if current_file_path and self.history[-1].get('filePath') == current_file_path:
|
||||
# 安全地更新content,只有当content存在时才更新
|
||||
if 'content' in realtime_action:
|
||||
self.history[-1]['content'] = realtime_action['content']
|
||||
return
|
||||
return
|
||||
self.history.append(realtime_action)
|
||||
# 限制历史记录的数量
|
||||
if len(self.history) > 500: # 假设最多保留500条记录
|
||||
self.history.pop(0) # 移除最旧的记录
|
||||
|
||||
|
||||
def get_deltatime_mmss(self, start_time, end_time):
|
||||
|
||||
@@ -5,9 +5,9 @@ api_key = sk-aMpnWklN2IbsK44d1kNpy6YOP9bk1pdPJjFeEmbb0a5ytEFf
|
||||
model=gpt-4.1-nano
|
||||
|
||||
[VSCODE_WEB]
|
||||
url = https://hsamooc.com
|
||||
url = https://hsamooc.cn
|
||||
[CODE_LIKE]
|
||||
url = https://hsamooc.com/vsc-like
|
||||
url = /vsc-like
|
||||
#http://asengine.net:8282
|
||||
[VSCODE_WEB_PATH]
|
||||
is_wsl = true
|
||||
@@ -27,6 +27,6 @@ bucket = hsamooc-cdn-1374354408
|
||||
region = ap-guangzhou
|
||||
|
||||
[ASE_ENGINE]
|
||||
url = https://asengine.net
|
||||
url = https://test.asengine.net
|
||||
namespace = /socket.io/7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1
|
||||
url_token = 7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1
|
||||
|
||||
@@ -148,12 +148,14 @@
|
||||
});
|
||||
|
||||
// 防抖处理非粘贴的编辑操作
|
||||
const content = changes.text;
|
||||
// 确保始终使用完整的编辑器内容而非仅变更部分
|
||||
const fullContent = editor.getValue();
|
||||
debounceSendToServer({
|
||||
type: 'fileEdit',
|
||||
filePath: filePath,
|
||||
content: content,
|
||||
content: fullContent,
|
||||
config: code_like_config,
|
||||
timestamp: new Date().getTime() // 添加时间戳以便追踪
|
||||
}, 5000
|
||||
);
|
||||
// saveFile
|
||||
|
||||
Reference in New Issue
Block a user