10 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
CakeCN
6c48d79076 change clear to sync 2025-11-18 23:32:41 +08:00
CakeCN
5e6e040753 fix 2025-11-18 22:42:29 +08:00
8 changed files with 287 additions and 39 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 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)

View File

@@ -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

View File

@@ -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", "")

View File

@@ -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

View File

@@ -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):

View File

@@ -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

View File

@@ -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