8 Commits

Author SHA1 Message Date
CakeCN
f8f7a9d104 建立连接OK提示 2025-12-08 13:28:41 +08:00
CakeCN
a210f5bac3 no chapter_dict show 2025-12-08 13:25:46 +08:00
CakeCN
dc9f246841 no app log in client 2025-12-08 12:56:56 +08:00
CakeCN
74f26c72fd url fix 2025-12-08 12:17:14 +08:00
CakeCN
aae10819ca more log 2025-12-08 11:35:09 +08:00
CakeCN
42549bb32a 构建test机制(hsamooc.cn上只跑test) 2025-12-02 22:40:03 +08:00
CakeCN
93cf991779 client: 增加ACK支持 2025-12-02 22:31:38 +08:00
CakeCN
788fbb07da ack-client 2025-12-02 22:31:34 +08:00
6 changed files with 249 additions and 38 deletions

View 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确认即可。
这种设计既保证了通信的可靠性,又不会影响业务逻辑的执行效率,是一种高效且可靠的通信方式。

View File

@@ -5,6 +5,7 @@ import os
from datetime import datetime from datetime import datetime
from typing import Dict, Callable, Any, Optional from typing import Dict, Callable, Any, Optional
from functools import partial from functools import partial
from threading import Event
class ASEngineClient: class ASEngineClient:
def __init__(self, server_url: str, namespace: str, id:str): def __init__(self, server_url: str, namespace: str, id:str):
@@ -29,7 +30,8 @@ class ASEngineClient:
self._on_connect_callback_entry=None self._on_connect_callback_entry=None
self._on_connect_callback_initData=None self._on_connect_callback_initData=None
# ACK相关
self.ack_events: Dict[str, Event] = {} # 存储ack_id和对应的Event对象
# 注册基础事件处理函数 # 注册基础事件处理函数
self._register_base_events() self._register_base_events()
@@ -63,6 +65,15 @@ class ASEngineClient:
def on_response(data): def on_response(data):
print(f"Received 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: def register_on_connect_entry(self, callback: Callable, entry: Any, init_data: Any)->bool:
if not callable(callback): if not callable(callback):
return False return False
@@ -169,12 +180,15 @@ class ASEngineClient:
self.sid = None self.sid = None
print("Disconnected from server") 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 route: 目标路由
:param text: 要发送的文本 :param text: 要发送的文本
:return: 是否发送成功 :param id: 发送者ID
:param wait_for_ack: 是否等待服务器ACK确认
:param timeout: 等待ACK的超时时间
:return: 是否发送成功若wait_for_ack为True则返回是否收到ACK
""" """
if not self.connected: if not self.connected:
print("Not connected to server") print("Not connected to server")
@@ -183,19 +197,53 @@ class ASEngineClient:
id = self.id id = self.id
try: try:
self.sio.emit(route, {'type': 'text', 'data': text, 'id': id}, namespace=self.namespace) # 生成唯一ack_id
print(f"Sent text to route '{route}': {text}") ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
return True
# 准备发送数据
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: except Exception as e:
print(f"Failed to send text to route '{route}': {e}") print(f"Failed to send text to route '{route}': {e}")
return False 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 route: 目标路由
:param voice_data: 语音数据字节流 :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: if id is None:
id = self.id id = self.id
@@ -204,8 +252,9 @@ class ASEngineClient:
return False return False
try: try:
# 编码语音数据为base64 # 生成唯一ack_id
# voice_data = 'data:image/jpeg;base64,' +base64.b64encode(voice_data).decode('utf-8') ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
# 发送数据 # 发送数据
self.sio.emit( self.sio.emit(
route, route,
@@ -214,24 +263,46 @@ class ASEngineClient:
'data': voice_data, 'data': voice_data,
'timestamp': int(time.time() * 1000), 'timestamp': int(time.time() * 1000),
'sampleRate': 16000, 'sampleRate': 16000,
'id': id 'id': id,
'ack_id': ack_id
}, },
namespace=self.namespace namespace=self.namespace
) )
print(f"Sent voice file to route '{route}'") 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: except Exception as e:
print(f"Failed to send voice to route '{route}': {e}") print(f"Failed to send voice to route '{route}': {e}")
return False 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 route: 目标路由
:param file_path: 图片文件路径 :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: if not self.connected:
print("Not connected to server") print("Not connected to server")
@@ -251,18 +322,41 @@ class ASEngineClient:
# 获取文件名 # 获取文件名
filename = os.path.basename(file_path) 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( self.sio.emit(
route, route,
{ {
'type': 'image', 'type': 'image',
'data': image_data, 'data': image_data,
'id': id 'id': id,
'ack_id': ack_id
}, },
namespace=self.namespace namespace=self.namespace
) )
print(f"Sent image file to route '{route}': {filename}") 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: except Exception as e:
print(f"Failed to send image to route '{route}': {e}") print(f"Failed to send image to route '{route}': {e}")
return False return False
@@ -286,6 +380,6 @@ class HSAEngineClient(ASEngineClient):
super().__init__(server_url, namespace, id) super().__init__(server_url, namespace, id)
self.chatmanager = chatmanager self.chatmanager = chatmanager
def loadmarkdown(self, fmd, fmdp, fsp): def loadmarkdown(self, fmd, fmdp, fsp):
self.send_text('markdown-in', fmd, self.id) self.send_text('markdown-in', fmd, self.id, wait_for_ack=True)
self.send_text('markdown-prompt-in', fmdp, self.id) self.send_text('markdown-prompt-in', fmdp, self.id, wait_for_ack=True)
self.send_text('score-prompt-in', fsp, self.id) self.send_text('score-prompt-in', fsp, self.id, wait_for_ack=True)

