diff --git a/demos/computer_use/README_en.md b/demos/computer_use/README_en.md
index d5b8509..2d00f75 100644
--- a/demos/computer_use/README_en.md
+++ b/demos/computer_use/README_en.md
@@ -20,7 +20,10 @@ Specific reference: [README_zh.md](./base_version/computer_use_server/README_en.
## Chapter 4: Advanced Usage
-In the code directory under the advanced_version package:
-- Frontend: static directory
-- Backend: [backend.py](./advanced_version/computer_use_server/backend.py)
-Specific reference: [README_en.md](./advanced_version/computer_use_server/README_en.md)
\ No newline at end of file
+In the `advanced_version` code package within the code directory:
+- **Frontend**:
+ - Basic GUI version: `computer-use-server/static` directory
+ - Advanced version: In the `adk-computer-use` code package within the code directory
+ - For details, please refer to: [README_zh.md](./advanced_version/adk-computer-use/README.md)
+- **Backend**: `backend.py`
+For details, please refer to: [README_zh.md](./advanced_version/computer_use_server/README_zh.md)
\ No newline at end of file
diff --git a/demos/computer_use/README_zh.md b/demos/computer_use/README_zh.md
index 5a08588..3254a67 100644
--- a/demos/computer_use/README_zh.md
+++ b/demos/computer_use/README_zh.md
@@ -21,6 +21,9 @@ pip install .
## 第四章: 进阶使用
在代码目录下advanced_version代码包中
-- 前端 static目录下
+- 前端
+ - 基础gui版本 computer-use-server/static目录下
+ - 高级版本 在代码目录下adk-computer-use代码包中
+ - 具体参考：[README_zh.md](./advanced_version/adk-computer-use/README.md)
- 后端 backend.py
具体参考：[README_zh.md](./advanced_version/computer_use_server/README_zh.md)
\ No newline at end of file
diff --git a/demos/computer_use/advanced_version/README_en.md b/demos/computer_use/advanced_version/README_en.md
new file mode 100644
index 0000000..56e8f83
--- /dev/null
+++ b/demos/computer_use/advanced_version/README_en.md
@@ -0,0 +1,11 @@
+## Computer Use Agent Advanced 🤖
+
+### Chapter 1: Backend Code
+
+The backend code is located in the `adk-computer-use` code package within the code directory.
+For specific details, please refer to: [README_zh.md](./adk-computer-use/README.md)
+
+### Chapter 2: Frontend Code
+
+The frontend code is located in the `advanced_version` code package within the code directory.
+For specific details, please refer to: [README_zh.md](./computer_use_server/README_en.md)
\ No newline at end of file
diff --git a/demos/computer_use/advanced_version/README_zh.md b/demos/computer_use/advanced_version/README_zh.md
new file mode 100644
index 0000000..1145e25
--- /dev/null
+++ b/demos/computer_use/advanced_version/README_zh.md
@@ -0,0 +1,11 @@
+# Computer Use Agent 进阶🤖
+
+## 第一章：后端code
+
+在代码目录下adk-computer-use代码包中
+具体参考：[README_zh.md](./adk-computer-use/README.md)
+
+## 第二章: 前端code
+
+在代码目录下advanced_version代码包中
+具体参考：[README_zh.md](./computer_use_server/README_zh.md)
\ No newline at end of file
diff --git a/demos/computer_use/advanced_version/computer_use_server/Dockerfile b/demos/computer_use/advanced_version/computer_use_server/Dockerfile
deleted file mode 100644
index fd51050..0000000
--- a/demos/computer_use/advanced_version/computer_use_server/Dockerfile
+++ /dev/null
@@ -1,46 +0,0 @@
-# 使用基础镜像
-FROM modelscope-registry.cn-beijing.cr.aliyuncs.com/modelscope-repo/python:3.10
-
-# 配置 pip 使用国内镜像源
-RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
-RUN pip config set global.trusted-host pypi.tuna.tsinghua.edu.cn
-
-# 切换到root用户执行需要权限的操作
-USER root
-
-# 更新包索引并安装 Nginx 和其他必要工具
-RUN apt-get update && \
- apt-get install -y nginx sudo && \
- rm -rf /var/lib/apt/lists/*
-
-# 创建用户并配置sudo权限
-RUN useradd -m -u 1000 user && \
- echo "user ALL=(ALL) NOPASSWD: ALL" >> /etc/sudoers
-USER user
-ENV PATH="/home/user/.local/bin:$PATH"
-
-# 设置工作目录
-WORKDIR /app
-
-# 复制依赖文件
-COPY --chown=user ./requirements.txt requirements.txt
-
-# 安装依赖
-RUN pip install --no-cache-dir --upgrade pip
-RUN for i in {1..10}; do pip install --no-cache-dir --upgrade -r requirements.txt && break || sleep 5; done
-RUN pip install --no-cache-dir spy-agent-build-sdk==0.0.29
-
-# 复制应用程序代码
-COPY --chown=user . /app
-
-# 复制 Nginx 配置文件
-COPY nginx.conf /etc/nginx/nginx.conf
-
-# 确保 start.sh 具有可执行权限
-RUN chmod +x /app/start.sh
-
-# 暴露端口
-EXPOSE 7860
-
-# 启动服务
-CMD ["/bin/bash", "/app/start.sh"]
diff --git a/demos/computer_use/advanced_version/computer_use_server/README_en.md b/demos/computer_use/advanced_version/computer_use_server/README_en.md
index dcb285d..19afff1 100644
--- a/demos/computer_use/advanced_version/computer_use_server/README_en.md
+++ b/demos/computer_use/advanced_version/computer_use_server/README_en.md
@@ -166,8 +166,8 @@ nano ~/.bash_profile # If you are using bash
# Add environment variables for example
# Cloud computer configuration
-export DASHSCOPE_API_KEY=""
-export ECD_DESKTOP_ID="your_desktop_id"
+export DASHSCOPE_API_KEY=
+export ECD_DESKTOP_ID=
# ... other configurations
# Save and run
diff --git a/demos/computer_use/advanced_version/computer_use_server/README_zh.md b/demos/computer_use/advanced_version/computer_use_server/README_zh.md
index 0968753..8a32da5 100644
--- a/demos/computer_use/advanced_version/computer_use_server/README_zh.md
+++ b/demos/computer_use/advanced_version/computer_use_server/README_zh.md
@@ -163,8 +163,8 @@ nano ~/.bash_profile # 如果你用的是 bash
# 添加环境变量例如
# 云电脑配置
-export DASHSCOPE_API_KEY=""
-export ECD_DESKTOP_ID="your_desktop_id"
+export DASHSCOPE_API_KEY=
+export ECD_DESKTOP_ID=
# ... 其他配置
# 保存后运行
diff --git a/demos/computer_use/advanced_version/computer_use_server/backend.py b/demos/computer_use/advanced_version/computer_use_server/backend.py
index ac16ba6..691fb49 100644
--- a/demos/computer_use/advanced_version/computer_use_server/backend.py
+++ b/demos/computer_use/advanced_version/computer_use_server/backend.py
@@ -6,6 +6,7 @@
import requests
import uuid
import weakref
+import socket
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse, JSONResponse
@@ -282,6 +283,153 @@ def to_dict(self):
# 全局状态管理器
state_manager = None
+# 机器标识
+MACHINE_ID = os.getenv("MACHINE_ID", socket.gethostname())
+
+
+# Redis发布订阅控制信号
+async def publish_control_signal(
+ user_id: str,
+ chat_id: str,
+ action: str,
+ **kwargs,
+):
+ """发布控制信号到Redis"""
+ if not state_manager:
+ return False
+
+ try:
+ channel = f"control:{user_id}:{chat_id}"
+ signal_data = {
+ "action": action,
+ "timestamp": time.time(),
+ "machine_id": MACHINE_ID,
+ **kwargs,
+ }
+
+ await state_manager.redis_client.publish(
+ channel,
+ json.dumps(signal_data),
+ )
+ logger.info(f"发布控制信号: {action} to {channel}")
+ return True
+ except Exception as e:
+ logger.error(f"发布控制信号失败: {e}")
+ return False
+
+
+async def listen_for_control_signals(user_id: str, chat_id: str, agent):
+ """监听Redis控制信号的后台任务"""
+ if not state_manager:
+ return
+
+ channel = f"control:{user_id}:{chat_id}"
+ pubsub = state_manager.redis_client.pubsub()
+
+ try:
+ await pubsub.subscribe(channel)
+ logger.info(f"开始监听控制信号: {channel}")
+
+ async for message in pubsub.listen():
+ if message["type"] == "message":
+ try:
+ signal_data = json.loads(message["data"])
+ action = signal_data.get("action")
+
+ logger.info(f"收到控制信号: {action} from {channel}")
+
+ if action == "stop":
+ # 设置停止标志
+ if hasattr(agent, "should_stop"):
+ agent.should_stop = True
+ # 更新Redis状态
+ await state_manager.update_chat_state(
+ user_id,
+ chat_id,
+ {"is_running": False, "stop_requested": True},
+ )
+ elif action == "interrupt_wait":
+ # 调用中断等待方法
+ if hasattr(agent, "interrupt_wait"):
+ agent.interrupt_wait()
+
+ except Exception as e:
+ logger.error(f"处理控制信号时出错: {e}")
+
+ except Exception as e:
+ logger.error(f"监听控制信号失败: {e}")
+ finally:
+ try:
+ await pubsub.unsubscribe(channel)
+ await pubsub.close()
+ except Exception as e:
+ logger.error(f"关闭pub/sub连接失败: {e}")
+
+
+async def check_stop_signal_from_redis(user_id: str, chat_id: str) -> bool:
+ """检查Redis中的停止信号"""
+ try:
+ chat_state = await state_manager.get_chat_state(user_id, chat_id)
+ if isinstance(chat_state, dict):
+ return not chat_state.get("is_running", True) or chat_state.get(
+ "stop_requested",
+ False,
+ )
+ else:
+ return not getattr(chat_state, "is_running", True) or getattr(
+ chat_state,
+ "stop_requested",
+ False,
+ )
+ except Exception as e:
+ logger.error(f"检查停止信号失败: {e}")
+ return False
+
+
+# Redis操作超时保护函数
+async def safe_redis_operation(
+ operation_func,
+ *args,
+ timeout=10.0,
+ max_retries=3,
+ **kwargs,
+):
+ """
+ 安全的Redis操作，带重试和超时保护
+
+ Args:
+ operation_func: Redis操作函数
+ timeout: 超时时间（秒）
+ max_retries: 最大重试次数
+ *args, **kwargs: 传递给操作函数的参数
+
+ Returns:
+ 操作结果，如果失败返回None
+ """
+ for attempt in range(max_retries):
+ try:
+ return await asyncio.wait_for(
+ operation_func(*args, **kwargs),
+ timeout=timeout,
+ )
+ except asyncio.TimeoutError:
+ logger.warning(f"Redis操作超时，尝试 {attempt + 1}/{max_retries}")
+ if attempt < max_retries - 1:
+ await asyncio.sleep(1.0 * (attempt + 1)) # 递增延迟
+ else:
+ logger.error("Redis操作最终超时失败")
+ return None
+ except Exception as e:
+ logger.warning(
+ f"Redis操作失败，尝试 {attempt + 1}/{max_retries}: {e}",
+ )
+ if attempt < max_retries - 1:
+ await asyncio.sleep(1.0 * (attempt + 1))
+ else:
+ logger.error(f"Redis操作最终失败: {e}")
+ return None
+ return None
+
if not hasattr(app.state, "running_agents"):
app.state.running_agents = weakref.WeakValueDictionary()
@@ -345,21 +493,6 @@ async def validate_user_session(
return False
-@app.on_event("startup")
-async def startup_event():
- global state_manager
- # 使用Redis状态管理器
- state_manager = RedisStateManager(
- phone_instance_ids=PHONE_INSTANCE_IDS,
- desktop_ids=DESKTOP_IDS,
- )
- await state_manager.initialize()
-
- # 启动心跳监控任务
- asyncio.create_task(state_manager._monitor_heartbeats())
- logger.info("Backend startup completed with Redis state manager")
-
-
# 请求/响应模型
class MessageContent(BaseModel):
type: str
@@ -380,7 +513,7 @@ class AgentConfig(BaseModel):
save_logs: bool = True
timeout: int = 120
pc_use_addon_info: str = ""
- max_steps: int = 15
+ max_steps: int = 20
user_id: str = ""
chat_id: str = ""
e2e_info: list = []
@@ -653,16 +786,15 @@ async def get_equipment(
)
# 处理资源排队情况
if status == AllocationStatus.WAIT_TIMEOUT:
+ pc_a = state_manager.pc_allocator
position = (
- state_manager.pc_allocator.get_chat_wait_position(
+ await pc_a.get_chat_position(
f"{user_id}:{chat_id}",
- )[0]
- )
- total_waiting = (
- state_manager.pc_allocator.get_queue_info()[
- "total_waiting"
- ]
- )
+ )
+ )[0]
+ total_waiting = (await pc_a.get_queue_info_async())[
+ "total_waiting"
+ ]
raise HTTPException(
status_code=429,
detail={
@@ -739,16 +871,15 @@ async def get_equipment(
)
# 处理资源排队情况
if status == AllocationStatus.WAIT_TIMEOUT:
+ pc_a = state_manager.phone_allocator
position = (
- state_manager.phone_allocator.get_chat_wait_position(
+ await pc_a.get_chat_position(
f"{user_id}:{chat_id}",
- )[0]
- )
- total_waiting = (
- state_manager.phone_allocator.get_queue_info()[
- "total_waiting"
- ]
- )
+ )
+ )[0]
+ total_waiting = (await pc_a.get_queue_info_async())[
+ "total_waiting"
+ ]
logger.warning(
f"Failed to allocate phone resource: {status}",
)
@@ -834,86 +965,6 @@ async def get_equipment(
}
-# API端点
-@app.post("/cua/init")
-async def init_task(request: InitRequest, user_id: str = ""):
- """触发异步环境初始化"""
- if not user_id:
- if request.user_id:
- user_id = request.user_id
- else:
- user_id = request.config.user_id
- logger.info(f"start init by user_id:{user_id}")
- chat_id = request.config.chat_id
- if not user_id or not chat_id:
- raise HTTPException(
- status_code=400,
- detail="user_id and chat_id are required",
- )
- logger.info(
- f"接收到任务请求，用户: {user_id}, 对话: {chat_id} "
- f"request: {json.dumps(request.model_dump(), ensure_ascii=False)}",
- )
- try:
- # 启动异步环境初始化操作
- operation_id = await state_manager.start_environment_operation(
- user_id,
- chat_id,
- "init",
- request.config.dict(),
- )
-
- return {
- "success": True,
- "operation_id": operation_id,
- "message": "Environment initialization started",
- "status": "initializing",
- }
- except HTTPException as he:
- raise he
- except Exception as e:
- print(f"Error starting init operation: {e}")
- raise HTTPException(status_code=500, detail=str(e))
-
-
-@app.post("/cua/run")
-async def run_task_stream(request: ComputerUseRequest, user_id: str = ""):
- """
- 流式任务执行接口，支持序列号机制和断线续传
- 如果提供sequence_number，则返回历史数据；否则执行新任务
- """
- # 从请求中提取对话ID和配置
- if not user_id:
- if request.user_id:
- user_id = request.user_id
- else:
- user_id = request.config.user_id
- chat_id = request.config.chat_id if request.config else ""
- logger.info(f"start run by user_id:{user_id}")
- if not user_id or not chat_id:
- raise HTTPException(400, "user_id and chat_id are required in config")
-
- # 验证用户会话有效性（非严格模式，因为心跳可能还没建立）
- await validate_user_session(user_id, chat_id, strict_mode=False)
-
- sequence_number = request.sequence_number
-
- logger.info(
- f"接收到任务请求，用户: {user_id}, 对话: {chat_id}, "
- f"序列号: {sequence_number} , request: "
- f"{json.dumps(request.model_dump(), ensure_ascii=False)}",
- )
- # 如果提供了序列号，返回历史数据（断线续传）
- if sequence_number is not None:
- return await _handle_resume_stream(user_id, chat_id, sequence_number)
-
- # 否则执行新任务
- if not request.input:
- raise HTTPException(400, "No input messages provided")
-
- return await _handle_new_stream(user_id, chat_id, request)
-
-
async def _handle_resume_stream(
user_id: str,
chat_id: str,
@@ -998,12 +1049,14 @@ async def resume_stream():
resume_stream(),
media_type="text/event-stream",
headers={
- "Cache-Control": "no-cache",
+ "Cache-Control": "no-cache, no-store, must-revalidate",
"Connection": "keep-alive",
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Credentials": "true",
- "X-Accel-Buffering": "no",
- "Content-Type": "text/event-stream",
+ "X-Accel-Buffering": "no", # 禁用nginx缓冲
+ "Content-Type": "text/event-stream; charset=utf-8",
+ "Transfer-Encoding": "chunked",
+ "Keep-Alive": "timeout=300, max=1000", # 设置keep-alive参数
},
)
@@ -1088,9 +1141,54 @@ def __init__(self, request):
logger.info(f"开始Agent执行，用户: {user_id}, 对话: {chat_id}")
# 执行Agent任务并处理流式输出
async_iterator = None
+
+ # 心跳机制变量
+ last_heartbeat = time.time()
+ heartbeat_interval = 30 # 30秒心跳间隔
+
try:
async_iterator = agent.run_async(context)
async for result in async_iterator:
+ # 检查是否需要发送心跳
+ current_time = time.time()
+ if current_time - last_heartbeat >= heartbeat_interval:
+ heartbeat_data = {
+ "object": "heartbeat",
+ "type": "heartbeat",
+ "timestamp": current_time,
+ "status": "alive",
+ "user_id": user_id,
+ "chat_id": chat_id,
+ }
+
+ # 尝试存储心跳到Redis（失败也不影响发送）
+ try:
+ heartbeat_sequence = await safe_redis_operation(
+ state_manager.store_stream_data,
+ user_id,
+ chat_id,
+ heartbeat_data,
+ task_id,
+ timeout=5.0,
+ max_retries=1,
+ )
+ heartbeat_data["sequence_number"] = (
+ heartbeat_sequence
+ )
+ except Exception as heartbeat_redis_error:
+ logger.warning(
+ f"心跳存储到Redis失败: {heartbeat_redis_error}",
+ )
+ heartbeat_data["sequence_number"] = None
+
+ # 发送心跳
+ heartbeat_json = json.dumps(
+ heartbeat_data,
+ ensure_ascii=False,
+ )
+ yield f"data: {heartbeat_json}\n\n"
+ last_heartbeat = current_time
+
try:
# 将Agent的输出转换为JSON格式
if hasattr(result, "model_dump"):
@@ -1099,15 +1197,24 @@ def __init__(self, request):
result_dict = _serialize(result)
# 直接使用Agent返回的原始数据，只添加序列号
- sequence_number = (
- await state_manager.store_stream_data(
- user_id,
- chat_id,
- result_dict,
- task_id,
- )
+ sequence_number = await safe_redis_operation(
+ state_manager.store_stream_data,
+ user_id,
+ chat_id,
+ result_dict,
+ task_id,
+ timeout=10.0,
+ max_retries=2,
)
- result_dict["sequence_number"] = sequence_number
+
+ if sequence_number is not None:
+ result_dict["sequence_number"] = sequence_number
+ else:
+ # Redis存储失败，但仍然发送数据
+ result_dict["sequence_number"] = None
+ result_dict["storage_warning"] = (
+ "数据未能存储到Redis"
+ )
json_str = json.dumps(result_dict, ensure_ascii=False)
yield f"data: {json_str}\n\n"
@@ -1120,40 +1227,58 @@ def __init__(self, request):
"type": "serialization_error",
}
- sequence_number = (
- await state_manager.store_stream_data(
- user_id,
- chat_id,
- error_data,
- task_id,
- )
+ sequence_number = await safe_redis_operation(
+ state_manager.store_stream_data,
+ user_id,
+ chat_id,
+ error_data,
+ task_id,
+ timeout=5.0,
+ max_retries=1,
)
- # 获取Redis中已标准化的错误数据
- stored_error_list = (
- await state_manager.get_stream_data_from_sequence(
- user_id,
- chat_id,
- sequence_number,
- task_id,
+ # 获取Redis中已标准化的错误数据（如果存储成功）
+ if sequence_number is not None:
+ stored_error_list = (
+ await state_manager.get_stream_seq(
+ user_id,
+ chat_id,
+ sequence_number,
+ task_id,
+ )
)
- )
- if stored_error_list:
- _d = json.dumps(
- stored_error_list[0],
- ensure_ascii=False,
- )
- yield f"data: {_d}\n\n"
+ if stored_error_list:
+ _d = json.dumps(
+ stored_error_list[0],
+ ensure_ascii=False,
+ )
+ yield f"data: {_d}\n\n"
+ else:
+ # 降级方案
+ error_output = {
+ "sequence_number": sequence_number,
+ "object": "error",
+ "status": "error",
+ "error": str(serialize_error),
+ "type": "error",
+ "data": error_data,
+ }
+ _data = json.dumps(
+ error_output,
+ ensure_ascii=False,
+ )
+ yield f"data: {_data}\n\n"
else:
- # 降级方案
+ # Redis存储失败，直接发送错误信息
error_output = {
- "sequence_number": sequence_number,
+ "sequence_number": None,
"object": "error",
"status": "error",
"error": str(serialize_error),
"type": "error",
"data": error_data,
+ "storage_warning": "错误信息未能存储到Redis",
}
_data = json.dumps(
error_output,
@@ -1170,35 +1295,55 @@ def __init__(self, request):
"type": "iteration_error",
}
- sequence_number = await state_manager.store_stream_data(
+ sequence_number = await safe_redis_operation(
+ state_manager.store_stream_data,
user_id,
chat_id,
error_data,
task_id,
+ timeout=5.0,
+ max_retries=1,
)
- # 获取Redis中已标准化的错误数据
- stored_error_list = (
- await state_manager.get_stream_data_from_sequence(
- user_id,
- chat_id,
- sequence_number,
- task_id,
+ # 获取Redis中已标准化的错误数据（如果存储成功）
+ if sequence_number is not None:
+ stored_error_list = (
+ await state_manager.get_stream_data_from_sequence(
+ user_id,
+ chat_id,
+ sequence_number,
+ task_id,
+ )
)
- )
- if stored_error_list:
- _d = json.dumps(stored_error_list[0], ensure_ascii=False)
- yield f"data: {_d}\n\n"
+ if stored_error_list:
+ _d = json.dumps(
+ stored_error_list[0],
+ ensure_ascii=False,
+ )
+ yield f"data: {_d}\n\n"
+ else:
+ # 降级方案
+ error_output = {
+ "sequence_number": sequence_number,
+ "object": "error",
+ "status": "error",
+ "error": str(iteration_error),
+ "type": "error",
+ "data": error_data,
+ }
+ _d = json.dumps(error_output, ensure_ascii=False)
+ yield f"data: {_d}\n\n"
else:
- # 降级方案
+ # Redis存储失败，直接发送错误信息
error_output = {
- "sequence_number": sequence_number,
+ "sequence_number": None,
"object": "error",
"status": "error",
"error": str(iteration_error),
"type": "error",
"data": error_data,
+ "storage_warning": "执行错误信息未能存储到Redis",
}
_d = json.dumps(error_output, ensure_ascii=False)
yield f"data: {_d}\n\n"
@@ -1225,35 +1370,55 @@ def __init__(self, request):
}
try:
- sequence_number = await state_manager.store_stream_data(
+ sequence_number = await safe_redis_operation(
+ state_manager.store_stream_data,
user_id,
chat_id,
error_data,
task_id,
+ timeout=5.0,
+ max_retries=1,
)
- # 获取Redis中已标准化的错误数据
- stored_error_list = (
- await state_manager.get_stream_data_from_sequence(
- user_id,
- chat_id,
- sequence_number,
- task_id,
+ # 获取Redis中已标准化的错误数据（如果存储成功）
+ if sequence_number is not None:
+ stored_error_list = (
+ await state_manager.get_stream_data_from_sequence(
+ user_id,
+ chat_id,
+ sequence_number,
+ task_id,
+ )
)
- )
- if stored_error_list:
- _d = json.dumps(stored_error_list[0], ensure_ascii=False)
- yield f"data: {_d}\n\n"
+ if stored_error_list:
+ _d = json.dumps(
+ stored_error_list[0],
+ ensure_ascii=False,
+ )
+ yield f"data: {_d}\n\n"
+ else:
+ # 降级方案
+ error_output = {
+ "sequence_number": sequence_number,
+ "object": "error",
+ "status": "error",
+ "error": str(e),
+ "type": "error",
+ "data": error_data,
+ }
+ _d = json.dumps(error_output, ensure_ascii=False)
+ yield f"data: {_d}\n\n"
else:
- # 降级方案
+ # Redis存储失败，直接发送错误信息
error_output = {
- "sequence_number": sequence_number,
+ "sequence_number": None,
"object": "error",
"status": "error",
"error": str(e),
"type": "error",
"data": error_data,
+ "storage_warning": "全局错误信息未能存储到Redis",
}
_d = json.dumps(error_output, ensure_ascii=False)
yield f"data: {_d}\n\n"
@@ -1296,16 +1461,102 @@ def __init__(self, request):
agent_stream(),
media_type="text/event-stream",
headers={
- "Cache-Control": "no-cache",
+ "Cache-Control": "no-cache, no-store, must-revalidate",
"Connection": "keep-alive",
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Credentials": "true",
- "X-Accel-Buffering": "no",
- "Content-Type": "text/event-stream",
+ "X-Accel-Buffering": "no", # 禁用nginx缓冲
+ "Content-Type": "text/event-stream; charset=utf-8",
+ "Transfer-Encoding": "chunked",
+ "Keep-Alive": "timeout=300, max=1000", # 设置keep-alive参数
},
)
+@app.on_event("startup")
+async def startup_event():
+ global state_manager
+ # 使用Redis状态管理器
+ state_manager = RedisStateManager(
+ phone_instance_ids=PHONE_INSTANCE_IDS,
+ desktop_ids=DESKTOP_IDS,
+ )
+ await state_manager.initialize()
+
+ # 同步实例ID配置到Redis，确保与环境变量保持一致
+ logger.info("Synchronizing instance IDs with environment variables")
+ await state_manager.sync_instance_ids(
+ phone_instance_ids=PHONE_INSTANCE_IDS,
+ desktop_ids=DESKTOP_IDS,
+ )
+
+ # 启动心跳监控任务
+ asyncio.create_task(state_manager._monitor_heartbeats())
+
+ # 注册当前机器到Redis
+ try:
+ await state_manager.redis_client.hset(
+ "machine_registry",
+ MACHINE_ID,
+ json.dumps(
+ {
+ "machine_id": MACHINE_ID,
+ "startup_time": time.time(),
+ "pid": os.getpid(),
+ },
+ ),
+ )
+ logger.info(f"机器 {MACHINE_ID} 已注册到Redis")
+ except Exception as e:
+ logger.error(f"注册机器信息失败: {e}")
+
+ logger.info(
+ "Backend startup completed with Redis "
+ f"state manager on machine {MACHINE_ID}",
+ )
+
+
+@app.post("/cua/init")
+async def init_task(request: InitRequest, user_id: str = ""):
+ """触发异步环境初始化"""
+ if not user_id:
+ if request.user_id:
+ user_id = request.user_id
+ else:
+ user_id = request.config.user_id
+ logger.info(f"start init by user_id:{user_id}")
+ chat_id = request.config.chat_id
+ if not user_id or not chat_id:
+ raise HTTPException(
+ status_code=400,
+ detail="user_id and chat_id are required",
+ )
+ logger.info(
+ f"接收到任务请求，用户: {user_id}, 对话: {chat_id} "
+ f"request: {json.dumps(request.model_dump(), ensure_ascii=False)}",
+ )
+ try:
+ # 启动异步环境初始化操作
+ operation_id = await state_manager.start_environment_operation(
+ user_id,
+ chat_id,
+ "init",
+ request.config.dict(),
+ )
+
+ return {
+ "success": True,
+ "operation_id": operation_id,
+ "message": "Environment initialization started",
+ "status": "initializing",
+ }
+ except HTTPException as he:
+ raise he
+ except Exception as e:
+ print(f"Error starting init operation: {e}")
+ raise HTTPException(status_code=500, detail=str(e))
+
+
@app.post("/cua/switch_environment")
async def switch_environment(request: InitRequest):
"""触发异步环境切换"""
@@ -1363,6 +1614,7 @@ async def get_operation_status(
)
try:
+ await state_manager.set_heartbeat(user_id, chat_id)
status_info = await state_manager.get_environment_operation(
user_id,
chat_id,
@@ -1376,6 +1628,7 @@ async def get_operation_status(
)
return {
"success": False,
+ "status": "failed",
"error": "Operation not found",
"user_id": user_id,
"chat_id": chat_id,
@@ -1392,15 +1645,88 @@ async def get_operation_status(
raise HTTPException(status_code=500, detail=str(e))
+@app.post("/cua/run")
+async def run_task_stream(request: ComputerUseRequest, user_id: str = ""):
+ """
+ 流式任务执行接口，支持序列号机制和断线续传
+ 如果提供sequence_number，则返回历史数据；否则执行新任务
+ """
+ # 从请求中提取对话ID和配置
+ if not user_id:
+ if request.user_id:
+ user_id = request.user_id
+ else:
+ user_id = request.config.user_id
+ chat_id = request.config.chat_id if request.config else ""
+ logger.info(f"start run by user_id:{user_id}")
+ if not user_id or not chat_id:
+ raise HTTPException(400, "user_id and chat_id are required in config")
+
+ # 验证用户会话有效性（非严格模式，因为心跳可能还没建立）
+ await validate_user_session(user_id, chat_id, strict_mode=False)
+
+ sequence_number = request.sequence_number
+
+ logger.info(
+ f"接收到任务请求，用户: {user_id}, 对话: {chat_id}, "
+ f"序列号: {sequence_number} , request: "
+ f"{json.dumps(request.model_dump(), ensure_ascii=False)}",
+ )
+ # 如果提供了序列号，返回历史数据（断线续传）
+ if sequence_number is not None:
+ return await _handle_resume_stream(user_id, chat_id, sequence_number)
+
+ # 否则执行新任务
+ if not request.input:
+ raise HTTPException(400, "No input messages provided")
+
+ return await _handle_new_stream(user_id, chat_id, request)
+
+
@app.get("/cua/stop")
async def stop_task(user_id: str, chat_id: str):
logger.info(f"stop task by user_id:{user_id} , chat_id: {chat_id}")
# 验证用户会话有效性（非严格模式，允许停止非活跃会话）
await validate_user_session(user_id, chat_id, strict_mode=False)
+ # 更新Redis中的停止状态
await state_manager.stop_task(user_id, chat_id)
- logger.info("task stopped")
- return {"success": True}
+
+ # 发布停止信号到所有机器
+ signal_sent = await publish_control_signal(user_id, chat_id, "stop")
+
+ # 记录运行Agent的机器信息
+ try:
+ chat_state = await state_manager.get_chat_state(user_id, chat_id)
+ if isinstance(chat_state, dict):
+ agent_machine = chat_state.get("agent_machine_id")
+ else:
+ agent_machine = getattr(chat_state, "agent_machine_id", None)
+
+ if agent_machine:
+ logger.info(f"停止信号已发送到机器 {agent_machine}")
+ else:
+ logger.info("未找到Agent运行的机器信息，信号已广播")
+ except Exception as e:
+ logger.error(f"获取Agent机器信息失败: {e}")
+
+ # 同时尝试本地停止（如果Agent在当前机器）
+ try:
+ composite_key = f"{user_id}:{chat_id}"
+ local_agent = app.state.running_agents.get(composite_key)
+ if local_agent and hasattr(local_agent, "should_stop"):
+ local_agent.should_stop = True
+ logger.info("本地Agent停止标志已设置")
+
+ # 如果找到本地Agent，也将其从字典中移除
+ if local_agent:
+ del app.state.running_agents[composite_key]
+ logger.info("本地Agent引用已清理")
+ except Exception as e:
+ logger.error(f"设置本地Agent停止标志失败: {e}")
+
+ logger.info(f"task stopped, signal_sent: {signal_sent}")
+ return {"success": True, "signal_sent": signal_sent}
@app.post("/cua/release")
@@ -1431,16 +1757,6 @@ async def heartbeat(user_id: str, chat_id: str):
f"Invalid chat session for user {user_id}, chat {chat_id}"
f", active: {current_active_chat}",
)
- # 为失效的会话返回明确的状态信息
- return {
- "success": False,
- "is_active_chat": False,
- "status": "chat_inactive",
- "message": "Current chat is no longer active. "
- "Another session may be active for this user.",
- "active_chat_id": current_active_chat,
- "error_code": "CHAT_INACTIVE",
- }
# 只为有效会话更新心跳记录
await state_manager.set_heartbeat(user_id, chat_id)
@@ -1470,7 +1786,7 @@ async def get_queue_status(user_id: str, chat_id: str, sandbox_type: str):
if sandbox_type == "pc_wuyin":
# 使用PC分配器并调用异步方法
position, status = (
- await state_manager.pc_allocator.get_chat_wait_position_async(
+ await state_manager.pc_allocator.get_chat_position(
user_id,
)
)
@@ -1488,18 +1804,15 @@ async def get_queue_status(user_id: str, chat_id: str, sandbox_type: str):
}
elif sandbox_type == "phone_wuyin":
# 使用手机分配器并调用异步方法
- position, status = (
- await state_manager.phone_allocator.get_chat_wait_position_async( # noqa E501
- user_id=user_id,
- )
+ s_p = state_manager.phone_allocator
+ position, status = await s_p.get_chat_position( # noqa E501
+ user_id=user_id,
)
if (
status == AllocationStatus.SUCCESS
or status == AllocationStatus.CHAT_ALREADY_ALLOCATED
):
- queue_info = (
- await state_manager.phone_allocator.get_queue_info_async()
- )
+ queue_info = await s_p.get_queue_info_async()
return {
"position": position,
"total_waiting": queue_info["total_waiting"],
@@ -1531,7 +1844,7 @@ async def get_queue_status(user_id: str, chat_id: str, sandbox_type: str):
}
-# 添加代理端点来验证 studio token
+# 添加代理端点来验证（魔搭场景接口） studio token
@app.get("/cua/proxy/validate-studio-token")
async def proxy_validate_studio_token(studio_token: str = Query(...)):
"""
@@ -1559,6 +1872,8 @@ def make_request():
@app.get("/cua/interrupt_wait")
async def interrupt_wait(user_id: str, chat_id: str):
+ # 人工干预接口
+ logger.info(f"interrupt_wait by user_id:{user_id} , chat_id: {chat_id}")
# 验证用户会话有效性（非严格模式）
await validate_user_session(user_id, chat_id, strict_mode=False)
@@ -1568,22 +1883,42 @@ async def interrupt_wait(user_id: str, chat_id: str):
if isinstance(chat_state, dict):
is_running = chat_state.get("is_running", False)
agent_running = chat_state.get("agent_running", False)
+ agent_machine = chat_state.get("agent_machine_id")
else:
is_running = getattr(chat_state, "is_running", False)
agent_running = getattr(chat_state, "agent_running", False)
+ agent_machine = getattr(chat_state, "agent_machine_id", None)
# 没有正在运行的任务
if not is_running or not agent_running:
raise HTTPException(status_code=400, detail="No running task")
- # 从应用状态中获取agent实例
- composite_key = f"{user_id}:{chat_id}"
- agent = app.state.running_agents.get(composite_key)
- if agent is None:
- raise HTTPException(status_code=400, detail="Agent instance not found")
+ # 发布中断等待信号到所有机器
+ signal_sent = await publish_control_signal(
+ user_id,
+ chat_id,
+ "interrupt_wait",
+ )
+
+ # 记录运行Agent的机器信息
+ if agent_machine:
+ logger.info(f"中断等待信号已发送到机器 {agent_machine}")
+ else:
+ logger.info("未找到Agent运行的机器信息，信号已广播")
- # 调用agent的中断方法
- agent.interrupt_wait()
+ # 同时尝试本地中断（如果Agent在当前机器）
+ try:
+ composite_key = f"{user_id}:{chat_id}"
+ local_agent = app.state.running_agents.get(composite_key)
+ if local_agent and hasattr(local_agent, "interrupt_wait"):
+ local_agent.interrupt_wait()
+ logger.info("本地Agent中断等待已调用")
+ elif local_agent is None:
+ logger.info("本地未找到Agent实例")
+ else:
+ logger.warning("本地Agent不支持interrupt_wait方法")
+ except Exception as e:
+ logger.error(f"调用本地Agent中断等待失败: {e}")
# 向前端发送状态更新
await state_manager.update_status(
@@ -1595,7 +1930,7 @@ async def interrupt_wait(user_id: str, chat_id: str):
"message": "Stop-wait request received",
},
)
- return {"success": True}
+ return {"success": True, "signal_sent": signal_sent}
if __name__ == "__main__":
diff --git a/demos/computer_use/advanced_version/computer_use_server/computer_use_agent.py b/demos/computer_use/advanced_version/computer_use_server/computer_use_agent.py
index 98dad4d..40d0017 100644
--- a/demos/computer_use/advanced_version/computer_use_server/computer_use_agent.py
+++ b/demos/computer_use/advanced_version/computer_use_server/computer_use_agent.py
@@ -65,7 +65,7 @@ def __init__(
sandbox_type = config.get("sandbox_type", "pc_wuyin")
status_callback = config.get("status_callback")
pc_use_add_info = config.get("pc_use_add_info", "")
- max_steps = config.get("max_steps", 10)
+ max_steps = config.get("max_steps", 20)
chat_id = config.get("chat_id", "")
user_id = config.get("user_id", "")
e2e_info = config.get("e2e_info", [])
@@ -383,7 +383,7 @@ async def run_async(
"step": "",
"stage": "start",
"type": "text",
- "text": f"🤖 开始执行任务: {instruction}\\n模式: {self.mode}",
+ "text": f"🤖 开始执行任务: {instruction}",
},
)
# 清楚上一次的停止信号
@@ -429,7 +429,7 @@ async def _execute_computer_use_task(
"step": f"{step_count}",
"stage": "output",
"type": "text",
- "text": f"🔄 执行第 {step_count}步",
+ "text": f"🔄 第 {step_count} 步",
},
)
step_info = {
@@ -473,7 +473,7 @@ async def _execute_computer_use_task(
"step": f"{step_count}",
"stage": "output",
"type": "text",
- "text": "🔍 正在分析屏幕截图...",
+ "text": "🔍 分析屏幕截图",
},
)
@@ -520,7 +520,7 @@ async def _execute_computer_use_task(
"step": f"{step_count}",
"stage": "error",
"type": "text",
- "text": f"❌ 错误: {error_msg}",
+ "text": f"错误: {error_msg}",
},
)
raise analyse_error
@@ -542,7 +542,7 @@ async def _execute_computer_use_task(
"step": f"{step_count}",
"stage": "output",
"type": "text",
- "text": "⚡ 执行操作中...",
+ "text": "⚡ 执行操作",
},
)
@@ -657,7 +657,7 @@ async def _execute_computer_use_task(
"step": f"{step_count}",
"stage": "error",
"type": "text",
- "text": f"❌ {error_msg}",
+ "text": f"{error_msg}",
},
)
continue
@@ -678,7 +678,7 @@ async def _execute_computer_use_task(
"step": "",
"stage": "limit_completed",
"type": "text",
- "text": f"⚠️ 达到最大步数限制 ({self.max_steps})，任务停止",
+ "text": f"达到最大步数限制 ({self.max_steps})，任务停止",
},
)
break
@@ -695,14 +695,36 @@ async def _execute_computer_use_task(
break
except Exception as e:
- error_msg = f"执行任务时出错: {str(e)}"
- logger.error(error_msg)
+ error_msg = str(e)
+ # 检查是否为GUI服务请求失败的错误
+ if (
+ "Error querying" in error_msg
+ and "GUI服务请求失败" in error_msg
+ ):
+ # 尝试提取请求ID
+ import re
+
+ request_id_match = re.search(
+ r'"request_id":"([^"]+)"',
+ error_msg,
+ )
+ if request_id_match:
+ request_id = request_id_match.group(1)
+ formatted_error = (
+ f"内部agent调用异常，请求ID: {request_id}"
+ )
+ else:
+ formatted_error = "内部agent调用异常"
+ else:
+ formatted_error = f"执行任务时出错: {error_msg}"
+
+ logger.error(f"执行任务时出错: {error_msg}")
yield DataContent(
data={
"step": "",
"stage": "error",
"type": "text",
- "text": f"❌ {error_msg}",
+ "text": formatted_error,
},
)
finally:
@@ -1360,27 +1382,7 @@ async def analyse_screenshot(self, step_count: int = None):
result = json.dumps(result_data, ensure_ascii=False)
except Exception as e:
- yield DataContent(
- data={
- "step": f"{step_count}",
- "stage": "error",
- "type": "SYSTEM",
- "text": "Error querying PC use model %s" % e,
- },
- )
logger.error(f"Error querying PC use model: {e}")
-
- # 发送分析阶段失败状态，确保前端不会卡在AI分析阶段
- yield DataContent(
- data={
- "step": f"{step_count}",
- "stage": "error",
- "type": "analysis_stage",
- "text": "Analysis failed",
- "timestamp": time.time(),
- "uuid": str(uuid4()),
- },
- )
raise RuntimeError(f"Error querying PC use model: {e}")
elif self.mode == "phone_use":
try:
diff --git a/demos/computer_use/advanced_version/computer_use_server/docker_start.sh b/demos/computer_use/advanced_version/computer_use_server/docker_start.sh
deleted file mode 100755
index bb0293f..0000000
--- a/demos/computer_use/advanced_version/computer_use_server/docker_start.sh
+++ /dev/null
@@ -1,7 +0,0 @@
-
-docker build -t computer_use_agent .
-
-docker run -d \
- --env-file .env \
- -p 7860:7860 \
- computer_use_agent
\ No newline at end of file
diff --git a/demos/computer_use/advanced_version/computer_use_server/nginx.conf b/demos/computer_use/advanced_version/computer_use_server/nginx.conf
deleted file mode 100644
index 9daa573..0000000
--- a/demos/computer_use/advanced_version/computer_use_server/nginx.conf
+++ /dev/null
@@ -1,132 +0,0 @@
-worker_processes auto;
-
-events {
- worker_connections 768;
-}
-
-http {
- merge_slashes off;
- client_max_body_size 2000M;
-
- sendfile on;
- tcp_nopush on;
- tcp_nodelay on;
- keepalive_timeout 65;
- types_hash_max_size 2048;
- default_type application/octet-stream;
-
- gzip on;
-
- map $http_upgrade $connection_upgrade {
- default upgrade;
- '' close;
- }
-
- # 在 ModelScope 环境中，所有服务都需要通过 7860 端口访问
- upstream backend_server {
- server 127.0.0.1:8002; # 后端服务
- }
-
- upstream static_server {
- server 127.0.0.1:8001; # 静态文件服务
- }
-
- server {
- listen 7860;
- server_name localhost;
-
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
-
- add_header Access-Control-Allow-Origin "*" always;
-
- # 根路径 - 指向静态HTML首页
- location = / {
- proxy_pass http://static_server/index.html;
- proxy_set_header Host $host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
-
- # 静态资源路径
- location / {
- proxy_pass http://static_server/;
- proxy_set_header Host $host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
-
- # 后端API路径
- location /backend/ {
- proxy_pass http://backend_server/;
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
-
- # SSE状态流路径
- location /sse/ {
- proxy_pass http://backend_server/sse/;
- proxy_read_timeout 86400s;
- proxy_send_timeout 86400s;
- proxy_connect_timeout 300s;
-
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
-
- proxy_http_version 1.1;
- proxy_set_header Upgrade $http_upgrade;
- proxy_set_header Connection "upgrade";
-
- proxy_buffering off;
- proxy_cache off;
- proxy_redirect off;
- }
-
- # 后端初始化API路径 - 增加超时时间
- location /backend/cua/init {
- proxy_pass http://backend_server/cua/init;
- proxy_connect_timeout 300s;
- proxy_send_timeout 300s;
- proxy_read_timeout 300s;
-
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
-
- # 后端运行API路径 - 增加超时时间
- location /backend/cua/run {
- proxy_pass http://backend_server/cua/run;
- proxy_connect_timeout 300s;
- proxy_send_timeout 300s;
- proxy_read_timeout 300s;
-
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
-
- # 添加对验证studio token路径的超时配置
- location /backend/cua/proxy/validate-studio-token {
- proxy_pass http://backend_server/cua/proxy/validate-studio-token;
- proxy_connect_timeout 60s;
- proxy_send_timeout 60s;
- proxy_read_timeout 60s;
-
- proxy_set_header Host $http_host;
- proxy_set_header X-Real-IP $remote_addr;
- proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
- proxy_set_header X-Forwarded-Proto $scheme;
- }
- }
-}
diff --git a/demos/computer_use/advanced_version/computer_use_server/redis_resource_allocator.py b/demos/computer_use/advanced_version/computer_use_server/redis_resource_allocator.py
index 3a2167f..c186667 100644
--- a/demos/computer_use/advanced_version/computer_use_server/redis_resource_allocator.py
+++ b/demos/computer_use/advanced_version/computer_use_server/redis_resource_allocator.py
@@ -80,10 +80,6 @@ async def initialize(self):
self.FREE_INSTANCES_KEY,
*self.instance_ids,
)
- await self.redis.expire(
- self.FREE_INSTANCES_KEY,
- self.ALLOCATION_TTL,
- )
logger.info(
f"Initialized {self.resource_type} resource pool with "
f"{len(self.instance_ids)} instances",
@@ -187,10 +183,6 @@ async def release_async(self, instance_id: str) -> AllocationStatus:
# 将实例放回可用池
await self.redis.sadd(self.FREE_INSTANCES_KEY, instance_id)
- await self.redis.expire(
- self.FREE_INSTANCES_KEY,
- self.ALLOCATION_TTL,
- )
logger.info(f"[{self.resource_type}] 资源 {instance_id} 已释放")
@@ -211,7 +203,7 @@ async def get_chat_allocation_async(
return instance_id, AllocationStatus.SUCCESS
return "", AllocationStatus.NOT_ALLOCATED
- async def get_chat_wait_position_async(
+ async def get_chat_position(
self,
user_id: str,
) -> Tuple[int, AllocationStatus]:
diff --git a/demos/computer_use/advanced_version/computer_use_server/redis_state_manager.py b/demos/computer_use/advanced_version/computer_use_server/redis_state_manager.py
index 93e6fb5..c81801a 100644
--- a/demos/computer_use/advanced_version/computer_use_server/redis_state_manager.py
+++ b/demos/computer_use/advanced_version/computer_use_server/redis_state_manager.py
@@ -30,6 +30,9 @@
AppStreamClient,
)
+# 心跳超时时间（秒）
+heartbeat_timeout = float(os.getenv("HEARTBEAT_TIMEOUT", 120))
+
class EnvironmentOperationStatus(Enum):
"""环境操作状态"""
@@ -106,24 +109,225 @@ def __init__(
)
# 过期时间配置
- self.CHAT_STATE_TTL = 3600 # 对话状态1小时过期
- self.HEARTBEAT_TTL = 30 # 心跳30秒过期
+ self.CHAT_STATE_TTL = 21600 # 对话状态6小时过期
+ self.HEARTBEAT_TTL = 3600 # 心跳120秒过期
self.STATUS_QUEUE_TTL = 300 # 状态队列5分钟过期
- self.EQUIPMENT_TTL = 3600 # 设备信息1小时过期
+ self.EQUIPMENT_TTL = 3600 # 设备信息6小时过期
self.STREAM_DATA_TTL = 1800 # 流式数据30分钟过期
- self.USER_ACTIVE_CHAT_TTL = 7200 # 用户活跃chat_id映射2小时过期
+ self.USER_ACTIVE_CHAT_TTL = 3600 # 用户活跃chat_id映射6小时过期
# 内存存储（兼容原有接口）
self.status_queues: Dict[str, asyncio.Queue] = {}
self.heartbeats: Dict[str, float] = {}
+ # ====== 本地缓存机制 ======
+ # 缓存资源配置信息，减少频繁Redis查询
+ self._resource_config_cache = {}
+ self._cache_ttl = 60 # 缓存60秒
+ self._last_cache_time = {}
+
+ def _is_cache_valid(self, cache_key: str) -> bool:
+ """检查缓存是否有效"""
+ if cache_key not in self._last_cache_time:
+ return False
+ return time.time() - self._last_cache_time[cache_key] < self._cache_ttl
+
+ def _get_cached_value(self, cache_key: str) -> Any:
+ """获取缓存值"""
+ if self._is_cache_valid(cache_key):
+ return self._resource_config_cache.get(cache_key)
+ return None
+
+ def _set_cached_value(self, cache_key: str, value: Any):
+ """设置缓存值"""
+ self._resource_config_cache[cache_key] = value
+ self._last_cache_time[cache_key] = time.time()
+
+ async def get_resource_config_cached(
+ self,
+ resource_type: str,
+ ) -> Dict[str, Any]:
+ """
+ 获取资源配置信息（带缓存）
+
+ Args:
+ resource_type: 'pc' 或 'phone'
+
+ Returns:
+ Dict: 资源配置信息
+ """
+ cache_key = f"resource_config_{resource_type}"
+
+ # 尝试从缓存获取
+ cached_value = self._get_cached_value(cache_key)
+ if cached_value is not None:
+ return cached_value
+
+ # 缓存未命中，从Redis或内存获取
+ try:
+ if resource_type == "pc":
+ config = {
+ "total_instances": (
+ len(self.desktop_ids) if self.desktop_ids else 0
+ ),
+ "instance_ids": self.desktop_ids,
+ "allocator_available": self.pc_allocator is not None,
+ }
+ elif resource_type == "phone":
+ config = {
+ "total_instances": (
+ len(self.phone_instance_ids)
+ if self.phone_instance_ids
+ else 0
+ ),
+ "instance_ids": self.phone_instance_ids,
+ "allocator_available": self.phone_allocator is not None,
+ }
+ else:
+ config = {"error": f"Unknown resource type: {resource_type}"}
+
+ # 缓存结果
+ self._set_cached_value(cache_key, config)
+ return config
+
+ except Exception as e:
+ logger.error(f"获取资源配置失败: {e}")
+ return {"error": str(e)}
+
+ async def get_user_active_chat_cached(self, user_id: str) -> Optional[str]:
+ """获取用户活跃chat_id（带缓存）"""
+ cache_key = f"user_active_chat_{user_id}"
+
+ # 短期缓存（10秒），因为活跃会话信息变化较频繁
+ if cache_key in self._last_cache_time:
+ if time.time() - self._last_cache_time[cache_key] < 10:
+ return self._resource_config_cache.get(cache_key)
+
+ # 从Redis获取
+ chat_id = await self.get_user_active_chat(user_id)
+
+ # 缓存结果
+ self._resource_config_cache[cache_key] = chat_id
+ self._last_cache_time[cache_key] = time.time()
+
+ return chat_id
+
+ # ====== 性能监控工具 ======
+
+ async def redis_operation_with_timing(
+ self,
+ operation_name: str,
+ operation_func,
+ *args,
+ **kwargs,
+ ):
+ """
+ 带性能监控的Redis操作包装器
+
+ Args:
+ operation_name: 操作名称（用于日志）
+ operation_func: Redis操作函数
+ *args, **kwargs: 传递给操作函数的参数
+
+ Returns:
+ 操作结果
+ """
+ start_time = time.time()
+ try:
+ result = await operation_func(*args, **kwargs)
+ duration = time.time() - start_time
+
+ # 记录慢查询（超过50ms）
+ if duration > 0.05:
+ logger.warning(
+ f"Slow Redis operation: {operation_name} "
+ f"took {duration:.3f}s",
+ )
+ elif duration > 0.01:
+ logger.info(
+ f"Redis operation: {operation_name} took {duration:.3f}s",
+ )
+
+ return result
+
+ except Exception as e:
+ duration = time.time() - start_time
+ logger.error(
+ f"Redis operation failed: {operation_name} "
+ f"took {duration:.3f}s, error: {e}",
+ )
+ raise
+
+ def get_performance_stats(self) -> Dict[str, Any]:
+ """获取性能统计信息"""
+ try:
+ return {
+ "cache_stats": {
+ "total_cached_keys": len(self._resource_config_cache),
+ "cache_hit_keys": list(self._resource_config_cache.keys()),
+ "cache_ttl": self._cache_ttl,
+ },
+ "redis_pool_stats": {
+ "max_connections": getattr(
+ self.redis_client.connection_pool,
+ "max_connections",
+ "N/A",
+ ),
+ "created_connections": getattr(
+ self.redis_client.connection_pool,
+ "created_connections",
+ "N/A",
+ ),
+ },
+ "memory_stats": {
+ "status_queues_count": len(self.status_queues),
+ "heartbeats_count": len(self.heartbeats),
+ },
+ "resource_stats": {
+ "pc_instances": (
+ len(self.desktop_ids) if self.desktop_ids else 0
+ ),
+ "phone_instances": (
+ len(self.phone_instance_ids)
+ if self.phone_instance_ids
+ else 0
+ ),
+ "pc_allocator_active": self.pc_allocator is not None,
+ "phone_allocator_active": self.phone_allocator is not None,
+ },
+ }
+ except Exception as e:
+ logger.error(f"获取性能统计失败: {e}")
+ return {"error": str(e)}
+
async def initialize(self):
"""初始化Redis连接和资源分配器"""
- self.redis_client = redis.from_url(self.redis_url)
+ # 优化Redis连接池配置，减少多机器部署延迟
+ try:
+ self.redis_client = redis.from_url(
+ self.redis_url,
+ max_connections=3000, # 增加连接池大小
+ retry_on_timeout=True, # 超时重试
+ socket_connect_timeout=10, # 连接超时（秒）
+ socket_timeout=10, # 读写超时（秒）
+ socket_keepalive=True, # 启用keepalive
+ health_check_interval=30, # 健康检查间隔
+ )
+ except Exception as e:
+ logger.warning(f"Redis连接池优化配置失败，使用基础配置: {e}")
+ # 降级到基础配置
+ self.redis_client = redis.from_url(
+ self.redis_url,
+ max_connections=20,
+ retry_on_timeout=True,
+ socket_connect_timeout=5,
+ socket_timeout=10,
+ )
+
await self.redis_client.ping()
logger.info(
- f"Redis连接成功: "
+ f"Redis连接成功 (优化连接池): "
f"{self.redis_url.rpartition('@')[-1] or self.redis_url}",
)
@@ -156,6 +360,158 @@ async def initialize(self):
f"{len(self.desktop_ids)} instances",
)
+ async def sync_instance_ids(
+ self,
+ phone_instance_ids: List[str] = None,
+ desktop_ids: List[str] = None,
+ ):
+ """同步实例ID配置，重新初始化资源分配器
+
+ Args:
+ phone_instance_ids: 新的手机实例ID列表
+ desktop_ids: 新的桌面实例ID列表
+ """
+ needs_reinit = False
+
+ # 检查手机实例ID是否有变化
+ if phone_instance_ids is not None:
+ if phone_instance_ids != self.phone_instance_ids:
+ logger.info(
+ f"Phone instance IDs changed from "
+ f"{self.phone_instance_ids} to {phone_instance_ids}",
+ )
+ self.phone_instance_ids = phone_instance_ids
+ needs_reinit = True
+
+ # 重新初始化手机资源分配器
+ if self.phone_instance_ids:
+ if self.phone_allocator:
+ # 清理旧的分配器
+ try:
+ await self.phone_allocator.cleanup()
+ except Exception as e:
+ logger.warning(f"清理旧手机分配器时出错: {e}")
+
+ # 创建新的分配器
+ self.phone_allocator = AsyncRedisResourceAllocator(
+ "phone",
+ self.phone_instance_ids,
+ self.redis_client,
+ )
+ await self.phone_allocator.initialize()
+ # 强制重新初始化Redis中的实例池
+ await self._force_reinitialize_resource_pool(
+ self.phone_allocator,
+ )
+ logger.info(
+ f"Phone resource allocator reinitialized with "
+ f"{len(self.phone_instance_ids)} instances",
+ )
+ else:
+ # 如果实例ID列表为空，设置分配器为None
+ if self.phone_allocator:
+ try:
+ await self.phone_allocator.cleanup()
+ except Exception as e:
+ logger.warning(f"清理手机分配器时出错: {e}")
+ self.phone_allocator = None
+ logger.info(
+ "Phone resource allocator disabled (no instance IDs)",
+ )
+ else:
+ # 即使没有变化，也要确保Redis中的状态正确
+ if self.phone_allocator:
+ await self._force_reinitialize_resource_pool(
+ self.phone_allocator,
+ )
+ logger.info(
+ f"Phone resource pool re-synced with"
+ f" {len(self.phone_instance_ids)} instances",
+ )
+
+ # 检查桌面实例ID是否有变化
+ if desktop_ids is not None:
+ if desktop_ids != self.desktop_ids:
+ logger.info(
+ f"Desktop IDs changed from {self.desktop_ids}"
+ f" to {desktop_ids}",
+ )
+ self.desktop_ids = desktop_ids
+ needs_reinit = True
+
+ # 重新初始化PC资源分配器
+ if self.desktop_ids:
+ if self.pc_allocator:
+ # 清理旧的分配器
+ try:
+ await self.pc_allocator.cleanup()
+ except Exception as e:
+ logger.warning(f"清理旧PC分配器时出错: {e}")
+
+ # 创建新的分配器
+ self.pc_allocator = AsyncRedisResourceAllocator(
+ "pc",
+ self.desktop_ids,
+ self.redis_client,
+ )
+ await self.pc_allocator.initialize()
+ # 强制重新初始化Redis中的实例池
+ await self._force_reinitialize_resource_pool(
+ self.pc_allocator,
+ )
+ logger.info(
+ "PC resource allocator reinitialized "
+ f"with {len(self.desktop_ids)} "
+ "instances",
+ )
+ else:
+ # 如果实例ID列表为空，设置分配器为None
+ if self.pc_allocator:
+ try:
+ await self.pc_allocator.cleanup()
+ except Exception as e:
+ logger.warning(f"清理PC分配器时出错: {e}")
+ self.pc_allocator = None
+ logger.info(
+ "PC resource allocator disabled (no instance IDs)",
+ )
+ else:
+ # 即使没有变化，也要确保Redis中的状态正确
+ if self.pc_allocator:
+ await self._force_reinitialize_resource_pool(
+ self.pc_allocator,
+ )
+ logger.info(
+ "PC resource pool re-synced with "
+ f"{len(self.desktop_ids)}"
+ " instances",
+ )
+
+ if needs_reinit:
+ logger.info("Instance IDs synchronization completed")
+ else:
+ logger.info("Instance IDs re-synced (no changes detected)")
+
+ async def _force_reinitialize_resource_pool(self, allocator):
+ """强制重新初始化Redis中的实例池"""
+ try:
+ # 清除现有的免费实例池
+ await self.redis_client.delete(allocator.FREE_INSTANCES_KEY)
+
+ # 重新设置实例池
+ if allocator.instance_ids:
+ await self.redis_client.sadd(
+ allocator.FREE_INSTANCES_KEY,
+ *allocator.instance_ids,
+ )
+ logger.info(
+ f"Force reinitialized {allocator.resource_type} "
+ f"resource pool with "
+ f"{len(allocator.instance_ids)} instances",
+ )
+ except Exception as e:
+ logger.error(f"Force reinitialize resource pool failed: {e}")
+
async def close(self):
"""关闭Redis连接"""
if self.redis_client:
@@ -220,13 +576,78 @@ def _user_active_chat_key(self, user_id: str) -> str:
"""生成用户活跃chat_id存储key"""
return f"{self.USER_ACTIVE_CHAT_PREFIX}{user_id}"
+ # ====== Redis Pipeline 批量操作优化方法 ======
+
+ async def batch_redis_operations(
+ self,
+ operations: List[Dict[str, Any]],
+ ) -> List[Any]:
+ """
+ 批量执行Redis操作，减少网络延迟
+
+ Args:
+ operations: 操作列表，每个操作包含 {'method':
+ 'set', 'args': [...], 'kwargs': {...}}
+
+ Returns:
+ List[Any]: 操作结果列表
+ """
+ if not operations:
+ return []
+
+ pipe = self.redis_client.pipeline()
+
+ for op in operations:
+ method = op["method"]
+ args = op.get("args", [])
+ kwargs = op.get("kwargs", {})
+
+ # 根据方法名调用对应的pipeline方法
+ pipe_method = getattr(pipe, method)
+ pipe_method(*args, **kwargs)
+
+ return await pipe.execute()
+
+ async def batch_set_with_expiry(
+ self,
+ key_value_ttl_list: List[tuple],
+ ) -> List[Any]:
+ """
+ 批量设置带过期时间的键值对
+
+ Args:
+ key_value_ttl_list: [(key, value, ttl), ...] 格式的列表
+ """
+ if not key_value_ttl_list:
+ return []
+
+ pipe = self.redis_client.pipeline()
+ for key, value, ttl in key_value_ttl_list:
+ pipe.setex(key, ttl, value)
+
+ return await pipe.execute()
+
+ async def batch_delete_keys(self, keys: List[str]) -> int:
+ """批量删除键"""
+ if not keys:
+ return 0
+
+ # Redis DELETE 命令支持一次删除多个键
+ return await self.redis_client.delete(*keys)
+
async def set_user_active_chat(self, user_id: str, chat_id: str):
"""设置用户的活跃chat_id"""
key = self._user_active_chat_key(user_id)
await self.redis_client.setex(key, self.USER_ACTIVE_CHAT_TTL, chat_id)
+ async def delete_user_active_chat(self, user_id: str):
+ """设置用户的活跃chat_id"""
+ key = self._user_active_chat_key(user_id)
+ await self.redis_client.setex(key, self.USER_ACTIVE_CHAT_TTL, None)
+
async def get_user_active_chat(self, user_id: str) -> Optional[str]:
"""获取用户的活跃chat_id"""
+ logger.info(f"获取用户 {user_id} 的活跃chat_id")
key = self._user_active_chat_key(user_id)
chat_id = await self.redis_client.get(key)
if chat_id:
@@ -236,38 +657,59 @@ async def get_user_active_chat(self, user_id: str) -> Optional[str]:
return None
async def cleanup_user_old_sessions(self, user_id: str, new_chat_id: str):
- """清理用户的旧会话资源，只保留新的chat_id"""
+ """清理用户的旧会话资源，只保留新的chat_id - 优化版"""
try:
- # 获取用户当前的活跃chat_id
- old_chat_id = await self.get_user_active_chat(user_id)
+ # 并行获取用户活跃chat_id和chat_state，减少串行等待
+ old_chat_id, _ = await asyncio.gather(
+ self.get_user_active_chat(user_id),
+ # 预先设置新的活跃chat_id，避免后续等待
+ self.set_user_active_chat(user_id, new_chat_id),
+ return_exceptions=True,
+ )
if old_chat_id and old_chat_id != new_chat_id:
logger.info(
f"清理用户 {user_id} 的旧会话 {old_chat_id}，激活新会话 {new_chat_id}",
)
- # 停止旧会话的任务
+ # 并行获取旧会话状态并检查是否需要停止任务
old_chat_state = await self.get_chat_state(
user_id,
old_chat_id,
)
+
+ # 并行执行停止任务和资源释放
+ cleanup_tasks = []
+
if old_chat_state.get("is_running"):
- await self.stop_task(user_id, old_chat_id)
- await asyncio.sleep(0.5) # 给任务一点时间停止
+ cleanup_tasks.append(self.stop_task(user_id, old_chat_id))
- # 用户级别资源释放：释放用户的所有资源
- await self.release_user_resources(user_id)
+ # 添加资源释放和数据清理任务
+ cleanup_tasks.extend(
+ [
+ self.release_user_resources(user_id),
+ self.cleanup_chat_data(user_id, old_chat_id),
+ ],
+ )
- # 清理旧会话的数据
- await self.cleanup_chat_data(user_id, old_chat_id)
+ # 并行执行所有清理任务
+ if cleanup_tasks:
+ await asyncio.gather(
+ *cleanup_tasks,
+ return_exceptions=True,
+ )
+ if old_chat_state.get("is_running"):
+ await asyncio.sleep(0.3) # 稍微减少等待时间
logger.info(f"用户 {user_id} 旧会话 {old_chat_id} 清理完成")
- logger.info(f"用户 {user_id} 旧会话 {old_chat_id} 清理完成")
- # 设置新的活跃chat_id
- await self.set_user_active_chat(user_id, new_chat_id)
except Exception as e:
logger.error(f"清理用户 {user_id} 旧会话时出错: {e}")
+ # 确保新会话ID至少被设置
+ try:
+ await self.set_user_active_chat(user_id, new_chat_id)
+ except Exception as set_error:
+ logger.error(f"设置新活跃会话失败: {set_error}")
async def validate_user_active_chat(
self,
@@ -275,6 +717,7 @@ async def validate_user_active_chat(
chat_id: str,
) -> bool:
"""验证chat_id是否为该user_id的活跃会话"""
+ logger.info(f"验证用户 {user_id} 的活跃会话 {chat_id}")
active_chat_id = await self.get_user_active_chat(user_id)
return active_chat_id == chat_id
@@ -415,24 +858,42 @@ async def store_equipment(
},
)
- # 只存储设备信息，不序列化对象本身（避免线程锁序列化问题）
+ # 使用Pipeline同时存储设备信息和更新对话状态，减少网络调用
info_key = f"{self._equipment_key(user_id, chat_id)}_info"
- await self.redis_client.setex(
- info_key,
- self.EQUIPMENT_TTL,
- json.dumps(equipment_info, default=str),
- )
+ state_key = self._chat_state_key(user_id, chat_id)
- # 同时更新对话状态中的设备引用
- await self.update_chat_state(
- user_id,
- chat_id,
+ # 准备对话状态更新数据
+ current_state = await self.get_chat_state(user_id, chat_id)
+ current_state.update(
{
"equipment_storage_status": "stored_in_redis",
"equipment_info": equipment_info,
+ "updated_at": time.time(),
},
)
+ # 使用Pipeline批量执行
+ operations = [
+ {
+ "method": "setex",
+ "args": [
+ info_key,
+ self.EQUIPMENT_TTL,
+ json.dumps(equipment_info, default=str),
+ ],
+ },
+ {
+ "method": "setex",
+ "args": [
+ state_key,
+ self.CHAT_STATE_TTL,
+ json.dumps(current_state, default=str),
+ ],
+ },
+ ]
+
+ await self.batch_redis_operations(operations)
+
logger.info(
f"设备信息已存储到Redis，用户: {user_id}, 对话: {chat_id}, "
f"类型: {equipment_info['equipment_type']}",
@@ -619,7 +1080,7 @@ def _is_already_standardized(self, data: Dict[str, Any]) -> bool:
# 如果包含多个标准化字段，认为已经被标准化过了
return has_standardized_fields
- async def get_stream_data_from_sequence(
+ async def get_stream_seq(
self,
user_id: str,
chat_id: str,
@@ -782,7 +1243,7 @@ async def get_expired_users(self, timeout: int = 20) -> List[str]:
return expired_users
async def cleanup_chat_data(self, user_id: str, chat_id: str):
- """清理对话相关数据"""
+ """清理对话相关数据 - 优化版，使用批量删除减少网络调用"""
keys_to_delete = [
self._chat_state_key(user_id, chat_id),
self._status_queue_key(user_id, chat_id),
@@ -791,28 +1252,31 @@ async def cleanup_chat_data(self, user_id: str, chat_id: str):
f"{self._equipment_key(user_id, chat_id)}_info",
]
- # 清理环境操作相关key
+ # 并行扫描所有相关键，减少扫描时间
composite_key = self._composite_key(user_id, chat_id)
- pattern = f"{self.ENV_OPERATION_PREFIX}{composite_key}:*"
- async for key in self.redis_client.scan_iter(match=pattern):
- if isinstance(key, bytes):
- key = key.decode("utf-8")
- keys_to_delete.append(key)
+ scan_patterns = [
+ f"{self.ENV_OPERATION_PREFIX}{composite_key}:*", # 环境操作相关key
+ f"{self.STREAM_DATA_PREFIX}{composite_key}:*", # 流式数据相关key
+ f"{self.STREAM_COUNTER_PREFIX}{composite_key}:*", # 计数器相关key
+ ]
- # 清理流式数据相关key
- stream_pattern = f"{self.STREAM_DATA_PREFIX}{composite_key}:*"
- async for key in self.redis_client.scan_iter(match=stream_pattern):
- if isinstance(key, bytes):
- key = key.decode("utf-8")
- keys_to_delete.append(key)
+ # 并行扫描所有模式
+ scan_tasks = []
+ for pattern in scan_patterns:
+ scan_tasks.append(self._scan_keys_by_pattern(pattern))
- counter_pattern = f"{self.STREAM_COUNTER_PREFIX}{composite_key}:*"
- async for key in self.redis_client.scan_iter(match=counter_pattern):
- if isinstance(key, bytes):
- key = key.decode("utf-8")
- keys_to_delete.append(key)
+ # 等待所有扫描完成
+ scan_results = await asyncio.gather(
+ *scan_tasks,
+ return_exceptions=True,
+ )
- # 清理停止信号相关key
+ # 收集所有需要删除的键
+ for result in scan_results:
+ if isinstance(result, list):
+ keys_to_delete.extend(result)
+
+ # 添加停止信号相关key
stop_signal_key = f"computer_use:stop_signal:{composite_key}"
keys_to_delete.append(stop_signal_key)
@@ -820,17 +1284,211 @@ async def cleanup_chat_data(self, user_id: str, chat_id: str):
try:
current_active_chat = await self.get_user_active_chat(user_id)
if current_active_chat == chat_id:
- # 只有当被清理的chat_id确实是该用户的活跃会话时才清理
user_active_chat_key = self._user_active_chat_key(user_id)
keys_to_delete.append(user_active_chat_key)
logger.info(f"清理用户 {user_id} 的活跃会话映射: {chat_id}")
except Exception as e:
logger.warning(f"检查用户活跃会话时出错: {e}")
+ # 使用批量删除，一次性删除所有键
if keys_to_delete:
- await self.redis_client.delete(*keys_to_delete)
+ deleted_count = await self.batch_delete_keys(keys_to_delete)
+ logger.info(
+ f"已清理对话 {user_id}:{chat_id} 的 {deleted_count} 个键",
+ )
+ else:
+ logger.info(f"对话 {user_id}:{chat_id} 没有需要清理的数据")
+
+ async def _scan_keys_by_pattern(self, pattern: str) -> List[str]:
+ """根据模式扫描键，返回键列表"""
+ keys = []
+ try:
+ async for key in self.redis_client.scan_iter(match=pattern):
+ if isinstance(key, bytes):
+ key = key.decode("utf-8")
+ keys.append(key)
+ except Exception as e:
+ logger.warning(f"扫描键模式 {pattern} 时出错: {e}")
+ return keys
- logger.info(f"已清理对话 {user_id}:{chat_id} 的所有数据")
+ # ====== 资源分配优化方法 ======
+
+ async def allocate_resource_with_queue_info(
+ self,
+ resource_type: str,
+ user_id: str,
+ timeout: int = 0,
+ ) -> tuple:
+ """
+ 优化的资源分配方法，一次性返回分配结果和排队信息
+
+ Args:
+ resource_type: 'pc' 或 'phone'
+ user_id: 用户ID
+ timeout: 超时时间
+
+ Returns:
+ tuple: (resource_id, status, queue_info)
+ - resource_id: 分配的资源ID或None
+ - status: 分配状态
+ - queue_info: 排队信息字典 (position, total_waiting) 或None
+ """
+ try:
+ allocator = (
+ self.pc_allocator
+ if resource_type == "pc"
+ else self.phone_allocator
+ )
+ if not allocator:
+ return None, AllocationStatus.RESOURCE_EXHAUSTED, None
+
+ # 尝试分配资源
+ resource_id, status = await allocator.allocate_async(
+ user_id,
+ timeout=timeout,
+ )
+
+ # 如果需要排队，同时获取排队信息
+ queue_info = None
+ if status == AllocationStatus.WAIT_TIMEOUT:
+ # 并行获取排队位置和总排队数
+ position_task = allocator.get_chat_position(user_id)
+ queue_info_task = allocator.get_queue_info_async()
+
+ (position, _), queue_data = await asyncio.gather(
+ position_task,
+ queue_info_task,
+ return_exceptions=True,
+ )
+
+ if not isinstance(position, Exception) and not isinstance(
+ queue_data,
+ Exception,
+ ):
+ queue_info = {
+ "queue_position": position + 1,
+ "total_waiting": queue_data.get("total_waiting", 0),
+ }
+
+ return resource_id, status, queue_info
+
+ except Exception as e:
+ logger.error(f"资源分配优化方法出错: {e}")
+ return None, AllocationStatus.RESOURCE_EXHAUSTED, None
+
+ async def batch_update_environment_operations(
+ self,
+ updates: List[Dict[str, Any]],
+ ) -> List[bool]:
+ """
+ 批量更新环境操作状态
+
+ Args:
+ updates: 更新列表，每个元素包含:
+ {
+ 'user_id': str,
+ 'chat_id': str,
+ 'operation_id': str,
+ 'updates': dict
+ }
+
+ Returns:
+ List[bool]: 每个更新的成功状态
+ """
+ if not updates:
+ return []
+
+ # 并行获取所有操作的当前状态
+ get_tasks = []
+ for update in updates:
+ task = self.get_environment_operation(
+ update["user_id"],
+ update["chat_id"],
+ update["operation_id"],
+ )
+ get_tasks.append(task)
+
+ current_operations = await asyncio.gather(
+ *get_tasks,
+ return_exceptions=True,
+ )
+
+ # 准备批量更新操作
+ redis_operations = []
+ chat_state_updates = []
+ results = []
+
+ for i, (update, current_op) in enumerate(
+ zip(updates, current_operations),
+ ):
+ try:
+ if isinstance(current_op, Exception) or not current_op:
+ results.append(False)
+ continue
+
+ # 更新操作数据
+ current_op.update(update["updates"])
+ current_op["updated_at"] = time.time()
+
+ # 准备Redis操作
+ user_id = update["user_id"]
+ chat_id = update["chat_id"]
+ operation_id = update["operation_id"]
+
+ key = self._env_operation_key(user_id, chat_id, operation_id)
+ current_key = self._env_operation_key(user_id, chat_id)
+ operation_json = json.dumps(current_op, default=str)
+
+ redis_operations.extend(
+ [
+ {
+ "method": "setex",
+ "args": [key, 3600, operation_json],
+ },
+ {
+ "method": "setex",
+ "args": [current_key, 3600, operation_json],
+ },
+ ],
+ )
+
+ # 准备对话状态更新
+ chat_state_updates.append(
+ (user_id, chat_id, {"current_env_operation": current_op}),
+ )
+ results.append(True)
+
+ except Exception as e:
+ logger.error(f"准备环境操作更新时出错: {e}")
+ results.append(False)
+
+ # 批量执行Redis操作
+ try:
+ if redis_operations:
+ await self.batch_redis_operations(redis_operations)
+
+ # 批量更新对话状态
+ if chat_state_updates:
+ state_update_tasks = []
+ for user_id, chat_id, updates_data in chat_state_updates:
+ task = self.update_chat_state(
+ user_id,
+ chat_id,
+ updates_data,
+ )
+ state_update_tasks.append(task)
+
+ await asyncio.gather(
+ *state_update_tasks,
+ return_exceptions=True,
+ )
+
+ except Exception as e:
+ logger.error(f"批量执行环境操作更新时出错: {e}")
+ # 将所有结果标记为失败
+ results = [False] * len(updates)
+
+ return results
async def start_environment_operation(
self,
@@ -1132,12 +1790,12 @@ async def release_user_resources(self, user_id: str):
f"[release_user_resources] 清理用户 {user_id} 的排队状态",
)
pc_position, pc_wait_status = (
- await self.pc_allocator.get_chat_wait_position_async(
+ await self.pc_allocator.get_chat_position(
user_id,
)
)
phone_position, phone_wait_status = (
- await self.phone_allocator.get_chat_wait_position_async(
+ await self.phone_allocator.get_chat_position(
user_id,
)
)
@@ -1227,7 +1885,7 @@ async def _notify_queued_users(self, resource_type: str):
else:
return
- queue_info = allocator.get_queue_info()
+ queue_info = await allocator.get_queue_info_async()
queued_users = queue_info.get("waiting_users", [])
# 为第一个排队对话尝试分配资源
@@ -1277,8 +1935,6 @@ async def _notify_queued_users(self, resource_type: str):
async def _monitor_heartbeats(self):
"""心跳监控任务 - 增强版，支持更可靠的超时检测和彻底清理"""
logger.info("Redis heartbeat monitor started")
- heartbeat_timeout = 30 # 心跳超时时间（秒）
-
while True:
try:
# 同时检查Redis和内存中的心跳记录
@@ -1294,7 +1950,10 @@ async def _monitor_heartbeats(self):
# 方法2：检查内存中的心跳记录（兜底检查）
memory_expired = []
for chat_id, last_heartbeat in self.heartbeats.items():
- if current_time - last_heartbeat > heartbeat_timeout:
+ if (
+ last_heartbeat
+ and current_time - last_heartbeat > heartbeat_timeout
+ ):
memory_expired.append(chat_id)
# 合并去重
@@ -1323,7 +1982,10 @@ async def _monitor_heartbeats(self):
memory_heartbeat,
)
- if current_time - latest_heartbeat > heartbeat_timeout:
+ if (
+ latest_heartbeat
+ and current_time - latest_heartbeat > heartbeat_timeout
+ ):
confirmed_expired.append(composite_key)
logger.warning(
f"[HeartbeatMonitor] 确认用户心跳超时: {composite_key} "
@@ -1349,46 +2011,218 @@ async def _monitor_heartbeats(self):
user_id,
chat_id,
)
- if chat_state.get("is_running"):
- logger.info(
- f"[HeartbeatMonitor] 停止用户任务: {composite_key}",
+ if chat_state:
+ released_resource_type = chat_state.get(
+ "sandbox_type",
)
- await self.stop_task(user_id, chat_id)
- await asyncio.sleep(1) # 给任务一点时间停止
+ if chat_state.get("is_running"):
+ logger.info(
+ "[HeartbeatMonitor] 停止用户任务:"
+ f" {composite_key}",
+ )
+ await self.stop_task(user_id, chat_id)
+ await asyncio.sleep(1) # 给任务一点时间停止
- # 2. 彻底清理用户的所有Redis资源
- await self._thorough_cleanup_user_resources(
+ is_valid = await self.validate_user_active_chat(
user_id,
chat_id,
)
+ que_ask = False
+ if not is_valid:
+ activate_chat_id = await self.get_user_active_chat(
+ user_id,
+ )
+ if not activate_chat_id:
+ logger.info(
+ "当前用户没有活跃会话，可以进行彻底清理",
+ )
- # 3. 清理内存心跳记录
- if composite_key in self.heartbeats:
- del self.heartbeats[composite_key]
+ # 2. 彻底清理用户的所有Redis资源
+ await self._thorough_cleanup_u_res(
+ user_id,
+ chat_id,
+ )
+ # 3. 清理内存心跳记录
+ if composite_key in self.heartbeats:
+ del self.heartbeats[composite_key]
+ que_ask = True
+ logger.info(
+ f"[HeartbeatMonitor] 成功彻底清理用户资源"
+ f": {composite_key}",
+ )
- logger.info(
- f"[HeartbeatMonitor] 成功彻底清理用户资源: {composite_key}",
- )
+ else:
+ logger.info(
+ "检测当前用户{user_id}活跃会话"
+ f"{activate_chat_id}是否心跳过期",
+ )
+ redis_heartbeat = await self.get_heartbeat(
+ user_id,
+ activate_chat_id,
+ )
- except Exception as e:
+ # 修复后的代码
+ if (
+ redis_heartbeat
+ and current_time - redis_heartbeat
+ > heartbeat_timeout
+ ):
+ confirmed_expired.append(composite_key)
+ _time = current_time - redis_heartbeat
+ logger.warning(
+ "[HeartbeatMonitor] 确认用户"
+ f"{user_id}最新活跃会话心跳超时: 超时"
+ f"({_time:.1f}秒)",
+ )
+ await self._thorough_cleanup_u_res(
+ user_id,
+ chat_id,
+ )
+ # 3. 清理内存心跳记录
+ if composite_key in self.heartbeats:
+ del self.heartbeats[composite_key]
+ que_ask = True
+ logger.info(
+ "[HeartbeatMonitor] 成功彻底清理"
+ f"用户资源: {composite_key}",
+ )
+ else:
+ logger.info(
+ "当前用户还有活跃会话，不兜底清除资源,清除取消",
+ )
+ else:
+ logger.info(
+ f"检测当前用户{user_id}活跃会话{chat_id}是否心跳过期",
+ )
+ redis_heartbeat = await self.get_heartbeat(
+ user_id,
+ chat_id,
+ )
+
+ if (
+ redis_heartbeat
+ and current_time - redis_heartbeat
+ > heartbeat_timeout
+ ):
+ confirmed_expired.append(composite_key)
+ _time = current_time - redis_heartbeat
+ logger.warning(
+ f"[HeartbeatMonitor] 确认用户{user_id}"
+ "最新活跃会话心跳超时: "
+ f"(超时 {_time:.1f}秒)",
+ )
+ await self._thorough_cleanup_u_res(
+ user_id,
+ chat_id,
+ )
+ # 3. 清理内存心跳记录
+ if composite_key in self.heartbeats:
+ del self.heartbeats[composite_key]
+
+ logger.info(
+ "[HeartbeatMonitor] 成功彻底清理用户资源"
+ f": {composite_key}",
+ )
+
+ else:
+ logger.info(
+ "当前用户还有活跃会话，不兜底清除资源,清除取消",
+ )
+
+ if que_ask:
+ logger.info(
+ f"[release_user_resources] 清理用户"
+ f" {user_id} 的排队状态",
+ )
+ (
+ pc_position,
+ pc_wait_status,
+ ) = await self.pc_allocator.get_chat_position(
+ user_id,
+ )
+ (
+ phone_position,
+ phone_wait_status,
+ ) = await self.phone_allocator.get_chat_position(
+ user_id,
+ )
+
+ if pc_wait_status == AllocationStatus.SUCCESS:
+ await self.pc_allocator.cancel_wait_async(
+ user_id,
+ )
+ logger.info(
+ "[release_user_resources] 已取消用户"
+ f" {user_id} 的PC排队",
+ )
+ if phone_wait_status == AllocationStatus.SUCCESS:
+ await self.phone_allocator.cancel_wait_async(
+ user_id,
+ )
+ logger.info(
+ "[release_user_resources] 已取消"
+ f"用户 {user_id} 的手机排队",
+ )
+
+ # 通知排队用户资源可用
+ logger.info(
+ "[release_user_resources] "
+ f"释放了 {released_resource_type} "
+ "资源，通知排队用户",
+ )
+ asyncio.create_task(
+ self._notify_queued_users(
+ released_resource_type,
+ ),
+ )
+
+ except (
+ Exception
+ ) as user_cleanup_exception: # 修复2: 正确命名异常变量
logger.error(
f"[HeartbeatMonitor] 清理用户 {composite_key} "
- f"资源时出错: {e}",
+ f"资源时出错: {user_cleanup_exception}",
)
import traceback
logger.error(
- f"[HeartbeatMonitor] 错误详情: "
- f"{traceback.format_exc()}",
+ f"[HeartbeatMonitor] 错误详情"
+ f": {traceback.format_exc()}",
)
await asyncio.sleep(10) # 调整监控间隔为10秒，更及时响应
- except Exception as e:
- logger.error(f"Redis heartbeat monitor error: {e}")
+ except Exception as monitor_exception: # 修复3: 正确命名异常变量
+ logger.error(
+ f"Redis heartbeat monitor error: {monitor_exception}",
+ )
+ import traceback
+
+ logger.error(
+ f"Monitor error details: {traceback.format_exc()}",
+ )
await asyncio.sleep(10)
- async def _thorough_cleanup_user_resources(
+ async def clear_user_chat_redis_data(
+ self,
+ user_id: str,
+ chat_id: str = None,
+ ):
+ if not chat_id:
+ chat_id = await self.get_user_active_chat(user_id)
+ is_valid = await self.validate_user_active_chat(
+ user_id,
+ chat_id,
+ )
+
+ if not is_valid:
+ # 2. 彻底清理用户的所有Redis资源
+ await self._thorough_cleanup_u_res(
+ user_id,
+ chat_id,
+ )
+
+ async def _thorough_cleanup_u_res(
self,
user_id: str,
chat_id: str,
@@ -1396,7 +2230,7 @@ async def _thorough_cleanup_user_resources(
"""彻底清理用户的所有Redis资源 - 增强版"""
try:
logger.info(
- f"[_thorough_cleanup_user_resources] 开始彻底清理用户 "
+ f"[_thorough_cleanup_u_res] 开始彻底清理用户 "
f"{user_id}:{chat_id} 的所有资源",
)
@@ -1405,27 +2239,25 @@ async def _thorough_cleanup_user_resources(
# 2. 清理资源分配器中的相关记录
await self._cleanup_allocator_records(user_id)
-
# 3. 清理对话相关的所有Redis数据
await self._cleanup_all_chat_data(user_id, chat_id)
# 4. 清理用户级别的Redis数据
await self._cleanup_user_level_data(user_id, chat_id)
-
logger.info(
- f"[_thorough_cleanup_user_resources] 彻底清理完成: "
+ f"[_thorough_cleanup_u_res] 彻底清理完成: "
f"{user_id}:{chat_id}",
)
except Exception as e:
logger.error(
- f"[_thorough_cleanup_user_resources] 彻底清理用户 "
+ f"[_thorough_cleanup_u_res] 彻底清理用户 "
f"{user_id}:{chat_id} 时出错: {e}",
)
import traceback
logger.error(
- f"[_thorough_cleanup_user_resources] 错误详情: "
+ f"[_thorough_cleanup_u_res] 错误详情: "
f"{traceback.format_exc()}",
)
@@ -1529,7 +2361,7 @@ async def _cleanup_allocator_records(self, user_id: str):
# 清理PC分配器中的排队记录
try:
pc_position, pc_wait_status = (
- await self.pc_allocator.get_chat_wait_position_async(
+ await self.pc_allocator.get_chat_position(
user_id,
)
)
@@ -1546,7 +2378,7 @@ async def _cleanup_allocator_records(self, user_id: str):
# 清理手机分配器中的排队记录
try:
phone_position, phone_wait_status = (
- await self.phone_allocator.get_chat_wait_position_async(
+ await self.phone_allocator.get_chat_position(
user_id,
)
)
@@ -1729,18 +2561,39 @@ async def _execute_environment_init(
# 检查用户是否已在同一chat_id中有活跃会话
current_active_chat = await self.get_user_active_chat(user_id)
is_same_session_reactivation = current_active_chat == chat_id
+ # 判断对话是否过期，如果未过期，不需要重新激活
+ activate_flag = False
+ if current_active_chat:
+ current_time = time.time()
+ redis_heartbeat = await self.get_heartbeat(
+ user_id,
+ current_active_chat,
+ )
+ if (
+ redis_heartbeat
+ and current_time - redis_heartbeat < heartbeat_timeout
+ ):
+ is_same_session_reactivation = True
+ activate_flag = True
if is_same_session_reactivation:
# 检查是否已有设备且不需要重启
- chat_state = await self.get_chat_state(user_id, chat_id)
+ chat_id_for = chat_id
+ if activate_flag:
+ logger.info(
+ f"检测当前用户{user_id}活跃会话{current_active_chat}还未过期，无需重新激活",
+ )
+ chat_id_for = current_active_chat
+
+ chat_state = await self.get_chat_state(user_id, chat_id_for)
equipment_info = await self.get_equipment_info(
user_id,
- chat_id,
+ chat_id_for,
)
if chat_state.get("equipment") or equipment_info:
logger.info(
- f"用户 {user_id} 在 chat {chat_id} 已有设备，刷新认证信息后重用",
+ f"用户 {user_id} 在 chat {chat_id_for} 已有设备，刷新认证信息后重用",
)
# 需要刷新认证信息，特别是auth_code等一次性凭证
if equipment_info:
@@ -2672,13 +3525,17 @@ async def _init_equipment_async(
)
# 初始化PC设备
if sandbox_type == "pc_wuyin":
- # 修改为用户级别的资源分配
+ # 使用优化的资源分配方法，减少重复查询
logger.info(
f"[_init_equipment_async] 开始为用户 {user_id} 分配PC资源",
)
- desktop_id, status = await self.pc_allocator.allocate_async(
- user_id, # 使用user_id而不是chat_id
- timeout=0,
+
+ desktop_id, status, queue_info = (
+ await self.allocate_resource_with_queue_info(
+ "pc",
+ user_id,
+ timeout=0,
+ )
)
logger.info(f"启动desktop_id: {desktop_id}, status: {status}")
@@ -2706,28 +3563,30 @@ async def _init_equipment_async(
},
)
- # 处理资源排队情况
+ # 处理资源排队情况 - 使用预先获取的队列信息
if status == AllocationStatus.WAIT_TIMEOUT:
logger.info("资源排队超时")
- position = (
- await self.pc_allocator.get_chat_wait_position_async(
- user_id,
+ if queue_info:
+ raise HTTPException(
+ status_code=429,
+ detail={
+ "message": "All PC resources are currently in use",
+ "queue_position": queue_info["queue_position"],
+ "total_waiting": queue_info["total_waiting"],
+ "type": "queued",
+ },
+ )
+ else:
+ # 降级处理：如果队列信息获取失败，使用默认值
+ raise HTTPException(
+ status_code=429,
+ detail={
+ "message": "All PC resources are currently in use",
+ "queue_position": 1,
+ "total_waiting": 1,
+ "type": "queued",
+ },
)
- )[
- 0
- ] # 使用user_id
- total_waiting = (
- await self.pc_allocator.get_queue_info_async()
- )["total_waiting"]
- raise HTTPException(
- status_code=429,
- detail={
- "message": "All PC resources are currently in use",
- "queue_position": position + 1,
- "total_waiting": total_waiting,
- "type": "queued",
- },
- )
if not (
status == AllocationStatus.SUCCESS
@@ -2850,10 +3709,13 @@ async def _init_equipment_async(
# 初始化手机设备
elif sandbox_type == "phone_wuyin":
- # 修改为用户级别的资源分配
- instance_id, status = await self.phone_allocator.allocate_async(
- user_id, # 使用user_id而不是chat_id
- timeout=0,
+ # 使用优化的资源分配方法，减少重复查询
+ instance_id, status, queue_info = (
+ await self.allocate_resource_with_queue_info(
+ "phone",
+ user_id,
+ timeout=0,
+ )
)
print(f"启动 instance_id: {instance_id}, status: {status}")
@@ -2880,25 +3742,31 @@ async def _init_equipment_async(
},
)
- # 处理资源排队情况
+ # 处理资源排队情况 - 使用预先获取的队列信息
if status == AllocationStatus.WAIT_TIMEOUT:
- position = (
- await self.phone_allocator.get_chat_wait_position_async(
- user_id, # 使用user_id
- )
- )[0]
- total_waiting = (
- await self.phone_allocator.get_queue_info_async()
- )["total_waiting"]
- raise HTTPException(
- status_code=429,
- detail={
- "message": "All phone resources are currently in use",
- "queue_position": position + 1,
- "total_waiting": total_waiting,
- "type": "queued",
- },
- )
+ if queue_info:
+ raise HTTPException(
+ status_code=429,
+ detail={
+ "message": "All phone resources are"
+ " currently in use",
+ "queue_position": queue_info["queue_position"],
+ "total_waiting": queue_info["total_waiting"],
+ "type": "queued",
+ },
+ )
+ else:
+ # 降级处理：如果队列信息获取失败，使用默认值
+ raise HTTPException(
+ status_code=429,
+ detail={
+ "message": "All phone resources "
+ "are currently in use",
+ "queue_position": 1,
+ "total_waiting": 1,
+ "type": "queued",
+ },
+ )
if not (
status == AllocationStatus.SUCCESS
diff --git a/demos/computer_use/advanced_version/computer_use_server/start.sh b/demos/computer_use/advanced_version/computer_use_server/start.sh
index d81c127..3f6378a 100755
--- a/demos/computer_use/advanced_version/computer_use_server/start.sh
+++ b/demos/computer_use/advanced_version/computer_use_server/start.sh
@@ -18,12 +18,6 @@ BLUE=$(printf '\033[0;34m')
GREEN=$(printf '\033[0;32m')
NC=$(printf '\033[0m')
-#export PYTHONPATH="../../../..:../../../../src:$PYTHONPATH"
-#echo "🔧 已设置 PYTHONPATH: $PYTHONPATH"
-#
-## 启动后端服务
-#echo "🔧 启动后端服务..."
-#python3 backend.py &
# 启动后端服务
echo "🔧 启动后端服务 (http://localhost:8002)..."
export PYTHONPATH=$(pwd):$PYTHONPATH && python backend.py 2>&1 | sed "s/^/${BLUE}[FastAPI]${NC} /" &
@@ -37,10 +31,6 @@ echo "🎨 启动前端静态资源服务..."
cd static || { echo "❌ 无法进入 static 目录"; exit 1; }
python3 -m http.server 8001 --bind 127.0.0.1 &
-# 启动 Nginx 如果本地不需要，可以注释
-#echo "🌐 启动 Nginx 服务..."
-#sudo nginx
-
echo "✅ 服务已启动!"
echo "📱 访问地址: http://localhost:8001/index.html"
echo ""
diff --git a/demos/computer_use/advanced_version/computer_use_server/static/index.html b/demos/computer_use/advanced_version/computer_use_server/static/index.html
index a119718..9d75b2b 100644
--- a/demos/computer_use/advanced_version/computer_use_server/static/index.html
+++ b/demos/computer_use/advanced_version/computer_use_server/static/index.html
@@ -379,6 +379,14 @@
+
+
+
-
+
@@ -470,9 +478,6 @@
-
欢迎使用Computer Use Agent！
+
@@ -635,6 +643,8 @@ 等待Sandbox启动
let countdownInterval = null;
let userId = 'user_a'; // 默认用户ID
let chatId = 'chat_a'; // 默认对话ID
+ let isTokenValidated = false; // Token验证状态
+ let tempUserId = null; // 临时用户ID，避免验证期间冲突
// 从URL参数获取userId和chatId
function getUserAndChatIdFromUrl() {
@@ -652,17 +662,21 @@ 等待Sandbox启动
if (isLocal) {
// 本地环境直接返回本地用户和对话ID
console.log("检测为本地环境，使用本地用户和对话ID");
- return { userId: 'local_user', chatId: 'local_chat' };
+ isTokenValidated = true; // 本地环境直接标记为已验证
+ const chatId = 'chat_' + Math.random().toString(36).substr(2, 9);
+ return { userId: 'local_user', chatId: chatId };
} else {
// 远程环境保持原有逻辑
console.log("检测为远程环境，使用远程认证");
const urlParams = new URLSearchParams(window.location.search);
const studioToken = urlParams.get('studio_token');
if (studioToken) {
+ // 生成临时唯一用户ID，避免验证期间冲突
+ tempUserId = 'temp_' + Date.now() + '_' + Math.random().toString(36).substr(2, 9);
// 调用后端API验证token并获取userId，chatId随机生成
const chatId = 'chat_' + Math.random().toString(36).substr(2, 9);
validateStudioToken(studioToken, chatId);
- return { userId: 'user_a', chatId: chatId }; // 先返回默认值，后续异步更新userId
+ return { userId: tempUserId, chatId: chatId }; // 先返回临时值，后续异步更新userId
}
// 如果没有提供studio_token，返回null表示未登录
return null;
@@ -671,6 +685,8 @@ 等待Sandbox启动
// 验证studio token并获取userId
async function validateStudioToken(studioToken, generatedChatId) {
+ showTokenValidationUI(true); // 显示验证中UI
+
try {
// 使用后端代理避免 CORS 问题
const response = await fetch(
@@ -682,18 +698,26 @@ 等待Sandbox启动
if (data.Success && data.Data && data.Data.Uuid) {
userId = data.Data.Uuid; // 更新userId
chatId = generatedChatId; // 使用生成的chatId
+ isTokenValidated = true; // 标记Token验证完成
userStateManager.setCurrentUser(chatId);
console.log(`用户验证成功: userId=${userId}, chatId=${chatId}`);
+
+ showTokenValidationUI(false); // 隐藏验证中UI
+ enableUserOperations(); // 启用用户操作
+
return { userId, chatId };
} else {
showErrorMessage("Token验证失败，请检查您的Studio Token");
+ showTokenValidationUI(false);
}
} else {
showErrorMessage("Token验证失败，请检查您的Studio Token");
+ showTokenValidationUI(false);
}
} catch (error) {
console.error("检查studio token时出错:", error);
showErrorMessage("Token验证出错，请稍后重试");
+ showTokenValidationUI(false);
}
return null;
}
@@ -716,8 +740,70 @@ 等待Sandbox启动
userStateManager.setCurrentUser(chatId);
setupEventListeners();
setupConfigToggle(); // 添加配置区域折叠功能
+
+ // 如果Token尚未验证完成，禁用用户操作
+ if (!isTokenValidated) {
+ disableUserOperations();
+ }
});
+ // 显示/隐藏Token验证UI
+ function showTokenValidationUI(show) {
+ const validationContainer = document.getElementById('tokenValidationContainer');
+ if (validationContainer) {
+ validationContainer.style.display = show ? 'block' : 'none';
+ }
+ }
+
+ // 禁用用户操作（Token验证期间）
+ function disableUserOperations() {
+ // 禁用激活环境按钮
+ const activateBtn = document.getElementById('activateBtn');
+ if (activateBtn) {
+ activateBtn.disabled = true;
+ activateBtn.innerHTML = ' 等待身份验证...';
+ }
+
+ // 禁用输入框和发送按钮
+ const taskInput = document.getElementById('taskInput');
+ const sendBtn = document.getElementById('sendBtn');
+ if (taskInput) {
+ taskInput.disabled = true;
+ taskInput.placeholder = '请等待身份验证完成...';
+ }
+ if (sendBtn) {
+ sendBtn.disabled = true;
+ }
+
+ // 禁用配置相关的输入
+ const configInputs = document.querySelectorAll('#configCardBody input, #configCardBody select');
+ configInputs.forEach(input => {
+ input.disabled = true;
+ });
+ }
+
+ // 启用用户操作（Token验证完成后）
+ function enableUserOperations() {
+ // 启用激活环境按钮
+ const activateBtn = document.getElementById('activateBtn');
+ if (activateBtn) {
+ activateBtn.disabled = false;
+ activateBtn.innerHTML = ' 激活环境';
+ }
+
+ // 保持输入框和发送按钮禁用状态（需要先激活环境）
+ const taskInput = document.getElementById('taskInput');
+ if (taskInput) {
+ taskInput.placeholder = '请先激活环境后再输入任务...';
+ }
+
+ // 启用配置相关的输入
+ const configInputs = document.querySelectorAll('#configCardBody input, #configCardBody select');
+ configInputs.forEach(input => {
+ input.disabled = false;
+ });
+ }
+
// 显示需要登录的提示信息
function showLoginRequiredMessage() {
// 禁用所有操作按钮
@@ -3439,7 +3525,7 @@ 欢迎使用Computer Use Agent！
'帮我在抖音搜一下济南游玩攻略相关的视频，并播放收藏',
'帮我在通义APP中询问济南的两天旅游攻略'];
const pcPresets = ['在edge浏览器中搜索阿里巴巴的股价，然后在WPS中新建一个表格，在第一列填入公司名，在第二列填入股价',
- '新建一个空白PPT，然后在第一张幻灯片中插入一段艺术字形式的文本，内容为阿里巴巴',
+ '在WPS新建一个空白PPT，然后在第一张幻灯片中插入一段艺术字形式的文本，内容为阿里巴巴',
'进入bilibili，查看雷军的视频，然后点赞第一个视频',
"帮我在 WPS Office 文档中写一段阿里巴巴的简介，并把字体颜色改为蓝色",
"将系统颜色设置为浅色模式"];
@@ -3575,7 +3661,6 @@ 等待Sandbox启动
// 本地开发环境
return currentOrigin.replace(':8001', ':8002');
-
}
// 获取基础URL（静态资源URL）
diff --git a/demos/computer_use/agents/gui_agent_app_v2.py b/demos/computer_use/agents/gui_agent_app_v2.py
index 4d4b668..e18b420 100644
--- a/demos/computer_use/agents/gui_agent_app_v2.py
+++ b/demos/computer_use/agents/gui_agent_app_v2.py
@@ -30,7 +30,7 @@ class GuiAgentResponse(BaseModel):
thought: str
explanation: str
action: str
- action_params: dict
+ action_params: str
operation: str
diff --git a/demos/computer_use/base_version/computer_use_server/README_en.md b/demos/computer_use/base_version/computer_use_server/README_en.md
index 36bd529..c42dd2b 100644
--- a/demos/computer_use/base_version/computer_use_server/README_en.md
+++ b/demos/computer_use/base_version/computer_use_server/README_en.md
@@ -102,8 +102,8 @@ nano ~/.zshrc # If you are using zsh (default in macOS Catalina and later)
nano ~/.bash_profile # If you are using bash
# Add environment variables for example
-export DASHSCOPE_API_KEY=""
-export ECD_DESKTOP_ID="your_desktop_id"
+export DASHSCOPE_API_KEY=
+export ECD_DESKTOP_ID=
export EDS_OSS_ACCESS_KEY_ID=
export EDS_OSS_ACCESS_KEY_SECRET=
export EDS_OSS_BUCKET_NAME=
diff --git a/demos/computer_use/base_version/computer_use_server/README_zh.md b/demos/computer_use/base_version/computer_use_server/README_zh.md
index 6bb59fa..e944096 100644
--- a/demos/computer_use/base_version/computer_use_server/README_zh.md
+++ b/demos/computer_use/base_version/computer_use_server/README_zh.md
@@ -95,8 +95,8 @@ nano ~/.zshrc # 如果你用的是 zsh（macOS Catalina 及以后默认）
nano ~/.bash_profile # 如果你用的是 bash
# 添加环境变量例如
-export DASHSCOPE_API_KEY=""
-export ECD_DESKTOP_ID="your_desktop_id"
+export DASHSCOPE_API_KEY=
+export ECD_DESKTOP_ID=
export EDS_OSS_ACCESS_KEY_ID=
export EDS_OSS_ACCESS_KEY_SECRET=
export EDS_OSS_BUCKET_NAME=
diff --git a/demos/computer_use/sandbox_center/sandboxes/cloud_computer_wy.py b/demos/computer_use/sandbox_center/sandboxes/cloud_computer_wy.py
index 551a178..e6bea5c 100644
--- a/demos/computer_use/sandbox_center/sandboxes/cloud_computer_wy.py
+++ b/demos/computer_use/sandbox_center/sandboxes/cloud_computer_wy.py
@@ -1056,13 +1056,8 @@ def contains_chinese(text):
# 打开 Windows 搜索栏
pyautogui.press('win') # 按下 Win 键
time.sleep(0.5)
-pyautogui.typewrite(name) # 直接输入应用名称
-
-# 如果包含中文，则使用粘贴方式（备用方案）
-if contains_chinese(name):
- pyperclip.copy(name)
- pyautogui.hotkey(ctrl_key, 'v') # 使用 Ctrl+V 粘贴
-
+pyperclip.copy(name)
+pyautogui.hotkey(ctrl_key, 'v') # 使用 Ctrl+V 粘贴
# 回车确认
time.sleep(1)
pyautogui.press('enter')