feat:用户修改步骤后重新执行实现

backend/AgentCoord/RehearsalEngine_V2/ExecutePlan_Optimized.py

@@ -428,7 +428,7 @@ def executePlan_streaming_dynamic(
             if execution_id:
                 # 动态模式：循环获取下一个步骤
                 # 等待新步骤的最大次数（避免无限等待）
-                max_empty_wait_cycles = 60  # 最多等待60次，每次等待1秒
+                max_empty_wait_cycles = 5  # 最多等待60次，每次等待1秒
                 empty_wait_count = 0
 
                 while True:

backend/server.py

@@ -18,6 +18,7 @@ from AgentCoord.PlanEngine.AgentSelectModify import (
 import os
 import yaml
 import argparse
+from typing import List, Dict
 
 # initialize global variables
 yaml_file = os.path.join(os.getcwd(), "config", "config.yaml")
@@ -39,6 +40,40 @@ app.config['SECRET_KEY'] = 'agentcoord-secret-key'
 socketio = SocketIO(app, cors_allowed_origins="*", async_mode='threading')
 
 
+def truncate_rehearsal_log(RehearsalLog: List, restart_from_step_index: int) -> List:
+    """
+    截断 RehearsalLog，只保留指定索引之前的步骤结果
+
+    Args:
+        RehearsalLog: 原始日志列表
+        restart_from_step_index: 重新执行的起始步骤索引（例如：1 表示保留步骤0，从步骤1重新执行）
+
+    Returns:
+        截断后的 RehearsalLog
+
+    示例:
+        restart_from_step_index = 1
+        RehearsalLog = [step0, object0, step1, object1, step2, object2]
+        返回 = [step0, object0]  # 只保留步骤0的结果
+    """
+    truncated_log = []
+    step_count = 0
+
+    for logNode in RehearsalLog:
+        if logNode.get("LogNodeType") == "step":
+            # 只保留 restart_from_step_index 之前的步骤
+            if step_count < restart_from_step_index:
+                truncated_log.append(logNode)
+            step_count += 1
+        elif logNode.get("LogNodeType") == "object":
+            # object 节点：如果对应的 step 在保留范围内，保留它
+            # 策略：保留所有在截断点之前的 object
+            if step_count <= restart_from_step_index:
+                truncated_log.append(logNode)
+
+    return truncated_log
+
+
 @app.route("/fill_stepTask_TaskProcess", methods=["post"])
 def Handle_fill_stepTask_TaskProcess():
     incoming_data = request.get_json()
@@ -412,7 +447,7 @@ def handle_ping():
 def handle_execute_plan_optimized_ws(data):
     """
     WebSocket版本：优化版流式执行计划
-    支持步骤级流式 + 动作级智能并行 + 动态追加步骤
+    支持步骤级流式 + 动作级智能并行 + 动态追加步骤 + 从指定步骤重新执行
 
     请求格式：
     {
@@ -422,7 +457,8 @@ def handle_execute_plan_optimized_ws(data):
             "plan": {...},
             "num_StepToRun": null,
             "RehearsalLog": [],
-            "enable_dynamic": true  # 是否启用动态追加步骤
+            "enable_dynamic": true,  # 是否启用动态追加步骤
+            "restart_from_step_index": 1  # 可选：从指定步骤重新执行（例如1表示从步骤2重新执行）
         }
     }
     """
@@ -434,6 +470,13 @@ def handle_execute_plan_optimized_ws(data):
         num_StepToRun = incoming_data.get("num_StepToRun")
         RehearsalLog = incoming_data.get("RehearsalLog", [])
         enable_dynamic = incoming_data.get("enable_dynamic", False)
+        restart_from_step_index = incoming_data.get("restart_from_step_index")  # 新增：支持从指定步骤重新执行
+
+        # 如果指定了重新执行起始步骤，截断 RehearsalLog
+        if restart_from_step_index is not None:
+            print(f"🔄 从步骤 {restart_from_step_index + 1} 重新执行，正在截断 RehearsalLog...")
+            RehearsalLog = truncate_rehearsal_log(RehearsalLog, restart_from_step_index)
+            print(f"✅ RehearsalLog 已截断，保留 {sum(1 for node in RehearsalLog if node.get('LogNodeType') == 'step')} 个步骤的结果")
 
         # 如果前端传入了execution_id，使用前端的；否则生成新的
         execution_id = incoming_data.get("execution_id")
@@ -1650,6 +1693,66 @@ def handle_resume_execution(data):
         })
 
 
+@socketio.on('stop_execution')
+def handle_stop_execution(data):
+    """
+    WebSocket版本：停止任务执行
+
+    请求格式：
+    {
+        "id": "request-id",
+        "action": "stop_execution",
+        "data": {
+            "goal": "任务描述"
+        }
+    }
+    """
+    request_id = data.get('id')
+    incoming_data = data.get('data', {})
+
+    try:
+        from AgentCoord.RehearsalEngine_V2.execution_state import execution_state_manager
+
+        goal = incoming_data.get('goal', '')
+
+        # 检查当前执行的任务是否匹配
+        current_goal = execution_state_manager.get_goal()
+        if current_goal and current_goal != goal:
+            print(f"⚠️  任务目标不匹配: 当前={current_goal}, 请求={goal}")
+            emit('response', {
+                'id': request_id,
+                'status': 'error',
+                'error': '任务目标不匹配'
+            })
+            return
+
+        # 调用执行状态管理器停止
+        success = execution_state_manager.stop_execution()
+
+        if success:
+            print(f"🛑 [DEBUG] 停止成功! 当前状态: {execution_state_manager.get_status().value}")
+            print(f"🛑 [DEBUG] should_stop: {execution_state_manager._should_stop}")
+            emit('response', {
+                'id': request_id,
+                'status': 'success',
+                'data': {"message": "已停止执行"}
+            })
+        else:
+            print(f"⚠️  [DEBUG] 停止失败，当前状态: {execution_state_manager.get_status().value}")
+            emit('response', {
+                'id': request_id,
+                'status': 'error',
+                'error': f'无法停止，当前状态: {execution_state_manager.get_status().value}'
+            })
+
+    except Exception as e:
+        print(f"❌ 停止执行失败: {str(e)}")
+        emit('response', {
+            'id': request_id,
+            'status': 'error',
+            'error': str(e)
+        })
+
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(