View File

@@ -27,7 +27,7 @@ def get_text_file(file_name):
response = requests.get(file_name) response = requests.get(file_name)
if response.status_code == 200: if response.status_code == 200:
print(file_name) # print(file_name)
# print(f"get_text_file: {response.text}") # print(f"get_text_file: {response.text}")
return response.text return response.text

View File

@@ -13,6 +13,7 @@ from ..services.markdown_service import load_full_markdown_file
from ..services.course_service import save_learning_progress from ..services.course_service import save_learning_progress
from ..extension_ase.ase_client import HSAEngineClient from ..extension_ase.ase_client import HSAEngineClient
from ..services.user_service import get_or_load_current_user from ..services.user_service import get_or_load_current_user
from ..services.course_service import load_learning_progress
@@ -76,7 +77,6 @@ class AgentNamespace(Namespace):
chatmanager.function_manager.add_function(next_chapter, {'user_uuid': user_uuid}) 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) progress = load_learning_progress(user_uuid, course_id, chapter_name, lesson_name)
need_skip_chapters = 0 need_skip_chapters = 0
@@ -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) 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}') 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.bb = backboard_manager.get_backboard(user_uuid)
chatmanager.next_chapter() if not progress:
chatmanager.next_chapter()
# 如果有保存的进度,需要跳过已经完成的章节 # 如果有保存的进度,需要跳过已经完成的章节
for _ in range(need_skip_chapters): for _ in range(need_skip_chapters):
@@ -211,26 +212,19 @@ class AgentNamespace(Namespace):
print(f"on_connect_to_ase {client_entry}, {namespace_entry}, {init_data}") print(f"on_connect_to_ase {client_entry}, {namespace_entry}, {init_data}")
namespace_entry, ase_client = namespace_entry namespace_entry, ase_client = namespace_entry
namespace_entry.emit('open-iframe',"", room=init_data['room'], namespace='/agent') namespace_entry.emit('open-iframe',"", room=init_data['room'], namespace='/agent')
namespace_entry.emit('message', "Agents服务已连接, 加载教案中", room=init_data['room'], namespace='/agent')
# 将异步emit改为同步requests.post请求
try: try:
# 构建请求URL使用配置中的ASE_ENGINE_URL
base_url = current_app.config['ASE_ENGINE_URL'] base_url = current_app.config['ASE_ENGINE_URL']
clear_url = f"{base_url}/clear_user_session" clear_url = f"{base_url}/clear_user_session"
# 准备请求参数
payload = { payload = {
'namespace_url': current_app.config['ASE_ENGINE_URL_TOKEN'], # WebSocket命名空间URL 'namespace_url': current_app.config['ASE_ENGINE_URL_TOKEN'], # WebSocket命名空间URL
'user_id': init_data['room'], # 用户ID对应session_info.id 'user_id': init_data['room'], # 用户ID对应session_info.id
'clear_type': 'all' # 清理类型,默认为全部 'clear_type': 'all' # 清理类型,默认为全部
} }
# 添加token认证头
headers = { headers = {
'Content-Type': 'application/json' 'Content-Type': 'application/json'
} }
# 发送同步POST请求
response = requests.post(clear_url, json=payload, headers=headers) response = requests.post(clear_url, json=payload, headers=headers)
response.raise_for_status() # 检查请求是否成功 response.raise_for_status() # 检查请求是否成功
print(f"Clear user session successful: {response.json()}") print(f"Clear user session successful: {response.json()}")
@@ -238,7 +232,8 @@ class AgentNamespace(Namespace):
print(f"Error clearing user session: {e}") print(f"Error clearing user session: {e}")
ase_client.chatmanager.load_next_chapter() ase_client.chatmanager.load_next_chapter()
ase_client.send_text("chapter-start", "") ase_client.send_text("chapter-start", "")
namespace_entry.emit('message', "Multi-Agents服务已连接", room=init_data['room'], namespace='/agent') namespace_entry.emit('message', "教案加载完毕", room=init_data['room'], namespace='/agent')
#接受消息回来直接让message监听发送 #接受消息回来直接让message监听发送
def receive_ase_paste_detected(client_entry: HSAEngineClient, namespace_entry: Namespace, data, init_data): def receive_ase_paste_detected(client_entry: HSAEngineClient, namespace_entry: Namespace, data, init_data):
@@ -278,7 +273,6 @@ class AgentNamespace(Namespace):
else: chatmanager.scores.append(data.get('data', None)) 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) user.set_course_progress(chatmanager.chat_historys, chatmanager.scores, chatmanager.course_id, chatmanager.chapter_name, chatmanager.lesson_name)
chatmanager.load_next_chapter() chatmanager.load_next_chapter()
time.sleep(0.5)
chatmanager.ase_client.send_text("chapter-start", "") chatmanager.ase_client.send_text("chapter-start", "")

