Merge branch 'main' into test
This commit is contained in:
CakeCN
326
Html/apps/_docs/ase_engine_data_sync.md
Normal file
326
Html/apps/_docs/ase_engine_data_sync.md
Normal file
@@ -0,0 +1,326 @@
|
||||
# ASEngine 数据同步机制文档
|
||||
|
||||
## 1. 概述
|
||||
|
||||
本文档介绍了如何通过 `learning_progess_process.py` 实现与 ASEngine（大模型框架后端）之间的运行时数据下载和上传。该机制为客户端与大模型服务之间提供了可靠的数据同步方案，适用于需要在不同服务之间共享和同步用户数据的场景。
|
||||
|
||||
## 2. 使用场景
|
||||
|
||||
### 2.1 核心场景
|
||||
|
||||
- **用户学习进度同步**：将用户在客户端的学习进度同步到大模型框架后端，确保跨设备访问一致性
|
||||
- **对话历史管理**：从大模型框架获取用户对话历史，用于客户端展示和分析
|
||||
- **多服务数据共享**：在客户端应用与大模型服务之间建立数据桥接，实现数据互通
|
||||
- **分布式系统协作**：支持多个服务实例之间的数据同步，确保系统一致性
|
||||
|
||||
### 2.2 扩展场景
|
||||
|
||||
- **离线数据同步**：支持在网络恢复后进行数据同步
|
||||
- **数据备份与恢复**：通过上传/下载机制实现数据备份和恢复。
|
||||
- **多租户数据隔离**：通过 namespace_token 实现不同租户数据隔离
|
||||
|
||||
## 3. 实现原理
|
||||
|
||||
### 3.1 系统架构
|
||||
|
||||
```
|
||||
┌─────────────────────────┐ ┌─────────────────────────┐
|
||||
│ 客户端应用 │ │ ASEngine │
|
||||
│ (learning_progess_process.py) │ (大模型框架后端) │
|
||||
├─────────────────────────┤ ├─────────────────────────┤
|
||||
│ 1. get_multiagents_dialog_list │ ←─── 下载数据 ──── │ /api/sync/download │
|
||||
│ 2. upload_learning_progress_to_cloud │ ──── 上传数据 ────→ │ /api/sync/upload │
|
||||
└─────────────────────────┘ └─────────────────────────┘
|
||||
```
|
||||
|
||||
### 3.2 数据流向
|
||||
|
||||
1. **数据下载流程**：
|
||||
|
||||
- 客户端调用 `get_multiagents_dialog_list()` 方法
|
||||
- 发送 POST 请求到 ASEngine 的 `/api/sync/download` 接口
|
||||
- 获取用户对话历史数据
|
||||
- 数据用于本地处理或存储
|
||||
2. **数据上传流程**：
|
||||
|
||||
- 客户端调用 `upload_learning_progress_to_cloud()` 方法
|
||||
- 序列化待上传的数据（处理特殊类型如 datetime）
|
||||
- 发送 POST 请求到 ASEngine 的 `/api/sync/upload` 接口
|
||||
- 接收上传结果并记录日志
|
||||
|
||||
## 4. 核心功能说明
|
||||
|
||||
### 4.1 数据下载功能
|
||||
|
||||
```python
|
||||
def get_multiagents_dialog_list(user_id: str) -> List[Dict]:
|
||||
"""
|
||||
从 MultiAgents 框架获取用户对话列表
|
||||
|
||||
参数：
|
||||
- user_id: 用户ID
|
||||
|
||||
返回：
|
||||
- Dict[List]: 对话列表(key 为 History database)
|
||||
"""
|
||||
# 实现细节...
|
||||
```
|
||||
|
||||
**功能特点**：
|
||||
|
||||
- 从配置中动态获取 ASEngine URL 和认证信息
|
||||
- 支持超时控制（10秒）
|
||||
- 完善的错误处理和日志记录
|
||||
- 返回结构化数据，便于后续处理
|
||||
|
||||
### 4.2 数据上传功能
|
||||
|
||||
```python
|
||||
def upload_learning_progress_to_cloud(progress: Dict) -> Dict:
|
||||
"""
|
||||
将学习进度上传到云服务
|
||||
|
||||
参数：
|
||||
- progress: 学习进度数据字典
|
||||
|
||||
返回：
|
||||
- Dict: 上传结果，包含success和message字段
|
||||
"""
|
||||
# 实现细节...
|
||||
```
|
||||
|
||||
**功能特点**：
|
||||
|
||||
- 自动处理数据序列化，支持复杂数据类型
|
||||
- 完善的错误处理机制
|
||||
- 详细的日志记录，便于问题排查
|
||||
- 返回标准化结果格式
|
||||
|
||||
### 4.3 辅助功能
|
||||
|
||||
#### 4.3.1 学习进度保存
|
||||
|
||||
```python
|
||||
def save_learning_progress(chatmanager, user_id: str) -> bool:
|
||||
"""
|
||||
保存用户学习进度到MongoDB
|
||||
"""
|
||||
# 实现细节...
|
||||
```
|
||||
|
||||
#### 4.3.2 学习进度加载
|
||||
|
||||
```python
|
||||
def load_learning_progress(user_id: str, material_id: str, chapter_name: str, lesson_name: str) -> Dict:
|
||||
"""
|
||||
加载用户学习进度
|
||||
"""
|
||||
# 实现细节...
|
||||
```
|
||||
|
||||
## 5. 配置方法
|
||||
|
||||
### 5.1 环境配置
|
||||
|
||||
在 Flask 应用的配置中添加以下配置项：
|
||||
|
||||
```python
|
||||
# ASEngine 配置
|
||||
ASE_ENGINE_URL = "https://your-ase-engine-url.com" # ASEngine 服务地址
|
||||
ASE_ENGINE_URL_TOKEN = "your-namespace-token" # 命名空间认证令牌
|
||||
```
|
||||
|
||||
### 5.2 依赖安装
|
||||
|
||||
确保项目中已安装以下依赖：
|
||||
|
||||
```bash
|
||||
pip install requests # HTTP 请求库
|
||||
pip install pymongo # MongoDB 驱动（如果使用MongoDB存储）
|
||||
```
|
||||
|
||||
## 6. 使用示例
|
||||
|
||||
### 6.1 数据下载示例
|
||||
|
||||
```python
|
||||
from apps.services.course_services.learning_progess_process import get_multiagents_dialog_list
|
||||
|
||||
# 获取用户对话历史
|
||||
dialog_list = get_multiagents_dialog_list(user_id="user123")
|
||||
|
||||
# 处理对话数据
|
||||
his_data = dialog_list.get('data', [])
|
||||
for dialog in his_data:
|
||||
# 处理单个对话
|
||||
print(f"对话ID: {dialog.get('id')}, 内容: {dialog.get('content')}")
|
||||
```
|
||||
|
||||
### 6.2 数据上传示例
|
||||
|
||||
```python
|
||||
from apps.services.course_services.learning_progess_process import upload_learning_progress_to_cloud
|
||||
|
||||
# 准备待上传的数据
|
||||
progress_data = {
|
||||
"user_id": "user123",
|
||||
"multiagents_dialogs": {
|
||||
"databases": {...}, # 对话数据库
|
||||
"global_databases": {...}, # 全局数据库
|
||||
"session_info_out": {...} # 会话信息
|
||||
}
|
||||
}
|
||||
|
||||
# 上传数据到 ASEngine
|
||||
result = upload_learning_progress_to_cloud(progress_data)
|
||||
|
||||
# 处理上传结果
|
||||
if result["success"]:
|
||||
print("数据上传成功")
|
||||
else:
|
||||
print(f"数据上传失败: {result['message']}")
|
||||
```
|
||||
|
||||
### 6.3 完整流程示例
|
||||
|
||||
```python
|
||||
from apps.services.course_services.learning_progess_process import (
|
||||
get_multiagents_dialog_list,
|
||||
save_learning_progress,
|
||||
load_learning_progress,
|
||||
upload_learning_progress_to_cloud
|
||||
)
|
||||
|
||||
# 1. 从 ASEngine 获取对话历史
|
||||
dialog_list = get_multiagents_dialog_list(user_id="user123")
|
||||
|
||||
# 2. 保存到本地 MongoDB
|
||||
save_learning_progress(chatmanager, user_id="user123")
|
||||
|
||||
# 3. 从本地加载学习进度
|
||||
progress = load_learning_progress(
|
||||
user_id="user123",
|
||||
material_id="material456",
|
||||
chapter_name="第1章",
|
||||
lesson_name="第1节"
|
||||
)
|
||||
|
||||
# 4. 将学习进度上传回 ASEngine
|
||||
if progress["exists"]:
|
||||
upload_result = upload_learning_progress_to_cloud(progress["data"])
|
||||
print(f"上传结果: {upload_result}")
|
||||
```
|
||||
|
||||
## 7. 代码结构与扩展
|
||||
|
||||
### 7.1 核心模块
|
||||
|
||||
| 模块 | 功能描述 | 文件位置 |
|
||||
| --------------- | -------------------------- | ------------------------------------------------------------- |
|
||||
| 数据同步服务 | 处理与 ASEngine 的数据交互 | `apps/services/course_services/learning_progess_process.py` |
|
||||
| JSON 序列化工具 | 处理复杂数据类型序列化 | `apps/utils/json_serializer.py` |
|
||||
| 配置管理 | 管理 ASEngine 连接配置 | `apps/config.py` |
|
||||
|
||||
### 7.2 扩展建议
|
||||
|
||||
1. **添加重试机制**：
|
||||
|
||||
```python
|
||||
def upload_with_retry(progress: Dict, max_retries: int = 3) -> Dict:
|
||||
for i in range(max_retries):
|
||||
result = upload_learning_progress_to_cloud(progress)
|
||||
if result["success"]:
|
||||
return result
|
||||
time.sleep(1) # 指数退避算法可进一步优化
|
||||
return {"success": False, "message": f"上传失败，已重试{max_retries}次"}
|
||||
```
|
||||
2. **添加数据校验**：
|
||||
|
||||
```python
|
||||
def validate_progress_data(progress: Dict) -> bool:
|
||||
"""验证进度数据格式"""
|
||||
required_fields = ["user_id", "multiagents_dialogs"]
|
||||
for field in required_fields:
|
||||
if field not in progress:
|
||||
return False
|
||||
return True
|
||||
```
|
||||
3. **支持异步操作**：
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import aiohttp
|
||||
|
||||
async def async_upload_learning_progress(progress: Dict) -> Dict:
|
||||
"""异步上传学习进度"""
|
||||
# 异步实现
|
||||
pass
|
||||
```
|
||||
|
||||
## 8. 最佳实践
|
||||
|
||||
### 8.1 错误处理
|
||||
|
||||
- 始终检查 API 返回结果的 `success` 字段
|
||||
- 实现适当的日志记录，便于问题排查
|
||||
- 对网络异常和超时进行合理处理
|
||||
|
||||
### 8.2 性能优化
|
||||
|
||||
- 避免频繁的小数据上传，考虑批量处理
|
||||
- 对大文件进行压缩后上传
|
||||
- 合理设置超时时间，避免长时间阻塞
|
||||
|
||||
### 8.3 安全性
|
||||
|
||||
- 确保 ASEngine URL 和令牌安全存储，避免硬编码
|
||||
- 使用 HTTPS 协议进行数据传输
|
||||
- 对敏感数据进行加密处理
|
||||
|
||||
### 8.4 监控与告警
|
||||
|
||||
- 添加数据同步成功率监控
|
||||
- 对同步失败情况进行告警
|
||||
- 记录详细的操作日志，便于审计
|
||||
|
||||
## 9. 后续项目参考建议
|
||||
|
||||
### 9.1 架构设计
|
||||
|
||||
- **分层设计**：将数据同步逻辑与业务逻辑分离，提高代码可维护性
|
||||
- **接口标准化**：定义统一的数据格式和接口规范，便于扩展
|
||||
- **依赖注入**：使用依赖注入模式，提高代码可测试性
|
||||
|
||||
### 9.2 代码实现
|
||||
|
||||
- **模块化**：将不同功能模块拆分为独立函数或类
|
||||
- **类型注解**：使用 Type Hints 提高代码可读性和类型安全性
|
||||
- **文档完善**：为每个函数添加详细的文档字符串
|
||||
- **测试覆盖**：编写单元测试和集成测试，确保功能正确性
|
||||
|
||||
### 9.3 扩展功能
|
||||
|
||||
- **支持多种数据格式**：如 JSON、Protobuf 等
|
||||
- **添加数据版本控制**：支持不同版本数据的兼容处理
|
||||
- **实现增量同步**：只同步变化的数据，提高同步效率
|
||||
- **支持断点续传**：对于大文件上传实现断点续传功能
|
||||
|
||||
## 10. 总结
|
||||
|
||||
本文档介绍了基于 `learning_progess_process.py` 实现的 ASEngine 数据同步机制，包括使用场景、实现原理、核心功能、配置方法和使用示例。该机制为客户端与大模型服务之间提供了可靠的数据同步方案，可作为后续项目的参考模板。
|
||||
|
||||
通过遵循本文档中的最佳实践和扩展建议，可以构建更加健壮、高效和可扩展的数据同步系统，满足不同场景下的数据同步需求。
|
||||
|
||||
## 11. 版本历史
|
||||
|
||||
| 版本 | 日期 | 描述 | 作者 |
|
||||
| ---- | ---------- | ------------ | ------------ |
|
||||
| v1.0 | 2025-12-10 | 初始版本 | 系统自动生成 |
|
||||
| v1.1 | YYYY-MM-DD | 添加异步支持 | XXX |
|
||||
| v1.2 | YYYY-MM-DD | 实现增量同步 | XXX |
|
||||
|
||||
## 12. 参考资料
|
||||
|
||||
- [Flask 官方文档](https://flask.palletsprojects.com/)
|
||||
- [Requests 库文档](https://requests.readthedocs.io/)
|
||||
- [ASEngine API 文档](https://your-ase-engine-docs.com/)
|
||||
@@ -86,7 +86,6 @@ class AgentNamespace(Namespace):
|
||||
chatmanager.function_manager.add_function(next_chapter, {'user_uuid': user_uuid})
|
||||
|
||||
if not chatmanager.restart:# 尝试加载用户学习进度
|
||||
self.app.logger.info(f"尝试加载用户学习进度: {user_id}, {course_id}, {chapter_name}, {lesson_name}")
|
||||
progress_result = load_learning_progress(user_id, course_id, chapter_name, lesson_name)
|
||||
need_skip_chapters = 0
|
||||
|
||||
@@ -97,7 +96,6 @@ class AgentNamespace(Namespace):
|
||||
# 根据scores的数量确定需要跳过的章节数
|
||||
need_skip_chapters = len(progress['scores'])
|
||||
chatmanager.scores = progress['scores']
|
||||
self.app.logger.info(f"恢复用户学习进度: {user_id}, 已完成 {need_skip_chapters} 个章节")
|
||||
|
||||
# 恢复聊天历史
|
||||
if 'chat_historys' in progress:
|
||||
@@ -113,7 +111,6 @@ class AgentNamespace(Namespace):
|
||||
# 使用默认值，确保有完整的结构
|
||||
chatmanager.chat_historys = progress_result['data']['chat_historys']
|
||||
chatmanager.scores = progress_result['data']['scores']
|
||||
self.app.logger.info(f"未找到学习记录，使用默认值: {progress_result['message']}")
|
||||
|
||||
fmd,fmdp,fsp = load_full_markdown_file(course_id, chapter_name, lesson_name)
|
||||
user_uuid2ase_client[user_uuid] = HSAEngineClient(current_app.config["ASE_ENGINE_URL"], current_app.config["ASE_ENGINE_NAMESPACE"],id=user_id, chatmanager = chatmanager)
|
||||
@@ -186,10 +183,8 @@ class AgentNamespace(Namespace):
|
||||
|
||||
def on_language(self, language):
|
||||
ex = current_app.extensions
|
||||
self.app.logger.info(f"Language changed to: {language}")
|
||||
|
||||
def on_message(self, data):
|
||||
self.app.logger.info(f"Message from client: {data}")
|
||||
ex = current_app.extensions
|
||||
chatmanager = ex["user_uuid2chatmanager"]
|
||||
uuid = session.get('user_uuid')
|
||||
@@ -201,11 +196,9 @@ class AgentNamespace(Namespace):
|
||||
chatmanager[uuid].chat_historys.append({'role': 'user', 'content': data['data']})
|
||||
|
||||
if data['type'] == 'function': # 在这里 前端会发送允许执行的function与参数
|
||||
self.app.logger.info(f"function_call: {data['data']}")
|
||||
d = data['data']
|
||||
res = chatmanager[uuid].function_call(d['data']['name'], d['data'].get('args', {}))
|
||||
d['data'] = res.to_dict()
|
||||
self.app.logger.info(f"function_call_res: {d['data']}")
|
||||
emit(d['route'], d, room=uuid, namespace='/agent')
|
||||
chatmanager[uuid].chat_historys.append({'role': 'user', 'content': data['data']})
|
||||
user_uuid2ase_client[uuid].send_text(d['route'], data['data'])
|
||||
@@ -216,10 +209,9 @@ class AgentNamespace(Namespace):
|
||||
uuid = session.get('user_uuid')
|
||||
user_uuid2ase_client = ex["user_uuid2ase_client"]
|
||||
res = user_uuid2ase_client[uuid].send_voice('voice_in', voice_data['data'])
|
||||
self.app.logger.info(f"Send voicce to route 'voice_in' success: {res}")
|
||||
|
||||
|
||||
def on_initiative(self,data): # 主动call
|
||||
self.app.logger.info("User active function call")
|
||||
ex = current_app.extensions
|
||||
uuid = session.get('user_uuid')
|
||||
manager = ex["user_uuid2chatmanager"][uuid]
|
||||
@@ -247,7 +239,6 @@ class AgentNamespace(Namespace):
|
||||
chatmanager = user_uuid2chatmanager.get(uuid)
|
||||
if chatmanager:
|
||||
save_result = save_learning_progress(chatmanager, user_id)
|
||||
self.app.logger.info(f"保存学习进度结果: {save_result}")
|
||||
|
||||
except Exception as e:
|
||||
self.app.logger.error(f"保存学习进度时发生异常: {str(e)}")
|
||||
|
||||
Reference in New Issue
Block a user