推进 ActionExecutor 相关的动态插拔式行动调度机制

- 移除先前构想的 SpecializedPartnerInputData 及相关类，无论是自反思、向用户求助还是用户主动干预，都应当通过语义识别来作用于对应行动事件，使用固定行动id的机制不足以支撑这种机制
- 在 ActionCore 中新增执行中行动的 phaser 管理逻辑
- 新增几个异常类，适用于行动数据加载的异常情况
- 新增 ActionIdentifier 负责行动干预意图的识别
-

Partner-Main/src/main/java/work/slhaf/partner/core/action/ActionCapability.java

@@ -1,12 +1,15 @@
 package work.slhaf.partner.core.action;
 
+import lombok.NonNull;
 import work.slhaf.partner.api.agent.factory.capability.annotation.Capability;
 import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.action.entity.MetaAction;
 import work.slhaf.partner.core.action.entity.cache.CacheAdjustData;
 
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Phaser;
 
 @Capability(value = "action")
 public interface ActionCapability {
@@ -27,4 +30,13 @@ public interface ActionCapability {
     ExecutorService getExecutor(ActionCore.ExecutorType type);
 
     Set<String> getExistedMetaActions();
+
+    void putPhaserRecord(Phaser phaser, ActionData actionData);
+
+    void removePhaserRecord(Phaser phaser);
+
+    ActionCore.PhaserRecord getPhaserRecord(String tendency, String source);
+
+    MetaAction loadMetaAction(@NonNull String actionKey);
+
 }

Partner-Main/src/main/java/work/slhaf/partner/core/action/ActionCore.java

@@ -1,20 +1,26 @@
 package work.slhaf.partner.core.action;
 
+import cn.hutool.core.bean.BeanUtil;
+import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import work.slhaf.partner.api.agent.factory.capability.annotation.CapabilityCore;
 import work.slhaf.partner.api.agent.factory.capability.annotation.CapabilityMethod;
 import work.slhaf.partner.common.vector.VectorClient;
 import work.slhaf.partner.core.PartnerCore;
 import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.action.entity.MetaAction;
 import work.slhaf.partner.core.action.entity.MetaActionInfo;
 import work.slhaf.partner.core.action.entity.cache.ActionCacheData;
 import work.slhaf.partner.core.action.entity.cache.CacheAdjustData;
 import work.slhaf.partner.core.action.entity.cache.CacheAdjustMetaData;
+import work.slhaf.partner.core.action.exception.ActionDataNotFoundException;
+import work.slhaf.partner.core.action.exception.MetaActionNotFoundException;
 
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Phaser;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
@@ -44,7 +50,11 @@ public class ActionCore extends PartnerCore<ActionCore> {
     private final ExecutorService platformExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
     private final ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
 
+    /**
+     * 已存在的行动程序，键为目录名，值为从目录加载的行动程序元信息
+     */
     private final LinkedHashMap<String, MetaActionInfo> existedMetaActions = new LinkedHashMap<>();
+    private final List<PhaserRecord> phaserRecords = new ArrayList<>();
 
     public ActionCore() throws IOException, ClassNotFoundException {
         new ActionWatchService(existedMetaActions, virtualExecutor).launch();
@@ -67,7 +77,6 @@ public class ActionCore extends PartnerCore<ActionCore> {
                 .collect(Collectors.toList());
     }
 
-
     @CapabilityMethod
     public synchronized void putPendingActions(String userId, ActionData actionData) {
         pendingActions.computeIfAbsent(userId, k -> {
@@ -171,6 +180,48 @@ public class ActionCore extends PartnerCore<ActionCore> {
         return existedMetaActions.keySet();
     }
 
+    @CapabilityMethod
+    public synchronized void putPhaserRecord(Phaser phaser, ActionData actionData) {
+        phaserRecords.add(new PhaserRecord(phaser, actionData));
+    }
+
+    @CapabilityMethod
+    public synchronized void removePhaserRecord(Phaser phaser) {
+        PhaserRecord remove = null;
+        for (PhaserRecord record : phaserRecords) {
+            if (record.phaser.equals(phaser)) {
+                remove = record;
+            }
+        }
+
+        if (remove != null) {
+            phaserRecords.remove(remove);
+        }
+    }
+
+    @CapabilityMethod
+    public PhaserRecord getPhaserRecord(String tendency, String source) {
+        for (PhaserRecord record : phaserRecords) {
+            ActionData data = record.actionData;
+            if (data.getTendency().equals(tendency) && data.getSource().equals(source)) {
+                return record;
+            }
+        }
+        throw new ActionDataNotFoundException("未找到对应的 Phaser 记录: tendency=" + tendency + ", source=" + source);
+    }
+
+    @CapabilityMethod
+    public MetaAction loadMetaAction(@NonNull String actionKey) {
+        for (MetaActionInfo actionInfo : existedMetaActions.values()) {
+            if (actionInfo.getKey().equals(actionKey)) {
+                MetaAction metaAction = new MetaAction();
+                BeanUtil.copyProperties(actionInfo, metaAction);
+                return metaAction;
+            }
+        }
+        throw new MetaActionNotFoundException("未找到对应的行动程序信息" + actionKey);
+    }
+
     /**
      * 命中缓存且评估通过时
      *
@@ -268,4 +319,7 @@ public class ActionCore extends PartnerCore<ActionCore> {
     public enum ExecutorType {
         VIRTUAL, PLATFORM
     }
+
+    public record PhaserRecord(Phaser phaser, ActionData actionData) {
+    }
 }

Partner-Main/src/main/java/work/slhaf/partner/core/action/ActionWatchService.java

@@ -21,9 +21,9 @@ import static work.slhaf.partner.common.Constant.Path.ACTION_PROGRAM;
 @Slf4j
 class ActionWatchService {
 
-    private HashMap<Path, WatchKey> registeredPaths = new HashMap<>();
+    private final HashMap<Path, WatchKey> registeredPaths = new HashMap<>();
-    private LinkedHashMap<String, MetaActionInfo> existedMetaActions;
+    private final LinkedHashMap<String, MetaActionInfo> existedMetaActions;
-    private ExecutorService virtualExecutor;
+    private final ExecutorService virtualExecutor;
 
     public ActionWatchService(LinkedHashMap<String, MetaActionInfo> existedMetaActions, ExecutorService virtualExecutor) {
         this.existedMetaActions = existedMetaActions;
@@ -114,7 +114,7 @@ class ActionWatchService {
     private boolean checkComplete(Path thisDir) {
         File[] files = thisDir.toFile().listFiles();
         if (files == null) {
-            log.error("[{}]当前目录无法访问: [{}]", thisDir);
+            log.error("当前目录无法访问: [{}]", thisDir);
             return false;
         }
         boolean existedAction = false;

Partner-Main/src/main/java/work/slhaf/partner/core/action/entity/ActionData.java

@@ -17,6 +17,7 @@ public abstract class ActionData {
     protected String result;
     protected String reason;
     protected String description;
+    protected String source;
 
     public enum ActionStatus {
         SUCCESS, FAILED, EXECUTING, WAITING, PREPARE

Partner-Main/src/main/java/work/slhaf/partner/core/action/entity/MetaAction.java

@@ -26,7 +26,7 @@ public class MetaAction implements Comparable<MetaAction>, Runnable {
     /**
      * 行动结果，包括执行状态和相应内容(执行结果或者错误信息)
      */
-    private Result result = new Result();
+    private final Result result = new Result();
     /**
      * 执行顺序，升序排列
      */
@@ -49,7 +49,6 @@ public class MetaAction implements Comparable<MetaAction>, Runnable {
     public void run() {
         File action = loadFromFile();
         if (!action.exists()) {
-            result = new Result();
             result.setSuccess(false);
             result.setData("Action file not found: " + action.getAbsolutePath());
         }
@@ -60,7 +59,6 @@ public class MetaAction implements Comparable<MetaAction>, Runnable {
                 case SCRIPT -> executeScript(action);
             }
         } catch (Exception e) {
-            result = new Result();
             result.setSuccess(false);
             result.setData(e.getMessage());
         }
@@ -89,7 +87,7 @@ public class MetaAction implements Comparable<MetaAction>, Runnable {
     @Data
     public static class Result {
         private boolean success = true;
-        private String data;
+        private String data = null;
     }
 
 }

Partner-Main/src/main/java/work/slhaf/partner/core/action/exception/ActionDataNotFoundException.java

@@ -0,0 +1,13 @@
+package work.slhaf.partner.core.action.exception;
+
+import work.slhaf.partner.api.agent.runtime.exception.AgentRuntimeException;
+
+public class ActionDataNotFoundException extends AgentRuntimeException {
+    public ActionDataNotFoundException(String message) {
+        super(message);
+    }
+
+    public ActionDataNotFoundException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

Partner-Main/src/main/java/work/slhaf/partner/core/action/exception/MetaActionNotFoundException.java

@@ -0,0 +1,13 @@
+package work.slhaf.partner.core.action.exception;
+
+import work.slhaf.partner.api.agent.runtime.exception.AgentRuntimeException;
+
+public class MetaActionNotFoundException extends AgentRuntimeException {
+    public MetaActionNotFoundException(String message) {
+        super(message);
+    }
+
+    public MetaActionNotFoundException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

Partner-Main/src/main/java/work/slhaf/partner/core/memory/MemoryCore.java

@@ -488,7 +488,7 @@ public class MemoryCore extends PartnerCore<MemoryCore> {
         return targetParentNode;
     }
 
-    public void updateCacheCounter(List<String> topicPath) {
+    private void updateCacheCounter(List<String> topicPath) {
         ConcurrentHashMap<List<String>, Integer> memoryNodeCacheCounter = cache.memoryNodeCacheCounter;
         if (memoryNodeCacheCounter.containsKey(topicPath)) {
             Integer tempCount = memoryNodeCacheCounter.get(topicPath);

Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionExecutor.java

@@ -12,7 +12,9 @@ import work.slhaf.partner.core.action.entity.ActionData;
 import work.slhaf.partner.core.action.entity.ImmediateActionData;
 import work.slhaf.partner.core.action.entity.MetaAction;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Phaser;
 
@@ -26,8 +28,6 @@ public class ActionExecutor extends AgentRunningSubModule<List<ImmediateActionDa
     private ExecutorService virtualExecutor;
     private ExecutorService platformExecutor;
 
-    private HashMap<String, PhaserActionChain> phaserRecorder = new HashMap<>();
-
     @Init
     public void init() {
         virtualExecutor = actionCapability.getExecutor(ActionCore.ExecutorType.VIRTUAL);
@@ -48,6 +48,9 @@ public class ActionExecutor extends AgentRunningSubModule<List<ImmediateActionDa
             LinkedHashMap<Integer, List<MetaAction>> actionChain = actionData.getActionChain();
             List<MetaAction> virtual = new ArrayList<>();
             List<MetaAction> platform = new ArrayList<>();
+            Phaser phaser = new Phaser();
+            phaser.register();
+            actionCapability.putPhaserRecord(phaser, actionData);
             actionChain.forEach((k, v) -> {
                 for (MetaAction metaAction : v) {
                     // 根据io类型放入合适的列表
@@ -57,42 +60,44 @@ public class ActionExecutor extends AgentRunningSubModule<List<ImmediateActionDa
                         platform.add(metaAction);
                     }
                 }
-                runGroupAction(virtual, platform, actionChain);
+                runGroupAction(virtual, platform, actionChain, phaser);
+                virtual.clear();
+                platform.clear();
+                phaser.arriveAndAwaitAdvance();
             });
+            actionCapability.removePhaserRecord(phaser);
         });
+
     }
 
     // 使用phaser来承担同组的动态任务新增
-    private void runGroupAction(List<MetaAction> virtual, List<MetaAction> platform, LinkedHashMap<Integer, List<MetaAction>> actionChain) {
+    private void runGroupAction(List<MetaAction> virtual, List<MetaAction> platform, LinkedHashMap<Integer, List<MetaAction>> actionChain, Phaser phaser) {
-        Phaser phaser = new Phaser();
+        runGroupAction(virtual, virtualExecutor, phaser);
-        phaser.register();
+        runGroupAction(platform, platformExecutor, phaser);
-        String groupId = UUID.randomUUID().toString();
-        phaserRecorder.put(groupId, new PhaserActionChain(phaser, actionChain));
-        runGroupAction(virtual, virtualExecutor, actionChain, phaser);
-        runGroupAction(platform, platformExecutor, actionChain, phaser);
-        phaserRecorder.remove(groupId);
-        phaser.arriveAndAwaitAdvance();
     }
 
-    private void runGroupAction(List<MetaAction> actions, ExecutorService executor, LinkedHashMap<Integer, List<MetaAction>> actionChain, Phaser phaser) {
+    private void runGroupAction(List<MetaAction> actions, ExecutorService executor, Phaser phaser) {
         for (MetaAction action : actions) {
             phaser.register();
             executor.execute(() -> {
-                MetaAction.Result result = action.getResult();
+                try {
-                do {
+                    MetaAction.Result result = action.getResult();
-                    // 该循环对应LLM的调整参数后重试
+                    do {
-                    if (!result.isSuccess()) {
+                        // 该循环对应LLM的调整参数后重试
-                        //TODO LLM决策是重构参数、执行自对话反思、还是选择向用户求助(通过cognationCore暴露方法，可能需要修改其他模块以进行适应)，仅重构参数时无需结束当前循环
+                        if (!result.isSuccess()) {
-                        // 若使用Phaser作为执行线程与反思、求助等调用流程的同步协调，应当需要额外维护Phaser全局字段，获取到反思结果或者用户反馈后，
+                            //TODO LLM决策是重构参数、执行自对话反思、还是选择向用户求助(通过cognationCore暴露方法，可能需要修改其他模块以进行适应)，仅重构参数时无需结束当前循环
-                        // 调用对应的phaser注册任务，在ActionExecutor中动态添加任务至actionChain,同时启动异步执行
+                            // 若使用Phaser作为执行线程与反思、求助等调用流程的同步协调，应当需要额外维护Phaser全局字段，获取到反思结果或者用户反馈后，
-                        // 而且由于执行与放入的为同一个MetaAction对象，所以执行结果可被当前行动链获取，但virtual、executor两个列表似乎不行，需要重构执行模式，建议将行动链直接重构为LinkedHashMap，order为键
+                            // 调用对应的phaser注册任务，在ActionExecutor中动态添加任务至actionChain,同时启动异步执行
-                        String input = getInput(result.getData());
+                            // 而且由于执行与放入的为同一个MetaAction对象，所以执行结果可被当前行动链获取，但virtual、executor两个列表似乎不行，需要重构执行模式，建议将行动链直接重构为LinkedHashMap，order为键
+                            String input = getInput(result.getData());
 
-                    }
+                        }
-                    action.run();
+                        action.run();
-                } while (!result.isSuccess());
+                    } while (!result.isSuccess());
-                //TODO 将执行结果写入特定对话角色记忆(cognationCore暴露方法)
+                    //TODO 将执行结果写入特定对话角色记忆(cognationCore暴露方法)
-                phaser.arriveAndDeregister();
+                } finally {
+                    phaser.arriveAndDeregister();
+                }
             });
         }
     }
@@ -112,6 +117,4 @@ public class ActionExecutor extends AgentRunningSubModule<List<ImmediateActionDa
         return false;
     }
 
-    private record PhaserActionChain(Phaser phaser, LinkedHashMap<Integer, List<MetaAction>> actionChain) {
-    }
 }

Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/identifier/ActionIdentifier.java

@@ -0,0 +1,40 @@
+package work.slhaf.partner.module.modules.action.identifier;
+
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.module.common.module.PreRunningModule;
+import work.slhaf.partner.runtime.interaction.data.context.PartnerRunningFlowContext;
+
+import java.util.HashMap;
+
+/**
+ * 负责识别潜在的行动干预信息，作用于正在进行或已存在的行动池中内容
+ */
+@AgentModule(name = "action_identifier", order = 2)
+public class ActionIdentifier extends PreRunningModule implements ActivateModel {
+
+    @Override
+    protected void doExecute(PartnerRunningFlowContext context) {
+
+    }
+
+    @Override
+    public String modelKey() {
+        return "";
+    }
+
+    @Override
+    public boolean withBasicPrompt() {
+        return false;
+    }
+
+    @Override
+    protected HashMap<String, String> getPromptDataMap(PartnerRunningFlowContext context) {
+        return null;
+    }
+
+    @Override
+    protected String moduleName() {
+        return "";
+    }
+}

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/PartnerInteractionAdapter.java

@@ -3,7 +3,6 @@ package work.slhaf.partner.runtime.interaction;
 import work.slhaf.partner.api.agent.runtime.interaction.AgentInteractionAdapter;
 import work.slhaf.partner.runtime.interaction.data.PartnerInputData;
 import work.slhaf.partner.runtime.interaction.data.PartnerOutputData;
-import work.slhaf.partner.runtime.interaction.data.SpecializedPartnerInputData;
 import work.slhaf.partner.runtime.interaction.data.context.PartnerRunningFlowContext;
 
 public class PartnerInteractionAdapter extends AgentInteractionAdapter<PartnerInputData, PartnerOutputData, PartnerRunningFlowContext> {
@@ -34,10 +33,6 @@ public class PartnerInteractionAdapter extends AgentInteractionAdapter<PartnerIn
         context.setSingle(inputData.isSingle());
         context.setPlatform(inputData.getPlatform());
         context.setInput(inputData.getContent());
-        context.setType(inputData.getInputType());
-        if (inputData instanceof SpecializedPartnerInputData specializedData) {
-            context.setPayload(specializedData.getPayload());
-        }
         return context;
     }
 }

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/WebSocketGateway.java

@@ -13,7 +13,8 @@ import work.slhaf.partner.api.agent.runtime.interaction.AgentGateway;
 import work.slhaf.partner.api.agent.runtime.interaction.AgentInteractionAdapter;
 import work.slhaf.partner.common.config.PartnerAgentConfigManager;
 import work.slhaf.partner.common.thread.InteractionThreadPoolExecutor;
-import work.slhaf.partner.runtime.interaction.data.*;
+import work.slhaf.partner.runtime.interaction.data.PartnerInputData;
+import work.slhaf.partner.runtime.interaction.data.PartnerOutputData;
 import work.slhaf.partner.runtime.interaction.data.context.PartnerRunningFlowContext;
 
 import java.net.InetSocketAddress;
@@ -140,12 +141,7 @@ public class WebSocketGateway extends WebSocketServer implements AgentGateway<Pa
 
     @Override
     public void onMessage(WebSocket webSocket, String s) {
-        JSONObject parsedObject = JSONObject.parseObject(s);
+        PartnerInputData inputData = JSONObject.parseObject(s, PartnerInputData.class);
-        PartnerInputType inputType = parsedObject.getObject(SpecializedPayloadConstant.TYPE, PartnerInputType.class);
-        PartnerInputData inputData = switch (inputType) {
-            case NORMAL -> parsedObject.to(PartnerInputData.class);
-            case SYSTEM, ASSIST_REQUEST, REFLECTION -> parsedObject.to(SpecializedPartnerInputData.class);
-        };
         userSessions.put(inputData.getUserInfo(), webSocket); // 注册连接
         receive(inputData);
     }

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/data/PartnerInputData.java

@@ -10,5 +10,4 @@ public class PartnerInputData extends AgentInputData {
     protected String userNickName;
     protected String platform;
     protected boolean single;
-    protected PartnerInputType inputType;
 }

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/data/PartnerInputType.java

@@ -1,5 +0,0 @@
-package work.slhaf.partner.runtime.interaction.data;
-
-public enum PartnerInputType {
-    NORMAL, REFLECTION, ASSIST_REQUEST, SYSTEM
-}

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/data/SpecializedPartnerInputData.java

@@ -1,12 +0,0 @@
-package work.slhaf.partner.runtime.interaction.data;
-
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-import java.util.Map;
-
-@EqualsAndHashCode(callSuper = true)
-@Data
-public class SpecializedPartnerInputData extends PartnerInputData {
-    protected Map<String, String> payload;
-}

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/data/SpecializedPayloadConstant.java

@@ -1,7 +0,0 @@
-package work.slhaf.partner.runtime.interaction.data;
-
-public class SpecializedPayloadConstant {
-    public static final String TASK_ID = "taskId";
-    public static final String ACTION_ID = "actionId";
-    public static final String TYPE = "inputType";
-}

Partner-Main/src/main/java/work/slhaf/partner/runtime/interaction/data/context/PartnerRunningFlowContext.java

@@ -5,7 +5,6 @@ import lombok.Data;
 import lombok.EqualsAndHashCode;
 import work.slhaf.partner.api.agent.runtime.interaction.flow.entity.RunningFlowContext;
 import work.slhaf.partner.module.common.entity.AppendPromptData;
-import work.slhaf.partner.runtime.interaction.data.PartnerInputType;
 import work.slhaf.partner.runtime.interaction.data.context.subcontext.CoreContext;
 import work.slhaf.partner.runtime.interaction.data.context.subcontext.ModuleContext;
 
@@ -13,7 +12,6 @@ import java.io.Serial;
 import java.time.LocalDateTime;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.UUID;
 
 @EqualsAndHashCode(callSuper = true)
@@ -32,9 +30,6 @@ public class PartnerRunningFlowContext extends RunningFlowContext {
     protected LocalDateTime dateTime;
     protected boolean single;
 
-    protected PartnerInputType type;
-    protected Map<String, String> payload;
-
     protected String input;
 
     protected CoreContext coreContext = new CoreContext();