View File

@@ -156,7 +156,6 @@ def load_chapters(markdown, markdown_prompt, score_prompt):
chapter_chain = [] chapter_chain = []
No = 1 No = 1
print (chapter_dict)
for chapter_name in chapter_sequence: 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"])) 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 No+=1

View File

@@ -5,9 +5,9 @@ api_key = sk-aMpnWklN2IbsK44d1kNpy6YOP9bk1pdPJjFeEmbb0a5ytEFf
model=gpt-4.1-nano model=gpt-4.1-nano
[VSCODE_WEB] [VSCODE_WEB]
url = https://hsamooc.com url = https://hsamooc.cn
[CODE_LIKE] [CODE_LIKE]
url = https://hsamooc.com/vsc-like url = /vsc-like
#http://asengine.net:8282 #http://asengine.net:8282
[VSCODE_WEB_PATH] [VSCODE_WEB_PATH]
is_wsl = true is_wsl = true
@@ -27,6 +27,6 @@ bucket = hsamooc-cdn-1374354408
region = ap-guangzhou region = ap-guangzhou
[ASE_ENGINE] [ASE_ENGINE]
url = https://asengine.net url = https://test.asengine.net
namespace = /socket.io/7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1 namespace = /socket.io/7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1
url_token = 7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1 url_token = 7100eebf7a909e1ee3f4e06766f512f0358ecb6ac51d5434c17d31dd70890dc1