mcpClients = new HashMap<>();
+ /**
+ * 动态生成的行动程序都将挂载至该 McpServer
+ *
+ * 相关目录按照以下格式进行组织:
+ *
+ * DYNAMIC_ACTION_PATH/action 名称/
+ *
+ * 每个action子目录下，除了相关的程序文件外，将额外提供一个 .meta.json 文件来提供相关描述文件，
+ * 该描述文件将携带 McpTools、MetaActionInfo 相关的所有信息，
+ * 故 McpDescServer 将只负责 Common Mcp Servers 的额外描述文件
+ */
+ private McpStatelessAsyncServer dynamicActionMcpServer;
+ /**
+ * 负责监听常规 MCP Server 的描述文件（描述文件主要用于添加原本 MCP Tools 不携带的信息，如前置依赖、后置依赖、是否 IO 密集等
+ *
+ * 目录按照以下格式组织:
+ *
+ * MCP_DESC_PATH/server::toolName.desc.json
+ *
+ * 该 MCP Server-Client 的作用为: 与 CommonMcp Clients 配合，补齐第三方 MCP 服务的描述信息
+ */
+ private McpStatelessAsyncServer mcpDescServer;
+
+ public LocalRunnerClient(ConcurrentHashMap existedMetaActions, ExecutorService executor, @Nullable String baseActionPath) {
+ super(existedMetaActions, executor, baseActionPath);
+ this.TMP_ACTION_PATH = buildPathStr(ACTION_PATH, "tmp");
+ this.DYNAMIC_ACTION_PATH = buildPathStr(ACTION_PATH, "dynamic");
+ this.MCP_SERVER_PATH = buildPathStr(ACTION_PATH, "mcp");
+ this.MCP_DESC_PATH = buildPathStr(MCP_SERVER_PATH, "desc");
+
+ createPath(TMP_ACTION_PATH);
+ createPath(DYNAMIC_ACTION_PATH);
+ createPath(MCP_SERVER_PATH);
+ createPath(MCP_DESC_PATH);
+
+ try {
+ registerDescMcp();
+ registerDynamicActionMcp();
+ registerCommonMcp();
+ } catch (IOException e) {
+ throw new ActionInitFailedException("目录监听器启动失败", e);
+ }
+ setupShutdownHook();
+ }
+
+ private void registerCommonMcp() throws IOException {
+ val ctx = new WatchContext(Path.of(MCP_SERVER_PATH), FileSystems.getDefault().newWatchService());
+ val common = new LocalWatchEventProcessor.Common(existedMetaActions, mcpClients, mcpClients.get(MCP_NAME_DESC), ctx);
+ new LocalWatchServiceBuild.BuildRegistry(ctx)
+ .initialLoad(common.buildLoad())
+ .registerCreate(common.buildCreate())
+ .registerDelete(common.buildDelete())
+ .registerModify(common.buildModify())
+ .registerOverflow(common.buildOverflow())
+ .commit(executor);
+ log.info("CommonMcp 文件监听注册完毕");
+ }
+
+ private void registerDescMcp() throws IOException {
+ InProcessMcpTransport.Pair pair = InProcessMcpTransport.pair();
+ McpSchema.ServerCapabilities serverCapabilities = McpSchema.ServerCapabilities.builder()
+ .resources(true, true)
+ .build();
+ mcpDescServer = McpServer.async(pair.serverSide())
+ .capabilities(serverCapabilities)
+ .jsonMapper(McpJsonMapper.getDefault())
+ .build();
+ registerDescMcpWatch();
+ log.info("DescMcp 文件监听注册完毕");
+ registerMcpClient(MCP_NAME_DESC, pair.clientSide(), 10);
+ log.info("DescMcp 注册完毕");
+
+ }
+
+ private void registerDescMcpWatch() throws IOException {
+ WatchContext ctx = new WatchContext(Path.of(MCP_DESC_PATH), FileSystems.getDefault().newWatchService());
+ LocalWatchEventProcessor.Desc desc = new LocalWatchEventProcessor.Desc(existedMetaActions, mcpDescServer, ctx);
+ new LocalWatchServiceBuild.BuildRegistry(ctx)
+ .initialLoad(desc.buildLoad())
+ .registerCreate(desc.buildCreate())
+ .registerDelete(desc.buildDelete())
+ .registerModify(desc.buildModify())
+ .registerOverflow(desc.buildOverflow())
+ .watchAll(true)
+ .commit(executor);
+ }
+
+ private void registerDynamicActionMcp() throws IOException {
+ InProcessMcpTransport.Pair pair = InProcessMcpTransport.pair();
+ McpSchema.ServerCapabilities serverCapabilities = McpSchema.ServerCapabilities.builder()
+ .tools(true)
+ .build();
+ dynamicActionMcpServer = McpServer.async(pair.serverSide())
+ .capabilities(serverCapabilities)
+ .jsonMapper(McpJsonMapper.getDefault())
+ .build();
+ // Tools 的执行逻辑应当高度一致化，但仍需要独立为不同 Tool
+ // 初始的加载逻辑通过 initialLoad 加载
+ // 后续的动态更新通过对应的 event 事件触发
+ registerDynamicActionMcpWatch();
+ log.info("DynamicActionMcp 文件监听注册完毕");
+ registerMcpClient(MCP_NAME_DYNAMIC, pair.clientSide(), 10);
+ log.info("DynamicActionMcp 注册完毕");
+ }
+
+ private void registerDynamicActionMcpWatch() throws IOException {
+ // MODIFY、CREATE、DELETE、OVERFLOW 都需要不同的处理方式
+ WatchContext ctx = new WatchContext(Path.of(DYNAMIC_ACTION_PATH), FileSystems.getDefault().newWatchService());
+ LocalWatchEventProcessor.Dynamic dynamic = new LocalWatchEventProcessor.Dynamic(existedMetaActions, dynamicActionMcpServer, ctx);
+ new LocalWatchServiceBuild.BuildRegistry(ctx)
+ .initialLoad(dynamic.buildLoad())
+ .registerCreate(dynamic.buildCreate())
+ .registerModify(dynamic.buildModify())
+ .registerDelete(dynamic.buildDelete())
+ .registerOverflow(dynamic.buildOverflow())
+ .watchAll(true)
+ .commit(executor);
+ }
+
+ @Override
+ protected RunnerResponse doRun(MetaAction metaAction) {
+ log.debug("执行行动: {}", metaAction);
+ RunnerResponse response;
+ try {
+ response = switch (metaAction.getType()) {
+ case MetaAction.Type.MCP -> doRunWithMcp(metaAction);
+ case MetaAction.Type.ORIGIN -> doRunWithOrigin(metaAction);
+ };
+ } catch (Exception e) {
+ response = new RunnerResponse();
+ response.setOk(false);
+ response.setData(e.getLocalizedMessage());
+ }
+ log.debug("行动执行结果: {}", response);
+ return response;
+ }
+
+ private RunnerResponse doRunWithOrigin(MetaAction metaAction) {
+ RunnerResponse response = new RunnerResponse();
+ File file = new File(metaAction.getLocation());
+ String ext = FileUtil.getSuffix(file);
+ if (ext == null || ext.isEmpty()) {
+ response.setOk(false);
+ response.setData("未知文件类型");
+ return response;
+ }
+ String[] commands = SystemExecHelper.buildCommands(ext, metaAction.getParams(), file.getAbsolutePath());
+ if (commands == null || commands.length == 0) {
+ response.setOk(false);
+ response.setData("不支持的文件类型: " + file.getName());
+ }
+ SystemExecHelper.Result execResult = SystemExecHelper.exec(commands);
+ response.setOk(execResult.isOk());
+ response.setData(execResult.getTotal());
+ return response;
+ }
+
+ private RunnerResponse doRunWithMcp(MetaAction metaAction) {
+ RunnerResponse response = new RunnerResponse();
+ McpSyncClient mcpClient = mcpClients.get(metaAction.getLocation());
+ McpSchema.CallToolRequest callToolRequest = McpSchema.CallToolRequest.builder()
+ .name(metaAction.getName())
+ .arguments(metaAction.getParams())
+ .build();
+ McpSchema.CallToolResult callToolResult = mcpClient.callTool(callToolRequest);
+ val callToolResultError = callToolResult.isError();
+ if (callToolResultError == null) {
+ response.setOk(false);
+ } else {
+ response.setOk(!callToolResultError);
+ }
+ response.setData(callToolResult.structuredContent().toString());
+ return response;
+ }
+
+ @Override
+ public String buildTmpPath(String actionKey, String codeType) {
+ return Path.of(TMP_ACTION_PATH, System.currentTimeMillis() + "-" + actionKey + codeType).toString();
+ }
+
+ @Override
+ public void tmpSerialize(MetaAction tempAction, String code, String codeType) throws IOException {
+ log.debug("行动程序临时序列化: {}", tempAction);
+ Path path = Path.of(tempAction.getLocation());
+ File file = path.toFile();
+ file.createNewFile();
+ Files.writeString(path, code);
+ log.debug("临时序列化完毕");
+ }
+
+ private static @NotNull Path createActionDir(String baseName, Path baseDir) {
+ Path actionDir = null;
+
+ // 原子地“抢占”目录名
+ for (int i = 0; ; i++) {
+ String dirName = (i == 0) ? baseName : baseName + "(" + i + ")";
+ Path candidate = baseDir.resolve(dirName);
+
+ try {
+ Files.createDirectory(candidate); // 原子操作
+ actionDir = candidate;
+ break;
+ } catch (FileAlreadyExistsException ignored) {
+ // 继续尝试下一个名字
+ } catch (IOException e) {
+ throw new ActionSerializeFailedException(
+ "无法创建行动目录: " + candidate.toAbsolutePath(), e
+ );
+ }
+ }
+ return actionDir;
+ }
+
+ @Override
+ public void persistSerialize(MetaActionInfo metaActionInfo, ActionFileMetaData fileMetaData) {
+ log.debug("行动程序持久序列化: {}", metaActionInfo);
+ val baseDir = Path.of(DYNAMIC_ACTION_PATH);
+
+ if (!Files.isDirectory(baseDir)) {
+ throw new ActionSerializeFailedException(
+ "目录不存在或不可用: " + baseDir.toAbsolutePath()
+ );
+ }
+
+ val baseName = fileMetaData.getName();
+ val ext = fileMetaData.getExt();
+
+ val actionDir = createActionDir(baseName, baseDir);
+
+ // 使用临时文件写入内容
+ val runTmp = actionDir.resolve("run." + ext + ".tmp");
+ val descTmp = actionDir.resolve("desc.json.tmp");
+
+ val runFinal = actionDir.resolve("run." + ext);
+ val descFinal = actionDir.resolve("desc.json");
+
+ try {
+ Files.writeString(runTmp, fileMetaData.getContent());
+ Files.writeString(descTmp, JSONObject.toJSONString(metaActionInfo));
+
+ // 原子提交
+ Files.move(runTmp, runFinal, StandardCopyOption.ATOMIC_MOVE);
+ Files.move(descTmp, descFinal, StandardCopyOption.ATOMIC_MOVE);
+ } catch (IOException e) {
+ // 失败清理
+ safeDelete(runTmp);
+ safeDelete(descTmp);
+ safeDelete(runFinal);
+ safeDelete(descFinal);
+ safeDelete(actionDir);
+ throw new ActionSerializeFailedException("行动文件写入失败", e);
+ }
+ log.debug("持久序列化结束");
+ }
+
+ private void safeDelete(Path path) {
+ try {
+ if (Files.exists(path)) {
+ Files.delete(path);
+ }
+ } catch (IOException ignored) {
+ }
+ }
+
+
+ @Override
+ public JSONObject listSysDependencies() {
+ // 先只列出系统/环境的 Python 依赖
+ // TODO 在 AgentConfigManager 内配置启用的脚本语言及对应的扩展名
+ // 这里的逻辑后续需要替换为“根据 AgentConfigManager 读取到的脚本语言启用情况，遍历并列出当前系统环境依赖”
+ // 还需要将返回值调整为相应的数据类
+ // 后续还需要将不同语言的处理逻辑分散到不同方法内，这里为了验证，先写死在当前方法
+ JSONObject sysDependencies = new JSONObject();
+ sysDependencies.put("language", "Python");
+ JSONArray dependencies = sysDependencies.putArray("dependencies");
+ SystemExecHelper.Result pyResult = SystemExecHelper.exec("pip", "list", "--format=freeze");
+ System.out.println(pyResult);
+ if (pyResult.isOk()) {
+ List resultList = pyResult.getResultList();
+ for (String result : resultList) {
+ JSONObject element = dependencies.addObject();
+ String[] split = result.split("==");
+ element.put("name", split[0]);
+ element.put("version", split[1]);
+ }
+ } else {
+ JSONObject element = dependencies.addObject();
+ element.put("error", pyResult.getTotal());
+ }
+ return sysDependencies;
+ }
+
+ private void registerMcpClient(String id, McpClientTransport clientTransport, int timeout) {
+ McpSyncClient client = McpClient.sync(clientTransport)
+ .requestTimeout(Duration.ofSeconds(timeout))
+ .clientInfo(new McpSchema.Implementation(id, "PARTNER"))
+ // 行动程序(现 MCP Tool)的描述文本将直接由resources返回
+ // 原因: ToolChange 发送的内容侧重调用，缺少可承担描述文本的字段
+ // ResourcesChange 事件传递的 Resource 可以由 Client 读取内容
+ // 预计在 Server 侧，收到客户端发送的新的行动程序信息，该信息由客户端处补充后，将其放置在指定位置
+ // 并写入描述文件、发起 ResourcesChange 事件
+ .build();
+ mcpClients.put(id, client);
+ }
+
+ private void setupShutdownHook() {
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ dynamicActionMcpServer.close();
+ mcpDescServer.close();
+ this.mcpClients.forEach((id, client) -> {
+ client.close();
+ log.info("[{}] MCP-Client 已关闭", id);
+ });
+ }));
+ }
+
+ private interface LocalWatchServiceBuild {
+ LocalWatchServiceBuild registerCreate(EventHandler handler);
+
+ LocalWatchServiceBuild registerModify(EventHandler handler);
+
+ LocalWatchServiceBuild registerDelete(EventHandler handler);
+
+ LocalWatchServiceBuild registerOverflow(EventHandler handler);
+
+ LocalWatchServiceBuild initialLoad(InitLoader loader);
+
+ LocalWatchServiceBuild watchAll(boolean watchAll);
+
+ void commit(ExecutorService executor);
+
+ interface EventHandler {
+ void handle(Path thisDir, Path context);
+ }
+
+ interface InitLoader {
+ void load();
+ }
+
+ class BuildRegistry implements LocalWatchServiceBuild {
+
+ private final Map, EventHandler> handlers = new HashMap<>();
+ private InitLoader initLoader;
+ private final WatchContext ctx;
+ private boolean watchAll = false;
+
+ private BuildRegistry(WatchContext ctx) {
+ this.ctx = ctx;
+ }
+
+ @Override
+ public LocalWatchServiceBuild registerCreate(EventHandler handler) {
+ ctx.kinds.add(StandardWatchEventKinds.ENTRY_CREATE);
+ handlers.put(StandardWatchEventKinds.ENTRY_CREATE, handler);
+ return this;
+ }
+
+ @Override
+ public LocalWatchServiceBuild registerModify(EventHandler handler) {
+ ctx.kinds.add(StandardWatchEventKinds.ENTRY_MODIFY);
+ handlers.put(StandardWatchEventKinds.ENTRY_MODIFY, handler);
+ return this;
+ }
+
+ @Override
+ public LocalWatchServiceBuild registerDelete(EventHandler handler) {
+ ctx.kinds.add(StandardWatchEventKinds.ENTRY_DELETE);
+ handlers.put(StandardWatchEventKinds.ENTRY_DELETE, handler);
+ return this;
+ }
+
+ @Override
+ public LocalWatchServiceBuild registerOverflow(EventHandler handler) {
+ ctx.kinds.add(StandardWatchEventKinds.OVERFLOW);
+ handlers.put(StandardWatchEventKinds.OVERFLOW, handler);
+ return this;
+ }
+
+ @Override
+ public LocalWatchServiceBuild initialLoad(InitLoader loader) {
+ initLoader = loader;
+ return this;
+ }
+
+ @Override
+ public LocalWatchServiceBuild watchAll(boolean watchAll) {
+ this.watchAll = watchAll;
+ return this;
+ }
+
+ @Override
+ public void commit(ExecutorService executor) {
+ registerPath();
+ if (initLoader != null)
+ initLoader.load();
+ executor.execute(buildWatchTask());
+ }
+
+ private void registerPath() {
+ Path root = ctx.root;
+ WatchService watchService = ctx.watchService;
+ Map watchKeys = ctx.watchKeys;
+ try {
+ WatchEvent.Kind[] kindsArray = ctx.kinds.toArray(WatchEvent.Kind[]::new);
+ WatchKey rootKey = root.register(watchService, kindsArray);
+ watchKeys.put(rootKey, root);
+ if (!watchAll) {
+ return;
+ }
+ Stream walk = Files.list(root).filter(Files::isDirectory);
+ for (Path dir : walk.toList()) {
+ WatchKey key = dir.register(watchService, kindsArray);
+ watchKeys.put(key, dir);
+ log.debug("注册目录监听: {}", dir);
+ }
+ walk.close();
+ } catch (IOException e) {
+ log.error("监听目录注册失败: ", e);
+ }
+ }
+
+ private Runnable buildWatchTask() {
+ return () -> {
+ String rootStr = ctx.root.toString();
+ log.info("行动程序目录监听器已启动，监听目录: {}", rootStr);
+ while (true) {
+ WatchKey key = null;
+ try {
+ key = ctx.watchService.take();
+ List> events = key.pollEvents();
+ for (WatchEvent e : events) {
+ WatchEvent.Kind kind = e.kind();
+ Object context = e.context();
+ log.debug("文件目录监听事件: {} - {} - {}", rootStr, kind.name(), context);
+ Path thisDir = (Path) key.watchable();
+ EventHandler handler = handlers.get(kind);
+ if (handler == null) {
+ continue;
+ }
+ handler.handle(thisDir, context instanceof Path ? thisDir.resolve((Path) context) : null);
+ }
+ } catch (InterruptedException e) {
+ log.info("监听线程被中断，准备退出...");
+ Thread.currentThread().interrupt(); // 恢复中断标志
+ break;
+ } catch (ClosedWatchServiceException e) {
+ log.info("WatchService 已关闭，监听线程退出。");
+ break;
+ } finally {
+ if (key != null) {
+ // reset 返回 false 表示该 key 已失效（目录被删、不可访问等）
+ boolean valid = key.reset();
+ if (!valid) {
+ log.info("WatchKey 已失效，停止监听该目录: {}", key.watchable());
+ ctx.watchKeys.remove(key);
+ if (key.watchable().equals(ctx.root)) {
+ try {
+ Files.createDirectories(ctx.root);
+ registerPath();
+ if (initLoader != null)
+ initLoader.load();
+ } catch (IOException e) {
+ log.error("重建根目录并重新注册监听失败: {}", ctx.root, e);
+ }
+ }
+ }
+ }
+ }
+ }
+ };
+ }
+
+ }
+ }
+
+ private record WatchContext(Path root, WatchService watchService, Map watchKeys,
+ List> kinds) {
+ private WatchContext(Path root, WatchService watchService) {
+ this(root, watchService, new HashMap<>(), new ArrayList<>());
+ }
+ }
+
+ private sealed static abstract class LocalWatchEventProcessor permits LocalWatchEventProcessor.Dynamic, LocalWatchEventProcessor.Desc, LocalWatchEventProcessor.Common {
+
+ protected final ConcurrentHashMap existedMetaActions;
+ protected final WatchContext ctx;
+
+ private LocalWatchEventProcessor(ConcurrentHashMap existedMetaActions, WatchContext ctx) {
+ this.existedMetaActions = existedMetaActions;
+ this.ctx = ctx;
+ }
+
+ protected abstract @NotNull LocalWatchServiceBuild.InitLoader buildLoad();
+
+ protected abstract @NotNull LocalWatchServiceBuild.EventHandler buildModify();
+
+ protected abstract @NotNull LocalWatchServiceBuild.EventHandler buildCreate();
+
+ protected abstract @NotNull LocalWatchServiceBuild.EventHandler buildDelete();
+
+ protected abstract @NotNull LocalWatchServiceBuild.EventHandler buildOverflow();
+
+ protected File[] loadFiles(Path root) {
+ // 在批量删除场景下，在接收到事件时目录等内容可能已被删除，此时不应该报错，而是返回一个‘异常值’
+ if (!Files.isDirectory(root)) {
+ return null;
+ }
+ return root.toFile().listFiles();
+ }
+
+ @SuppressWarnings("LoggingSimilarMessage")
+ private static final class Dynamic extends LocalWatchEventProcessor {
+
+ private final McpStatelessAsyncServer dynamicActionMcpServer;
+
+ private Dynamic(ConcurrentHashMap existedMetaActions, McpStatelessAsyncServer dynamicActionMcpServer, WatchContext ctx) {
+ super(existedMetaActions, ctx);
+ this.dynamicActionMcpServer = dynamicActionMcpServer;
+ }
+
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+ private boolean normalPath(Path path) {
+ val files = loadFiles(path);
+ if (files == null || files.length < 2) {
+ return false;
+ }
+ boolean desc = false;
+ int run = 0;
+ for (File f : files) {
+ String fileName = f.getName();
+ if (fileName.equals("desc.json")) {
+ desc = true;
+ }
+ if (fileName.startsWith("run.")) {
+ run++;
+ }
+ }
+ return run == 1 && desc;
+ }
+
+ private boolean addAction(String name, Path dir) {
+ File program = null;
+ try (Stream stream = Files.list(dir)) {
+ for (Path p : stream.toList()) {
+ if (p.getFileName().toString().startsWith("run."))
+ program = p.toFile();
+ }
+ } catch (Exception e) {
+ log.error("添加 action 失败", e);
+ return false;
+ }
+ MetaActionInfo info;
+ try {
+ info = JSONUtil.readJSONObject(dir.resolve("desc.json").toFile(), StandardCharsets.UTF_8).toBean(MetaActionInfo.class);
+ } catch (Exception e) {
+ log.error("desc.json 加载失败: {}", dir);
+ return false;
+ }
+ String actionKey = "local::" + name;
+ existedMetaActions.put(actionKey, info);
+ dynamicActionMcpServer.addTool(buildAsyncToolSpecification(info, program, actionKey, name)).subscribe();
+ return true;
+ }
+
+ private void removeAction(String name) {
+ existedMetaActions.remove("local::" + name);
+ dynamicActionMcpServer.removeTool(name).subscribe();
+ }
+
+ private BiFunction> buildToolHandler(File finalProgram) {
+ return (mcpTransportContext, callToolRequest) -> {
+ Map arguments = callToolRequest.arguments();
+ if (arguments == null) {
+ arguments = Map.of();
+ }
+ String ext = FileUtil.getSuffix(finalProgram);
+ String[] commands = SystemExecHelper.buildCommands(ext, arguments, finalProgram.getAbsolutePath());
+ if (commands == null) {
+ return Mono.just(McpSchema.CallToolResult.builder()
+ .addTextContent("未知文件类型: " + finalProgram.getName())
+ .isError(true)
+ .build());
+ }
+
+ return Mono.fromCallable(() -> {
+ SystemExecHelper.Result execResult = SystemExecHelper.exec(commands);
+ McpSchema.CallToolResult.Builder builder = McpSchema.CallToolResult.builder()
+ .isError(!execResult.isOk());
+ List resultList = execResult.getResultList();
+ if (resultList != null && !resultList.isEmpty()) {
+ builder.textContent(resultList);
+ builder.structuredContent(resultList);
+ } else {
+ builder.addTextContent(execResult.getTotal());
+ builder.structuredContent(execResult.getTotal());
+ }
+ return builder.build();
+ }).subscribeOn(Schedulers.boundedElastic());
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.InitLoader buildLoad() {
+ // 从该路径列出已存在的目录，每个目录对应不同的行动程序及描述文件，从描述文件加载程序信息
+ return () -> {
+ Path root = ctx.root;
+ File file = root.toFile();
+ if (file.isFile()) {
+ throw new ActionInitFailedException("未找到目录: " + root);
+ }
+ File[] files = file.listFiles();
+ if (files == null) {
+ throw new ActionInitFailedException("未正常读取目录: " + root);
+ }
+ for (File dir : files) {
+ if (!normalPath(dir.toPath())) {
+ continue;
+ }
+ addAction(dir.getName(), dir.toPath());
+ }
+ };
+ }
+
+ private McpStatelessServerFeatures.AsyncToolSpecification buildAsyncToolSpecification(MetaActionInfo
+ info, File program, String actionKey, String name) {
+ Map additional = Map.of("pre", info.getPreActions(),
+ "post", info.getPostActions(),
+ "strict_pre", info.isStrictDependencies(),
+ "io", info.isIo());
+ McpSchema.Tool tool = McpSchema.Tool.builder()
+ .name(name)
+ .description(info.getDescription())
+ .inputSchema(McpJsonMapper.getDefault(), JSONObject.toJSONString(info.getParams()))
+ .outputSchema(info.getResponseSchema())
+ .title(actionKey)
+ .meta(additional)
+ .build();
+ return McpStatelessServerFeatures.AsyncToolSpecification.builder()
+ .tool(tool)
+ .callHandler(buildToolHandler(program))
+ .build();
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildModify() {
+ return (thisDir, context) -> {
+ // 查看当前目录是否为空或者能否正常读取
+ if (!normalPath(thisDir)) {
+ return;
+ }
+ // 对应本地程序或者描述文件的修改行为
+ modify(thisDir, context);
+ };
+ }
+
+ private void modify(Path thisDir, Path context) {
+ String fileName = context.getFileName().toString();
+ if (fileName.equals("desc.json")) {
+ handleMetaModify(thisDir);
+ }
+ if (fileName.startsWith("run.")) {
+ handleProgramModify(thisDir);
+ }
+ }
+
+ private void handleProgramModify(Path thisDir) {
+ String name = thisDir.getFileName().toString();
+ String actionKey = "local::" + name;
+ // 检查是否存在当前 program 对应的 Tool
+ if (existedMetaActions.containsKey(actionKey)) {
+ return;
+ }
+ if (!addAction(name, thisDir)) {
+ removeAction(name);
+ }
+ }
+
+ private void handleMetaModify(Path thisDir) {
+ // 检查是否除了描述文件外还存在别的可执行文件
+ String name = thisDir.getFileName().toString();
+ if (!addAction(name, thisDir)) {
+ removeAction(name);
+ }
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildCreate() {
+ return (thisDir, context) -> {
+ if (thisDir.equals(ctx.root) && Files.isDirectory(context)) {
+ try {
+ context.register(ctx.watchService, ctx.kinds.toArray(WatchEvent.Kind[]::new));
+ } catch (IOException e) {
+ log.error("监听目录注册失败: {}", context);
+ }
+ }
+ if (normalPath(thisDir)) {
+ modify(thisDir, context);
+ }
+ if (Files.isDirectory(context) && normalPath(context)) {
+ File[] files = context.toFile().listFiles();
+ if (files == null) {
+ log.warn("目录无法访问: {}", context);
+ return;
+ }
+ for (File f : files) {
+ modify(context, f.toPath());
+ }
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildDelete() {
+ return (thisDir, context) -> {
+ // 如果发生在 root，且context为目录，则需要移除对应的 action
+ if (thisDir.equals(ctx.root)) {
+ String name = context.getFileName().toString();
+ Path candidate = ctx.root.resolve(name);
+
+ // 如果 root 下仍然存在一个同名目录
+ if (Files.isDirectory(candidate)) {
+ // 说明删的是同名文件，不是 action 目录
+ return;
+ }
+
+ removeAction(name);
+
+ AtomicReference toRemove = new AtomicReference<>();
+ ctx.watchKeys.forEach((key, path) -> {
+ if (path.getFileName().toString().equals(name)) {
+ key.cancel();
+ toRemove.set(key);
+ }
+ });
+ if (toRemove.get() != null) {
+ ctx.watchKeys.remove(toRemove.get());
+ }
+ return;
+ }
+
+ // 如果发生在非 root 目录内且 context 不符合 action 目录特征
+ // 由于只会监听 root 目录与 action 目录
+ // 所以此时则证明当前目录对应的行动已不可靠，需要移除
+ if (!thisDir.equals(ctx.root) && !normalPath(thisDir)) {
+ // 未通过校验则删除对应的 action，并在 DynamicActonMcpServer 中删除对应的工具
+ String name = thisDir.getFileName().toString();
+ removeAction(name);
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildOverflow() {
+ return (thisDir, context) -> {
+ // 直接从 existedMetaActions 中拿取现有的 key，上游也从这里发现可用工具
+ Set existed = existedMetaActions.keySet().stream().map(actionKey -> actionKey.split("::")[1]).collect(Collectors.toSet());
+ Set currentDirs = new HashSet<>();
+ // 按照预期 root 目录下有效 path 只包括各个 action 目录
+ // 排除非目录 path
+ try (Stream stream = Files.list(ctx.root).filter(Files::isDirectory)) {
+ stream.forEach(path -> {
+ String name = path.getFileName().toString();
+ currentDirs.add(name);
+
+ boolean contains = existed.contains(name);
+ boolean normal = normalPath(path);
+
+ // 如果该目录对应 action 已被记录，且符合 action 目录要求，则无处理
+ // 如果已被记录，但不符合，则移除行动
+ // 此时必定被监听
+ if (contains && !normal) {
+ removeAction(name);
+ }
+
+ // 如果 action 没有记录，但符合要求，由于此时必定尚未被监听，则注册监听且添加新 action
+ // 如果未被记录，且不符合要求，则只注册监听
+ if (!contains) {
+ boolean alreadyWatching = ctx.watchKeys.values().stream()
+ .anyMatch(p -> p.equals(path));
+ if (!alreadyWatching) {
+ try {
+ WatchKey watchKey = path.register(ctx.watchService, ctx.kinds.toArray(WatchEvent.Kind[]::new));
+ ctx.watchKeys.put(watchKey, path);
+ } catch (IOException e) {
+ log.error("监听目录注册失败: {}", path);
+ }
+ }
+
+ if (normal) {
+ addAction(name, path);
+ }
+ }
+ });
+ } catch (IOException e) {
+ log.error("目录无法读取: {}", ctx.root);
+ return;
+ }
+ for (String existedName : existed) {
+ if (!currentDirs.contains(existedName)) {
+ removeAction(existedName);
+ }
+ }
+ };
+ }
+ }
+
+ private static final class Desc extends LocalWatchEventProcessor {
+
+ private final McpStatelessAsyncServer mcpDescServer;
+ private final ConcurrentHashMap descCache = new ConcurrentHashMap<>();
+
+ private Desc(ConcurrentHashMap existedMetaActions, McpStatelessAsyncServer mcpDescServer, WatchContext ctx) {
+ super(existedMetaActions, ctx);
+ this.mcpDescServer = mcpDescServer;
+ }
+
+ private McpStatelessServerFeatures.@NotNull AsyncResourceSpecification buildAsyncResourceSpecification(String name, String uri) {
+ McpSchema.Resource resource = McpSchema.Resource.builder()
+ .name(name)
+ .title(name)
+ .description("Action descriptor for " + name)
+ .mimeType("application/json")
+ .uri(uri)
+ .build();
+ BiFunction> readHandler = (context, request) -> {
+ String requestUri = request.uri();
+ String result = descCache.get(requestUri);
+ if (result == null) {
+ return Mono.error(new NotFoundException("未找到 Resource: " + requestUri));
+ }
+ return Mono.just(new McpSchema.ReadResourceResult(List.of(new McpSchema.TextResourceContents(requestUri, "application/json", result))));
+ };
+ return new McpStatelessServerFeatures.AsyncResourceSpecification(resource, readHandler);
+ }
+
+ private boolean normal(String fileName) {
+ return fileName.endsWith(".desc.json") && fileName.contains("::");
+ }
+
+ private boolean normal(File file) {
+ String name = file.getName();
+ return normal(name);
+ }
+
+ private boolean normal(Path path) {
+ return normal(path.toFile());
+ }
+
+ private boolean addResource(File file) {
+ String name = file.getName();
+ if (!normal(name)) {
+ return false;
+ }
+ // 读取并解析为 MetaActionInfo，存入 resources
+ try {
+ MetaActionInfo info = JSONUtil.readJSONObject(file, StandardCharsets.UTF_8).toBean(MetaActionInfo.class);
+ String uri = ctx.root.resolve(name).toUri().toString();
+ descCache.put(uri, JSONObject.toJSONString(info));
+ mcpDescServer.addResource(buildAsyncResourceSpecification(name, uri)).block();
+ String actionKey = name.replace(".desc.json", "");
+ if (existedMetaActions.containsKey(actionKey)) {
+ existedMetaActions.put(actionKey, info);
+ }
+ } catch (Exception e) {
+ log.error("desc.json 解析失败: {}", file.getAbsolutePath());
+ return false;
+ }
+ return true;
+ }
+
+ private void removeResource(Path path) {
+ String uri = path.toUri().toString();
+ String actionKey = path.getFileName().toString().replace(".desc.json", "");
+
+ descCache.remove(uri);
+ mcpDescServer.removeResource(uri).block();
+ if (existedMetaActions.containsKey(actionKey)) {
+ resetMetaActionInfo(existedMetaActions.get(actionKey));
+ }
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.InitLoader buildLoad() {
+ return () -> {
+ // DescMcp 的加载逻辑只负责读取已有的 *.desc.json 并注册为 resources
+ // 正常来讲 root 直接对应 MCP_DESC_PATH，先检查 root 是否为目录，否则拒绝启动
+ val files = loadFiles(ctx.root);
+ for (File file : files) {
+ addResource(file);
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildModify() {
+ return (thisDir, context) -> {
+ // 排除目录事件、名称不符合要求的文件
+ String fileName = context.getFileName().toString();
+ if (!Files.isRegularFile(context) || !normal(fileName)) {
+ return;
+ }
+
+ if (!addResource(context.toFile())) {
+ removeResource(context);
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildCreate() {
+ return buildModify();
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildDelete() {
+ return (thisDir, context) -> {
+ // 排除被删除文件名称不符合要求的事件
+ String fileName = context.getFileName().toString();
+ if (!normal(fileName)) {
+ return;
+ }
+
+ // DELETE 事件发生后，需要移除对应的 descCache 条目;
+ // 如果存在对应的 info,也需要将其中的额外信息进行重置，只保留 Tools 自身的信息
+ removeResource(context);
+ };
+ }
+
+ private void resetMetaActionInfo(MetaActionInfo info) {
+ info.setIo(false);
+ info.getTags().clear();
+ info.getPreActions().clear();
+ info.getPostActions().clear();
+ info.setStrictDependencies(false);
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildOverflow() {
+ return (thisDir, context) -> {
+ // 对于 OVERFLOW 事件，需要依据当前目录下的所有 *.desc.json 针对现有内容进行修复
+ List files;
+ try (Stream stream = Files.list(ctx.root)) {
+ files = stream.filter(Files::isRegularFile).filter(this::normal).map(Path::toFile).toList();
+ } catch (IOException e) {
+ log.error("目录无法访问: {}", ctx.root);
+ return;
+ }
+ Set currentUriStr = new HashSet<>();
+
+ for (File file : files) {
+ MetaActionInfo info = null;
+ try {
+ info = JSONUtil.readJSONObject(file, StandardCharsets.UTF_8).toBean(MetaActionInfo.class);
+ } catch (Exception e) {
+ log.warn("desc.json 读取失败: {}", file.toPath());
+ }
+
+ boolean available = info != null;
+
+ // 如果读取成功，则更新 descCache
+ // 若在 existedMetaActions 中存在，则更新对应 info
+ String uriStr = file.toURI().toString();
+ currentUriStr.add(uriStr);
+
+ if (available) {
+ // 由于涉及内容均为 map，所以额外判断没有必要，直接进行 add 行为即可
+ addResource(file);
+ } else {
+ // 如果读取失败，则移除对应 descCache 条目
+ // 若在 existedMetaActions 中存在，则重置对应 info
+ removeResource(file.toPath());
+ }
+
+ }
+
+ List serverUris = mcpDescServer.listResources()
+ .map(McpSchema.Resource::uri)
+ .collectList()
+ .block();
+ if (serverUris == null) {
+ log.error("无法获取 DescMcpServer 持有的资源列表");
+ return;
+ }
+ for (String uri : serverUris) {
+ if (!currentUriStr.contains(uri)) {
+ removeResource(Paths.get(URI.create(uri)));
+ }
+ }
+ };
+ }
+ }
+
+ private static final class Common extends LocalWatchEventProcessor {
+
+ private final Map mcpClients;
+ private final Map mcpConfigFileCache = new HashMap<>();
+ private final McpSyncClient descClient;
+
+ @SuppressWarnings("BooleanMethodIsAlwaysInverted")
+ private boolean normalFile(File file) {
+ return file.exists() && file.isFile() && file.getName().endsWith(".json");
+ }
+
+ private Common(ConcurrentHashMap existedMetaActions, Map mcpClients, McpSyncClient descClient, WatchContext ctx) {
+ super(existedMetaActions, ctx);
+ this.mcpClients = mcpClients;
+ this.descClient = descClient;
+ }
+
+ /**
+ * 该部分主要发生在扫描到新的MCP Server描述文件时出现的注册逻辑
+ *
+ * @param id MCP Client 的 id
+ * @param mcpClientTransportParams MCP Server 的参数
+ */
+ private void registerMcpClient(String id, McpClientTransportParams mcpClientTransportParams) {
+ // 如果已存在同名 client，则需要先获取并关闭
+ val old = mcpClients.get(id);
+ val clientTransport = createTransport(mcpClientTransportParams);
+ val timeout = mcpClientTransportParams.timeout;
+ val client = McpClient.sync(clientTransport)
+ .requestTimeout(Duration.ofSeconds(timeout))
+ .clientInfo(new McpSchema.Implementation(id, "PARTNER"))
+ .build();
+
+ try {
+ for (McpSchema.Tool tool : client.listTools().tools()) {
+ val metaActionInfo = buildMetaActionInfo(id, tool);
+ existedMetaActions.put(id + "::" + tool.name(), metaActionInfo);
+ }
+ mcpClients.put(id, client);
+
+ if (old != null) {
+ old.close();
+ }
+ } catch (Exception e) {
+ log.warn("[{}] MCP client init failed, skipped (probably non-stdio-safe)", id, e);
+ client.close();
+ }
+
+ }
+
+ /**
+ * 构建 MetaActionInfo
+ *
+ * @param id MCP Client id
+ * @param tool 待构建 MetaActionInfo 的 MCP Tool
+ * @return MetaActionInfo 信息
+ */
+ private @NotNull MetaActionInfo buildMetaActionInfo(String id, McpSchema.Tool tool) {
+ // 先检查是否存在对应的描述文件
+ val toolName = tool.name(); //格式: mcpName::toolName
+ for (McpSchema.Resource resource : descClient.listResources().resources()) {
+ val resourceName = resource.name();
+ if (resourceName.equals(toolName) && resourceName.split("::")[0].equals(id)) {
+ val resourceContent = descClient.readResource(resource).contents().getFirst();
+ if (resourceContent instanceof McpSchema.TextResourceContents r) {
+ try {
+ return JSONObject.parseObject(r.text(), MetaActionInfo.class);
+ } catch (Exception e) {
+ log.error("序列化描述文件资源失败", e);
+ log.error("序列化描述文件资源失败, uri: {}", r.uri());
+ log.error("序列化描述文件资源失败, text: {}", r.text());
+ }
+ }
+ }
+ }
+
+ MetaActionInfo info = new MetaActionInfo();
+ info.setDescription(tool.description());
+ Map outputSchema = tool.outputSchema();
+ info.setResponseSchema(outputSchema == null ? JSONObject.of() : JSONObject.from(outputSchema));
+ info.setParams(tool.inputSchema().properties());
+
+ val meta = tool.meta();
+ if (meta != null) {
+ JSONObject metaJson = JSONObject.from(meta);
+ info.setIo(metaJson.getBoolean("io"));
+ info.setPreActions(metaJson.getList("pre", String.class));
+ info.setPostActions(metaJson.getList("post", String.class));
+ info.setStrictDependencies(metaJson.getBoolean("strict"));
+ info.setTags(metaJson.getList("tag", String.class));
+ }
+ return info;
+ }
+
+ private McpClientTransport createTransport(McpClientTransportParams mcpClientTransportParams) {
+ return switch (mcpClientTransportParams) {
+ case McpClientTransportParams.Stdio params -> {
+ val serverParameters = ServerParameters.builder(params.command).env(params.env).args(params.args).build();
+ yield new StdioClientTransport(serverParameters, McpJsonMapper.getDefault());
+ }
+ case McpClientTransportParams.Http params -> {
+ val customizer = new McpSyncHttpClientRequestCustomizer() {
+ @Override
+ public void customize(HttpRequest.Builder builder, String method, URI endpoint, String body, McpTransportContext context) {
+ params.headers.forEach(builder::setHeader);
+ }
+ };
+ yield HttpClientSseClientTransport.builder(params.baseUri).httpRequestCustomizer(customizer).sseEndpoint(params.endpoint).build();
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.InitLoader buildLoad() {
+ return () -> {
+ // For CommonMcp, we need to list all files in MCP_SERVER_PATH,
+ // and search for files with extend name .json,
+ // and then reading them as JSONObject to get McpClientTransportParams.
+ val files = loadFiles(ctx.root);
+
+ for (File file : files) {
+ if (!normalFile(file)) {
+ continue;
+ }
+ loadAndRegisterMcpClientsFromFile(file);
+ }
+ };
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildModify() {
+ /*
+ 发现文件更改事件时，读取该文件中存放的mcp配置，与现有的 MCP 记录对比
+ 根据其是否发生配置变动等，针对对应的 client 进行调整
+ 如果额外维护一个 文件-clientIds 的映射，可以解决删除某一mcp的情况
+ 但如果对于极端场景：从某一文件剪切并粘贴至另一文件，但后者先于前者保存
+ 此时就会出现问题，重复client无法被注册
+ 建议对于这种‘分布式’的配置存放方式，每个文件变更最好都触发全量加载
+ */
+ return (thisDir, context) -> checkAndReload(true);
+ }
+
+ private cn.hutool.json.JSONObject readJson(File file) {
+ try {
+ return JSONUtil.readJSONObject(file, StandardCharsets.UTF_8);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ private cn.hutool.json.JSONObject readMcp(cn.hutool.json.JSONObject json, String id) {
+ try {
+ return json.getJSONObject(id);
+ } catch (Exception ignored) {
+ return null;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private McpClientTransportParams readParams(cn.hutool.json.JSONObject mcp) {
+ val stdioKeys = Set.of("command", "args", "env");
+ val httpKeys = Set.of("uri", "endpoint", "headers");
+ val httpKey = Set.of("url");
+ val keys = mcp.keySet();
+ val timeout = mcp.getInt("timeout", 30);
+
+ if (keys.equals(stdioKeys)) {
+ val command = mcp.getStr("command");
+ val env = mcp.getBean("env", Map.class);
+ val args = mcp.getBeanList("args", String.class);
+ if (command == null || env == null || args == null) {
+ return null;
+ }
+ return new McpClientTransportParams.Stdio(timeout, command, env, args);
+ }
+
+ if (keys.equals(httpKeys)) {
+ val uri = mcp.getStr("uri");
+ val endpoint = mcp.getStr("endpoint");
+ val headers = mcp.getBean("headers", Map.class);
+ if (uri == null || endpoint == null || headers == null) {
+ return null;
+ }
+ return new McpClientTransportParams.Http(timeout, uri, endpoint, headers);
+ }
+
+ if (keys.equals(httpKey)) {
+ val url = mcp.getStr("url");
+ if (url == null) {
+ return null;
+ }
+ return new McpClientTransportParams.Http(timeout, url, "", Map.of());
+ }
+
+ return null;
+ }
+
+ private void checkAndReload(boolean trustCache) {
+ /*
+ for each file cannot present all mcp configurations,
+ we need to load all at once, and then compare them with existed records.
+ we will record existing mcp paramsCacheMap and id-params map for which is changed.
+
+ recording changedMap only cannot figure out which mcp was deleted,
+ so existingMcpIdSet attr is required
+ */
+ val changedMap = new HashMap();
+ val existingMcpIdSet = new HashSet();
+
+ val files = loadFiles(ctx.root);
+ for (File file : files) {
+ if (!normalFile(file)) {
+ continue;
+ }
+
+ // check if necessary stats changed, null record is seen as file changed
+ val fileRecord = mcpConfigFileCache.get(file);
+ boolean fileRecordExists = fileRecord != null;
+ if (fileRecordExists && !fileChanged(file, fileRecord) && trustCache) {
+ existingMcpIdSet.addAll(fileRecord.paramsCacheMap().keySet());
+ continue;
+ }
+
+ // if changed, read file and load mcp configurations
+ val mcpConfigJson = readJson(file);
+ if (mcpConfigJson == null) {
+ // uses old records to avoid abnormal deletion
+ if (fileRecordExists) {
+ existingMcpIdSet.addAll(fileRecord.paramsCacheMap().keySet());
+ }
+ continue;
+ }
+
+ val newFileRecord = new McpConfigFileRecord(file.lastModified(), file.length(), new HashMap<>());
+ for (String id : mcpConfigJson.keySet()) {
+ val mcp = readMcp(mcpConfigJson, id);
+ if (mcp == null) {
+ continue;
+ }
+
+ val params = readParams(mcp);
+ if (params == null) {
+ continue;
+ }
+
+ existingMcpIdSet.add(id);
+ newFileRecord.paramsCacheMap().put(id, params);
+
+ if (fileRecordExists) {
+ val paramsCache = fileRecord.paramsCacheMap().get(id);
+ if (paramsCache != null && paramsCache.equals(params)) {
+ continue;
+ }
+ }
+
+ changedMap.put(id, params);
+ }
+ mcpConfigFileCache.put(file, newFileRecord);
+ }
+
+ updateMcpClients(changedMap, existingMcpIdSet);
+ }
+
+ private void updateMcpClients(HashMap changedMap, HashSet existingMcpIdSet) {
+ // following attr changedMap, update or insert mcp clients
+ changedMap.forEach((id, params) -> {
+ // close outdated clients if exists
+ val oldClient = mcpClients.get(id);
+ if (oldClient != null) {
+ oldClient.close();
+ }
+ // create new clients
+ registerMcpClient(id, params);
+ });
+
+ // following attr existingMcpIdSet, align mcp clients
+ // new mcp clients and outdated clients has been updated in above logic
+ // this part focus on removing non-existing mcp
+ mcpClients.keySet().removeIf(id -> !existingMcpIdSet.contains(id));
+ // clear relevant tools' action info
+ existedMetaActions.keySet().removeIf(id -> !existingMcpIdSet.contains(id.split("::")[0]));
+ }
+
+ private boolean fileChanged(File file, McpConfigFileRecord fileRecord) {
+ val lastModified = file.lastModified();
+ val length = file.length();
+
+ return fileRecord.lastModified() != lastModified || fileRecord.length() != length;
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildOverflow() {
+ return (thisDir, context) -> checkAndReload(false);
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildCreate() {
+ return (thisDir, context) -> {
+ val file = context.toFile();
+ if (!normalFile(file)) {
+ return;
+ }
+
+ loadAndRegisterMcpClientsFromFile(file);
+ };
+ }
+
+ private void loadAndRegisterMcpClientsFromFile(File file) {
+ val mcpConfigJson = readJson(file);
+ if (mcpConfigJson == null) {
+ return;
+ }
+
+ val newFileRecord = new McpConfigFileRecord(file.lastModified(), file.length());
+ for (String id : mcpConfigJson.keySet()) {
+ val mcp = readMcp(mcpConfigJson, id);
+ if (mcp == null) {
+ continue;
+ }
+
+ val params = readParams(mcp);
+ if (params == null) {
+ continue;
+ }
+
+ registerMcpClient(id, params);
+ newFileRecord.paramsCacheMap().put(id, params);
+ }
+ mcpConfigFileCache.put(file, newFileRecord);
+ }
+
+ @Override
+ @NotNull
+ protected LocalWatchServiceBuild.EventHandler buildDelete() {
+ return (thisDir, context) -> {
+ val file = context.toFile();
+ if (!file.getName().endsWith(".json")) {
+ return;
+ }
+
+ val fileRecord = mcpConfigFileCache.remove(file);
+ if (fileRecord == null) {
+ return;
+ }
+
+ // clear from existedMetaActions and mcpClients
+ // client id comes from fileRecord.paramsCache
+ // actionKey from `id::toolName`
+ val clientIdSet = fileRecord.paramsCacheMap().keySet();
+ for (String clientId : clientIdSet) {
+ val client = mcpClients.remove(clientId);
+ if (client == null) {
+ continue;
+ }
+
+ val tools = client.listTools().tools();
+ for (McpSchema.Tool tool : tools) {
+ val actionKey = clientId + "::" + tool.name();
+ existedMetaActions.remove(actionKey);
+ }
+ client.close();
+ }
+ };
+ }
+
+ private record McpConfigFileRecord(long lastModified, long length,
+ Map paramsCacheMap) {
+ public McpConfigFileRecord(long lastModified, long length) {
+ this(lastModified, length, new HashMap<>());
+ }
+ }
+ }
+ }
+
+ private sealed abstract static class McpClientTransportParams permits McpClientTransportParams.Http, McpClientTransportParams.Stdio {
+ private final int timeout;
+
+ private McpClientTransportParams(int timeout) {
+ this.timeout = timeout;
+ }
+
+ private final static class Http extends McpClientTransportParams {
+ private final String baseUri;
+ private final String endpoint;
+ private final Map headers;
+
+ private Http(int timeout, String baseUri, String endpoint, Map header) {
+ super(timeout);
+ this.baseUri = baseUri;
+ this.endpoint = endpoint;
+ this.headers = header;
+ }
+ }
+
+ private final static class Stdio extends McpClientTransportParams {
+ private final String command;
+ private final Map env;
+ private final List args;
+
+ private Stdio(int timeout, String command, Map env, List args) {
+ super(timeout);
+ this.command = command;
+ this.env = env;
+ this.args = args;
+ }
+ }
+ }
+
+ private static class SystemExecHelper {
+
+ //TODO 后续需在加载时、或者通过配置文件获取可用命令并注册匹配
+ private static String[] buildCommands(String ext, Map params, String absolutePath) {
+ String command = switch (ext) {
+ case "py" -> "python";
+ case "sh" -> "bash";
+ default -> null;
+ };
+ if (command == null) {
+ return null;
+ }
+ int paramSize = params == null ? 0 : params.size();
+ String[] commands = new String[paramSize + 2];
+ commands[0] = command;
+ commands[1] = absolutePath;
+ AtomicInteger paramCount = new AtomicInteger(2);
+ if (params != null) {
+ params.forEach((param, value) -> commands[paramCount.getAndIncrement()] = "--" + param + "=" + value);
+ }
+ return commands;
+ }
+
+ private static Result exec(String... command) {
+ Result result = new Result();
+ List output = new ArrayList<>();
+ List error = new ArrayList<>();
+
+ try {
+ Process process = new ProcessBuilder(command)
+ .redirectErrorStream(false) // 分开读
+ .start();
+
+ Thread stdoutThread = new Thread(() -> {
+ try (BufferedReader r = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+ String line;
+ while ((line = r.readLine()) != null) {
+ output.add(line);
+ }
+ } catch (Exception ignored) {
+ }
+ });
+
+ Thread stderrThread = new Thread(() -> {
+ try (BufferedReader r = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+ String line;
+ while ((line = r.readLine()) != null) {
+ error.add(line);
+ }
+ } catch (Exception ignored) {
+ }
+ });
+
+ stdoutThread.start();
+ stderrThread.start();
+
+ int exitCode = process.waitFor();
+ stdoutThread.join();
+ stderrThread.join();
+
+ result.setOk(exitCode == 0);
+ result.setResultList(output.isEmpty() ? error : output);
+ result.setTotal(String.join("\n", output.isEmpty() ? error : output));
+
+ } catch (Exception e) {
+ result.setOk(false);
+ result.setTotal(e.getMessage());
+ }
+
+ return result;
+ }
+
+ @Data
+ private static class Result {
+ private boolean ok;
+ private String total;
+ private List resultList;
+ }
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/RunnerClient.java b/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/RunnerClient.java
new file mode 100644
index 00000000..ddc32673
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/RunnerClient.java
@@ -0,0 +1,107 @@
+package work.slhaf.partner.core.action.runner;
+
+import com.alibaba.fastjson2.JSONObject;
+import io.modelcontextprotocol.server.McpStatelessAsyncServer;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.jetbrains.annotations.Nullable;
+import work.slhaf.partner.core.action.entity.ActionFileMetaData;
+import work.slhaf.partner.core.action.entity.MetaAction;
+import work.slhaf.partner.core.action.entity.MetaAction.Result;
+import work.slhaf.partner.core.action.entity.MetaActionInfo;
+import work.slhaf.partner.core.action.exception.ActionInitFailedException;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+import static work.slhaf.partner.common.Constant.Path.DATA;
+import static work.slhaf.partner.common.util.PathUtil.buildPathStr;
+
+/**
+ * 执行客户端抽象类
+ *
+ * 只负责暴露序列化、执行等相应接口，具体逻辑交给下游实现
+ *
+ * 默认存在两类实现，{@link LocalRunnerClient} 和 {@link SandboxRunnerClient}
+ *
+ * LocalRunnerClient:
+ * -
+ * 对应本地运行环境，可在本地启动 MCP 客户端将 RunnerClient 暴露的能力接口转发至本地 MCP Client 并执行
+ *
+ * SandboxRunnerClient:
+ * -
+ * 对应沙盒运行环境，该 Client 仅作为沙盒环境的客户端，不持有额外能力，仅保持远端连接已存在行动的内容更新
+ *
+ *
+ */
+@Slf4j
+public abstract class RunnerClient {
+
+ protected final String ACTION_PATH;
+
+ protected final ConcurrentHashMap existedMetaActions;
+ protected final ExecutorService executor;
+ //TODO 仍可提供内部 MCP，但调用方式需要结合 AgentContext来获取，否则生命周期不合
+ protected McpStatelessAsyncServer innerMcpServer;
+
+ /**
+ * ActionCore 将注入虚拟线程池
+ */
+ public RunnerClient(ConcurrentHashMap existedMetaActions, ExecutorService executor, @Nullable String baseActionPath) {
+ this.existedMetaActions = existedMetaActions;
+ this.executor = executor;
+ baseActionPath = baseActionPath == null ? DATA : baseActionPath;
+ this.ACTION_PATH = buildPathStr(baseActionPath, "action");
+
+ createPath(ACTION_PATH);
+ }
+
+ /**
+ * 执行行动程序
+ */
+ public void submit(MetaAction metaAction) {
+ // 获取已存在行动列表
+ Result result = metaAction.getResult();
+ if (!result.getStatus().equals(Result.Status.WAITING)) {
+ return;
+ }
+ RunnerResponse response = doRun(metaAction);
+ result.setData(response.getData());
+ result.setStatus(response.isOk() ? Result.Status.SUCCESS : Result.Status.FAILED);
+ }
+
+ protected abstract RunnerResponse doRun(MetaAction metaAction);
+
+ public abstract String buildTmpPath(String actionKey, String codeType);
+
+ public abstract void tmpSerialize(MetaAction tempAction, String code, String codeType) throws IOException;
+
+ public abstract void persistSerialize(MetaActionInfo metaActionInfo, ActionFileMetaData fileMetaData);
+
+ protected void createPath(String pathStr) {
+ val path = Path.of(pathStr);
+ try {
+ Files.createDirectory(path);
+ } catch (IOException e) {
+ if (!Files.exists(path)) {
+ throw new ActionInitFailedException("目录创建失败: " + pathStr, e);
+ }
+ }
+ }
+
+ /**
+ * 列出执行环境下的系统依赖情况
+ */
+ public abstract JSONObject listSysDependencies();
+
+ @Data
+ public static class RunnerResponse {
+ private boolean ok;
+ private String data;
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/SandboxRunnerClient.java b/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/SandboxRunnerClient.java
new file mode 100644
index 00000000..7bc34fcf
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/core/action/runner/SandboxRunnerClient.java
@@ -0,0 +1,57 @@
+package work.slhaf.partner.core.action.runner;
+
+import com.alibaba.fastjson2.JSONObject;
+import work.slhaf.partner.core.action.entity.ActionFileMetaData;
+import work.slhaf.partner.core.action.entity.MetaAction;
+import work.slhaf.partner.core.action.entity.MetaActionInfo;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * 基于 Http 与 WebSocket 的沙盒执行器客户端，负责:
+ *
+ * -
+ * 发送行动单元数据
+ *
+ * -
+ * 实时更新获取已存在行动列表
+ *
+ * -
+ * 向传入的 MetaAction 回写执行结果
+ *
+ *
+ */
+public class SandboxRunnerClient extends RunnerClient {
+
+ public SandboxRunnerClient(ConcurrentHashMap existedMetaActions, ExecutorService executor) { // 连接沙盒执行器(websocket)
+ super(existedMetaActions, executor, null);
+ }
+
+ protected RunnerResponse doRun(MetaAction metaAction) {
+ // 调用沙盒执行器
+ return null;
+ }
+
+ @Override
+ public JSONObject listSysDependencies() {
+ return null;
+ }
+
+ @Override
+ public String buildTmpPath(String actionKey, String codeType) {
+ throw new UnsupportedOperationException("Unimplemented method 'buildTmpPath'");
+ }
+
+ @Override
+ public void tmpSerialize(MetaAction tempAction, String code, String codeType) throws IOException {
+ throw new UnsupportedOperationException("Unimplemented method 'tmpSerialize'");
+ }
+
+ @Override
+ public void persistSerialize(MetaActionInfo metaActionInfo, ActionFileMetaData fileMetaData) {
+ throw new UnsupportedOperationException("Unimplemented method 'persistSerialize'");
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCapability.java b/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCapability.java
index 81c4b290..4f9ac125 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCapability.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCapability.java
@@ -13,7 +13,6 @@ import java.util.concurrent.locks.Lock;
public interface CognationCapability {
List getChatMessages();
- void setChatMessages(List chatMessages);
void cleanMessage(List messages);
Lock getMessageLock();
void addMetaMessage(String userId, MetaMessage metaMessage);
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCore.java b/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCore.java
index 1157c1a8..675a7e6a 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCore.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/core/cognation/CognationCore.java
@@ -63,11 +63,6 @@ public class CognationCore extends PartnerCore {
return currentMemoryId;
}
- @CapabilityMethod
- public void setChatMessages(List chatMessages) {
- this.chatMessages = chatMessages;
- }
-
@CapabilityMethod
public void cleanMessage(List messages) {
messageLock.lock();
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/core/memory/MemoryCore.java b/Partner-Main/src/main/java/work/slhaf/partner/core/memory/MemoryCore.java
index 46924c05..3bd4be03 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/core/memory/MemoryCore.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/core/memory/MemoryCore.java
@@ -198,9 +198,9 @@ public class MemoryCore extends PartnerCore {
//尝试更新缓存
updateCache(topicPath, memoryResult);
} catch (Exception e) {
- log.error("[CoordinatedManager] selectMemory error: ", e);
- log.error("[CoordinatedManager] 路径: {}", topicPathStr);
- log.error("[CoordinatedManager] 主题树: {}", getTopicTree());
+ log.error("[{}] selectMemory error: ", getCoreKey(), e);
+ log.error("[{}] 路径: {}", getCoreKey(), topicPathStr);
+ log.error("[{}] 主题树: {}", getCoreKey(), getTopicTree());
memoryResult = new MemoryResult();
memoryResult.setRelatedMemorySliceResult(new ArrayList<>());
memoryResult.setMemorySliceResult(new CopyOnWriteArrayList<>());
@@ -211,7 +211,7 @@ public class MemoryCore extends PartnerCore {
@CapabilityMethod
public void updateActivatedSlices(String userId, List memorySlices) {
cache.activatedSlices.put(userId, memorySlices);
- log.debug("[CoordinatedManager] 已更新激活切片, userId: {}", userId);
+ log.debug("[{}] 已更新激活切片, userId: {}", getCoreKey(), userId);
}
@CapabilityMethod
@@ -488,7 +488,7 @@ public class MemoryCore extends PartnerCore {
return targetParentNode;
}
- public void updateCacheCounter(List topicPath) {
+ private void updateCacheCounter(List topicPath) {
ConcurrentHashMap, Integer> memoryNodeCacheCounter = cache.memoryNodeCacheCounter;
if (memoryNodeCacheCounter.containsKey(topicPath)) {
Integer tempCount = memoryNodeCacheCounter.get(topicPath);
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/common/entity/AppendPromptData.java b/Partner-Main/src/main/java/work/slhaf/partner/module/common/entity/AppendPromptData.java
index 050158c0..8abd56b0 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/module/common/entity/AppendPromptData.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/common/entity/AppendPromptData.java
@@ -5,7 +5,7 @@ import lombok.EqualsAndHashCode;
import work.slhaf.partner.api.common.entity.PersistableObject;
import java.io.Serial;
-import java.util.HashMap;
+import java.util.Map;
@EqualsAndHashCode(callSuper = true)
@Data
@@ -15,5 +15,5 @@ public class AppendPromptData extends PersistableObject {
private static final long serialVersionUID = 1L;
private String moduleName;
- private HashMap appendedPrompt;
+ private Map appendedPrompt;
}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/common/module/PostRunningModule.java b/Partner-Main/src/main/java/work/slhaf/partner/module/common/module/PostRunningModule.java
index f3c5f01f..dacd83fb 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/module/common/module/PostRunningModule.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/common/module/PostRunningModule.java
@@ -8,7 +8,7 @@ public abstract class PostRunningModule extends AgentRunningModule map = getPromptDataMap(context);
+ Map map = getPromptDataMap(context);
data.setAppendedPrompt(map);
context.setAppendedPrompt(data);
}
@@ -24,7 +24,7 @@ public abstract class PreRunningModule extends AgentRunningModule getPromptDataMap(PartnerRunningFlowContext context);
+ protected abstract Map getPromptDataMap(PartnerRunningFlowContext context);
/**
* 用于在CoreModule接收到的模块Prompt中标识模块名称
@@ -40,5 +40,4 @@ public abstract class PreRunningModule extends AgentRunningModule {
+ String userId = context.getUserId();
+ val preparedActions = actionCapability.listActions(ActionData.ActionStatus.PREPARE, userId);
+ // 分类成PLANNING和IMMEDIATE两类
+ Set scheduledActions = new HashSet<>();
+ Set immediateActions = new HashSet<>();
+ for (ActionData preparedAction : preparedActions) {
+ if (preparedAction instanceof ScheduledActionData actionInfo) {
+ scheduledActions.add(actionInfo);
+ } else if (preparedAction instanceof ImmediateActionData actionInfo) {
+ immediateActions.add(actionInfo);
+ }
+ }
+ actionExecutor.execute(new ActionExecutorInput(immediateActions));
+ actionScheduler.execute(scheduledActions);
+ });
+ }
+
+ @Override
+ protected boolean relyOnMessage() {
+ return false;
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionCorrector.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionCorrector.java
new file mode 100644
index 00000000..481f851b
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionCorrector.java
@@ -0,0 +1,52 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor;
+
+import com.alibaba.fastjson2.JSONObject;
+import lombok.val;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.CorrectorInput;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.CorrectorResult;
+
+/**
+ * 负责在单组行动执行后，根据行动意图与结果检查后续行动是否符合目的，必要时直接调整行动链，或发起自对话请求进行干预
+ */
+@AgentSubModule
+public class ActionCorrector extends AgentRunningSubModule implements ActivateModel {
+
+ @Override
+ public CorrectorResult execute(CorrectorInput input) {
+ val prompt = buildPrompt(input);
+ val chatResponse = singleChat(prompt);
+ return JSONObject.parseObject(chatResponse.getMessage(), CorrectorResult.class);
+ }
+
+ private String buildPrompt(CorrectorInput input) {
+ val prompt = new JSONObject();
+ prompt.put("[行动来源]", input.getSource());
+ prompt.put("[行动倾向]", input.getTendency());
+ prompt.put("[行动描述]", input.getDescription());
+ prompt.put("[行动原因]", input.getReason());
+
+ val messages = prompt.putArray("[近期对话]");
+ messages.addAll(input.getRecentMessages());
+
+ val memory = prompt.putArray("[已激活记忆]");
+ memory.addAll(input.getActivatedSlices());
+
+ val history = prompt.putArray("[已执行情况]");
+ history.addAll(input.getHistory());
+ return prompt.toJSONString();
+ }
+
+ @Override
+ public String modelKey() {
+ return "action_corrector";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionExecutor.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionExecutor.java
new file mode 100644
index 00000000..28b46149
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionExecutor.java
@@ -0,0 +1,322 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor;
+
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.factory.module.annotation.Init;
+import work.slhaf.partner.api.agent.factory.module.annotation.InjectModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore;
+import work.slhaf.partner.core.action.entity.*;
+import work.slhaf.partner.core.action.entity.ActionData.ActionStatus;
+import work.slhaf.partner.core.action.runner.RunnerClient;
+import work.slhaf.partner.core.cognation.CognationCapability;
+import work.slhaf.partner.core.memory.MemoryCapability;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.*;
+import work.slhaf.partner.module.modules.action.dispatcher.scheduler.ActionScheduler;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Phaser;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Slf4j
+@AgentSubModule
+public class ActionExecutor extends AgentRunningSubModule {
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+ @InjectCapability
+ private MemoryCapability memoryCapability;
+ @InjectCapability
+ private CognationCapability cognationCapability;
+
+ @InjectModule
+ private ParamsExtractor paramsExtractor;
+ @InjectModule
+ private ActionRepairer actionRepairer;
+ @InjectModule
+ private ActionCorrector actionCorrector;
+ @InjectModule
+ private ActionScheduler actionScheduler;
+
+ private ExecutorService virtualExecutor;
+ private ExecutorService platformExecutor;
+ private RunnerClient runnerClient;
+
+ private final AssemblyHelper assemblyHelper = new AssemblyHelper();
+
+ @Init
+ public void init() {
+ virtualExecutor = actionCapability.getExecutor(ActionCore.ExecutorType.VIRTUAL);
+ platformExecutor = actionCapability.getExecutor(ActionCore.ExecutorType.PLATFORM);
+ runnerClient = actionCapability.runnerClient();
+ }
+
+ /**
+ * 执行行动
+ *
+ * @param input ActionExecutor 输入内容
+ * @return 无返回，执行结果回写至 input 内部携带的 actionData 中
+ */
+ @Override
+ public Void execute(ActionExecutorInput input) {
+ val actions = input.getActions();
+ // 异步执行所有行动
+ for (ActionData actionData : actions) {
+ platformExecutor.execute(() -> {
+ val source = actionData.getSource();
+ if (actionData.getStatus() != ActionStatus.PREPARE) {
+ return;
+ }
+ val actionChain = actionData.getActionChain();
+ if (actionChain.isEmpty()) {
+ actionData.setStatus(ActionStatus.FAILED);
+ actionData.setResult("行动链为空");
+ return;
+ }
+ // 注册执行中行动
+ val phaser = new Phaser();
+ val phaserRecord = actionCapability.putPhaserRecord(phaser, actionData);
+ actionData.setStatus(ActionStatus.EXECUTING);
+
+ // 开始执行
+ val stageCursor = new Object() {
+ int stageCount;
+ boolean executingStageUpdated;
+ boolean stageCountUpdated;
+
+ void init() {
+ stageCount = 0;
+ executingStageUpdated = false;
+ stageCountUpdated = false;
+ update();
+ }
+
+ void requestAdvance() {
+ if (!stageCountUpdated) {
+ stageCount++;
+ stageCountUpdated = true;
+ }
+
+ if (stageCount < actionChain.size() && !executingStageUpdated) {
+ update();
+ executingStageUpdated = true;
+ }
+ }
+
+ boolean next() {
+ executingStageUpdated = false;
+ stageCountUpdated = false;
+ return stageCount < actionChain.size();
+ }
+
+ void update() {
+ val orderList = new ArrayList<>(actionChain.keySet());
+ orderList.sort(Integer::compareTo);
+ actionData.setExecutingStage(orderList.get(stageCount));
+ }
+ };
+
+ stageCursor.init();
+ do {
+ val metaActions = actionChain.get(actionData.getExecutingStage());
+
+ val listeningRecord = executeAndListening(metaActions, phaserRecord, source);
+ phaser.awaitAdvance(listeningRecord.phase());
+
+ // synchronized 同步防止 accepting 循环间、phase guard 判定后发生 stage 推进
+ // 导致新行动的 phaser 投放阶段错乱无法阻塞的场景
+ // 该 synchronized 将阶段推进与 accepting 监听 loop 捆绑为互斥的原子事件，避免了细粒度的 phaser 阶段竞态问题
+ synchronized (listeningRecord.accepting()) {
+ listeningRecord.accepting().set(false);
+
+ // 立即尝试推进，本次推进中，如果前方仍有未执行 stage，将执行一次阶段推进
+ stageCursor.requestAdvance();
+ }
+
+ try {
+ // 针对行动链进行修正，修正需要传入执行历史、行动目标等内容
+ // 如果后续运行 corrector 触发频率较高，可考虑增加重试机制
+ val correctorInput = assemblyHelper.buildCorrectorInput(actionData, source);
+ val correctorResult = actionCorrector.execute(correctorInput);
+ actionCapability.handleInterventions(correctorResult.getMetaInterventionList(), actionData);
+ } catch (Exception ignored) {
+ }
+
+ // 第二次尝试进行阶段推进，本次负责补充上一次在不存在 stage时，但 corrector 执行期间发生了 actionChain 的插入事件
+ // 如果第一次已经推进完毕，本次将会跳过
+ stageCursor.requestAdvance();
+ } while (stageCursor.next());
+
+ // 结束
+ actionCapability.removePhaserRecord(phaser);
+ if (actionData.getStatus() != ActionStatus.FAILED) {
+ // 如果是 ScheduledActionData, 则重置 ActionData 内容,记录执行历史与最终结果
+ if (actionData instanceof ScheduledActionData scheduledActionData) {
+ scheduledActionData.recordAndReset();
+ actionScheduler.execute(Set.of(scheduledActionData));
+ } else {
+ actionData.setStatus(ActionStatus.SUCCESS);
+ }
+ }
+ });
+
+ }
+ return null;
+
+ }
+
+ private MetaActionsListeningRecord executeAndListening(List metaActions, PhaserRecord phaserRecord, String source) {
+ AtomicBoolean accepting = new AtomicBoolean(true);
+ AtomicInteger cursor = new AtomicInteger();
+
+ CountDownLatch latch = new CountDownLatch(1);
+ val phaser = phaserRecord.phaser();
+ val phase = phaser.register();
+ platformExecutor.execute(() -> {
+ boolean first = true;
+ while (accepting.get()) {
+ synchronized (accepting) {
+ MetaAction next = null;
+
+ synchronized (metaActions) {
+ if (cursor.get() < metaActions.size()) {
+ next = metaActions.get(cursor.getAndIncrement());
+ }
+ }
+
+ if (next == null) {
+ Thread.onSpinWait();
+ continue;
+ }
+
+ if (phaser.getPhase() != phase) {
+ metaActions.remove(next);
+ log.warn("行动阶段已推进，丢弃该行动: {}", next);
+ continue;
+ }
+
+ ExecutorService executor = next.getIo() ? virtualExecutor : platformExecutor;
+ executor.execute(buildMataActionTask(next, phaserRecord, source));
+
+ if (first) {
+ phaser.arriveAndDeregister();
+ latch.countDown();
+ first = false;
+ }
+ }
+ }
+ });
+ try {
+ // 确保执行一次，防止没来得及注册任务就已经结束
+ latch.await();
+ } catch (InterruptedException ignored) {
+ }
+ return new MetaActionsListeningRecord(accepting, phase);
+ }
+
+ private Runnable buildMataActionTask(MetaAction metaAction, PhaserRecord phaserRecord, String source) {
+ val phaser = phaserRecord.phaser();
+ phaser.register();
+ return () -> {
+ val actionKey = metaAction.getKey();
+ try {
+ val result = metaAction.getResult();
+ do {
+ val actionData = phaserRecord.actionData();
+ val executingStage = actionData.getExecutingStage();
+ val historyActionResults = actionData.getHistory().get(executingStage);
+ val additionalContext = actionData.getAdditionalContext().get(executingStage);
+ val extractorInput = assemblyHelper.buildExtractorInput(metaAction, source, historyActionResults, additionalContext);
+ val extractorResult = paramsExtractor.execute(extractorInput);
+
+ if (extractorResult.isOk()) {
+ metaAction.getParams().putAll(extractorResult.getParams());
+ runnerClient.submit(metaAction);
+ val historyAction = new HistoryAction(actionKey, actionCapability.loadMetaActionInfo(actionKey).getDescription(), metaAction.getResult().getData());
+ actionData.getHistory()
+ .computeIfAbsent(executingStage, integer -> new ArrayList<>())
+ .add(historyAction);
+ } else {
+ val repairerInput = assemblyHelper.buildRepairerInput(historyActionResults, metaAction, source);
+ val repairerResult = actionRepairer.execute(repairerInput);
+ switch (repairerResult.getStatus()) {
+ // 如果本次修复被认为成功，则将补充的信息添加至 additionalContext
+ case RepairerResult.RepairerStatus.OK -> {
+ additionalContext.addAll(repairerResult.getFixedData());
+ result.setStatus(MetaAction.Result.Status.WAITING);
+ }
+ // 此处的修复失败来自系统内部的执行失败：其余方式均不可行时将回退至当前分支
+ case RepairerResult.RepairerStatus.FAILED -> {
+ result.setStatus(MetaAction.Result.Status.FAILED);
+ result.setData("行动执行失败");
+ }
+ // 此处对应已在 repairer 内发起外部请求，故在此处进行阻塞
+ case RepairerResult.RepairerStatus.ACQUIRE -> {
+ phaserRecord.interrupt();
+ result.setStatus(MetaAction.Result.Status.WAITING);
+ }
+ }
+ }
+ } while (result.getStatus().equals(MetaAction.Result.Status.WAITING));
+ } catch (Exception e) {
+ log.error("Action executing failed: {}", actionKey, e);
+ } finally {
+ phaser.arriveAndDeregister();
+ }
+ };
+ }
+
+ private record MetaActionsListeningRecord(AtomicBoolean accepting, int phase) {
+ }
+
+ @SuppressWarnings("InnerClassMayBeStatic")
+ private class AssemblyHelper {
+
+ private AssemblyHelper() {
+ }
+
+ private RepairerInput buildRepairerInput(List historyActionsResults, MetaAction action, String userId) {
+ RepairerInput input = new RepairerInput();
+ MetaActionInfo metaActionInfo = actionCapability.loadMetaActionInfo(action.getKey());
+ input.setHistoryActionResults(historyActionsResults);
+ input.setParams(metaActionInfo.getParams());
+ input.setRecentMessages(cognationCapability.getChatMessages());
+ input.setActionDescription(metaActionInfo.getDescription());
+ input.setUserId(userId);
+ return input;
+ }
+
+ private ExtractorInput buildExtractorInput(MetaAction action, String source, List historyActionResults,
+ List additionalContext) {
+ ExtractorInput input = new ExtractorInput();
+ input.setEvaluatedSlices(memoryCapability.getActivatedSlices(source));
+ input.setRecentMessages(cognationCapability.getChatMessages());
+ input.setMetaActionInfo(actionCapability.loadMetaActionInfo(action.getKey()));
+ input.setHistoryActionResults(historyActionResults);
+ input.setAdditionalContext(additionalContext);
+ return input;
+ }
+
+ private CorrectorInput buildCorrectorInput(ActionData actionData, String source) {
+ return CorrectorInput.builder()
+ .tendency(actionData.getTendency())
+ .source(actionData.getSource())
+ .reason(actionData.getReason())
+ .description(actionData.getDescription())
+ .history(actionData.getHistory().get(actionData.getExecutingStage()))
+ .status(actionData.getStatus())
+ .recentMessages(cognationCapability.getChatMessages())
+ .activatedSlices(memoryCapability.getActivatedSlices(source))
+ .build();
+ }
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionRepairer.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionRepairer.java
new file mode 100644
index 00000000..c5496b8a
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ActionRepairer.java
@@ -0,0 +1,228 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import com.alibaba.fastjson2.TypeReference;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.factory.module.annotation.Init;
+import work.slhaf.partner.api.agent.factory.module.annotation.InjectModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.api.chat.pojo.ChatResponse;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore.ExecutorType;
+import work.slhaf.partner.core.action.entity.MetaAction;
+import work.slhaf.partner.core.action.entity.MetaAction.Result;
+import work.slhaf.partner.core.action.runner.RunnerClient;
+import work.slhaf.partner.core.cognation.CognationCapability;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.GeneratorInput;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.GeneratorResult;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.RepairerInput;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.RepairerResult;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.RepairerResult.RepairerStatus;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 负责识别行动链的修复
+ *
+ * -
+ * 可通过协调 {@link DynamicActionGenerator} 生成新的行动单元并调用，获取所需的参数信息（必要时持久化）;
+ *
+ * -
+ * 也可以直接调用已存在的行动程序获取信息;
+ *
+ * -
+ * 如果上述都无法满足，将发起自对话借助干预模块进行操作或者借助自对话通道向用户发起沟通请求，该请求的目的一般为行动程序生成/调用指导或者用户侧的信息补充，后续还需要再走一遍参数修复流程
+ *
+ *
+ */
+@Slf4j
+@AgentSubModule
+public class ActionRepairer extends AgentRunningSubModule implements ActivateModel {
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+ @InjectCapability
+ private CognationCapability cognationCapability;
+
+ @InjectModule
+ private DynamicActionGenerator dynamicActionGenerator;
+
+ private final AssemblyHelper assemblyHelper = new AssemblyHelper();
+ private RunnerClient runnerClient;
+
+ @Init
+ void init() {
+ runnerClient = actionCapability.runnerClient();
+ }
+
+ @Override
+ public RepairerResult execute(RepairerInput data) {
+ RepairerResult result;
+ try {
+ String prompt = assemblyHelper.buildPrompt(data, null);
+ ChatResponse response = this.singleChat(prompt);
+ RepairerData repairerData = JSONObject.parseObject(response.getMessage(), RepairerData.class);
+ result = switch (repairerData.getRepairerType()) {
+ case ACTION_GENERATION ->
+ handleActionGeneration(JSONObject.parseObject(repairerData.getData(), GeneratorInput.class));
+ case ACTION_INVOCATION -> handleActionInvocation(
+ JSONObject.parseObject(repairerData.getData(), new TypeReference>() {
+ }));
+ case USER_INTERACTION -> handleUserInteraction(repairerData.getData());
+ };
+ if (!repairerData.getRepairerType().equals(RepairerType.USER_INTERACTION)
+ && result.getStatus().equals(RepairerResult.RepairerStatus.FAILED)) {
+ log.warn("常规行动修复失败，将尝试自对话通道");
+ prompt = assemblyHelper.buildPrompt(data, "常规行动修复失败，请尝试通过自对话通道获取必要的信息以完成行动参数的修复");
+ response = this.singleChat(prompt);
+ repairerData = JSONObject.parseObject(response.getMessage(), RepairerData.class);
+ handleUserInteraction(repairerData.getData());
+ }
+ } catch (Exception e) {
+ result = new RepairerResult();
+ result.setStatus(RepairerStatus.FAILED);
+ }
+ return result;
+ }
+
+ /**
+ * 负责根据输入内容进行行动单元的参数信息修复
+ *
+ * @param generatorInput 生成的行动单元参考内容，最好包含行动单元的执行逻辑
+ * @return 修复后的行动单元结果
+ */
+ private RepairerResult handleActionGeneration(GeneratorInput generatorInput) {
+ RepairerResult result = new RepairerResult();
+ GeneratorResult generatorResult = dynamicActionGenerator.execute(generatorInput);
+ MetaAction tempAction = generatorResult.getTempAction();
+ if (tempAction == null) {
+ result.setStatus(RepairerStatus.FAILED);
+ return result;
+ }
+
+ runnerClient.submit(tempAction);
+ // 根据 tempAction 的执行状态设置修复结果
+ Result actionResult = tempAction.getResult();
+ if (actionResult.getStatus() != MetaAction.Result.Status.SUCCESS) {
+ result.setStatus(RepairerStatus.FAILED);
+ return result;
+ }
+
+ result.setStatus(RepairerStatus.OK);
+ result.getFixedData().add(actionResult.getData());
+ return result;
+ }
+
+ /**
+ * 负责根据输入内容进行行动单元的参数信息修复
+ *
+ * @param actionKeys 需要调用的行动单元Key列表
+ * @return 修复后的行动单元结果
+ */
+ private RepairerResult handleActionInvocation(List actionKeys) {
+ RepairerResult result = new RepairerResult();
+ CountDownLatch latch = new CountDownLatch(actionKeys.size());
+ ExecutorService virtual = actionCapability.getExecutor(ExecutorType.VIRTUAL);
+ ExecutorService platform = actionCapability.getExecutor(ExecutorType.PLATFORM);
+ ExecutorService executor;
+ AtomicInteger failedCount = new AtomicInteger(0);
+ for (String key : actionKeys) {
+ MetaAction action = actionCapability.loadMetaAction(key);
+ executor = action.getIo() ? virtual : platform;
+ executor.execute(() -> {
+ try {
+ runnerClient.submit(action);
+ result.getFixedData().add(action.getResult().getData());
+ } catch (Exception e) {
+ log.error("行动单元执行失败: {}", key, e);
+ failedCount.incrementAndGet();
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+ try {
+ latch.await();
+ } catch (Exception e) {
+ log.warn("CountDownLatch 已中断");
+ }
+ if (actionKeys.size() - failedCount.get() > 0) {
+ result.setStatus(RepairerStatus.OK);
+ } else {
+ result.setStatus(RepairerStatus.FAILED);
+ }
+ return result;
+ }
+
+ private RepairerResult handleUserInteraction(String acquireContent) {
+ RepairerResult result = new RepairerResult();
+ result.setStatus(RepairerStatus.ACQUIRE);
+ // 发送自对话请求
+ return result;
+ }
+
+ @Override
+ public String modelKey() {
+ return "action_repairer";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+
+ @SuppressWarnings("InnerClassMayBeStatic")
+ @Data
+ private class RepairerData {
+ private RepairerType repairerType;
+ private String data;
+ }
+
+ private enum RepairerType {
+ ACTION_GENERATION,
+ ACTION_INVOCATION,
+ USER_INTERACTION
+ }
+
+ @SuppressWarnings("InnerClassMayBeStatic")
+ private class AssemblyHelper {
+ private AssemblyHelper() {
+ }
+
+ private String buildPrompt(RepairerInput data, String specialInstruction) {
+ JSONObject prompt = new JSONObject();
+
+ JSONObject actionData = prompt.putObject("[本次行动信息]");
+ actionData.put("[行动描述]", data.getActionDescription());
+ JSONObject actionParamsData = actionData.putObject("[行动参数说明]");
+ actionParamsData.putAll(data.getParams());
+
+ JSONArray historyData = prompt.putArray("[历史行动执行结果]");
+ data.getHistoryActionResults().forEach(historyAction -> {
+ JSONObject historyItem = new JSONObject();
+ historyItem.put("[行动Key]", historyAction.actionKey());
+ historyItem.put("[行动描述]", historyAction.description());
+ historyItem.put("[行动结果]", historyAction.result());
+ historyData.add(historyItem);
+ });
+
+ JSONArray messageData = prompt.putArray("[最近消息列表]");
+ messageData.addAll(data.getRecentMessages());
+
+ if (specialInstruction != null) {
+ prompt.put("[特殊指令]", specialInstruction);
+ }
+
+ return prompt.toString();
+ }
+
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/DynamicActionGenerator.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/DynamicActionGenerator.java
new file mode 100644
index 00000000..e366198b
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/DynamicActionGenerator.java
@@ -0,0 +1,91 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor;
+
+import com.alibaba.fastjson2.JSONObject;
+import lombok.val;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.factory.module.annotation.Init;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.api.chat.pojo.ChatResponse;
+import work.slhaf.partner.common.util.ExtractUtil;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.entity.GeneratedData;
+import work.slhaf.partner.core.action.entity.MetaAction;
+import work.slhaf.partner.core.action.runner.RunnerClient;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.GeneratorInput;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.GeneratorResult;
+
+/**
+ * 负责依据输入内容生成可执行的动态行动单元，并选择是否持久化至 SandboxRunner 容器内
+ */
+@AgentSubModule
+public class DynamicActionGenerator extends AgentRunningSubModule
+ implements ActivateModel {
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+
+ private RunnerClient runnerClient;
+
+ @Init
+ void init() {
+ runnerClient = actionCapability.runnerClient();
+ }
+
+ @Override
+ public GeneratorResult execute(GeneratorInput input) {
+ GeneratorResult result = new GeneratorResult();
+ try {
+ // 由于 SCRIPT 类型程序都是在 SandboxRunner 内部的磁盘上加载然后执行的，
+ // 所以此处的输入内容也只需要指定输入参数、临时key、是否持久化即可，路径将按照指定规则统一构建，不可交给LLM生成
+ String prompt = buildPrompt(input);
+
+ // 响应结果需要包含几个特殊数据: 依赖项、代码内容、是否序列化、响应数据释义
+ ChatResponse response = this.singleChat(prompt);
+ GeneratedData generatorData = JSONObject
+ .parseObject(ExtractUtil.extractJson(response.getMessage()), GeneratedData.class);
+
+ val location = runnerClient.buildTmpPath(input.getActionName(), generatorData.getCodeType());
+ MetaAction tempAction = new MetaAction(
+ input.getActionName(),
+ true,
+ MetaAction.Type.ORIGIN,
+ location
+ );
+ // 将临时行动单元序列化至临时文件夹，并设置程序路径、放置在队列中，等待执行状态变化，并根据序列化选项选择是否补充 MetaActionInfo 并持久序列化
+ // 通过 ActionCapability 暴露的接口，序列化至临时文件夹，同时返回Path对象并设置。队列建议交给 SandboxRunner
+ // 持有，包括监听与序列化线程
+ runnerClient.tmpSerialize(tempAction, generatorData.getCode(), generatorData.getCodeType());
+ if (generatorData.isSerialize()) {
+ waitingSerialize();
+ }
+ result.setTempAction(tempAction);
+ } catch (Exception e) {
+ result.setTempAction(null);
+ }
+ return result;
+ }
+
+ private void waitingSerialize() {
+ throw new UnsupportedOperationException("Unimplemented method 'waitingSerialize'");
+ }
+
+ private String buildPrompt(GeneratorInput data) {
+ JSONObject prompt = new JSONObject();
+ prompt.put("[行动描述]", data.getDescription());
+ // prompt.putObject("[行动参数]").putAll(data.getParams());
+ prompt.putObject("[行动参数描述]").putAll(data.getParamsDescription());
+ return prompt.toString();
+ }
+
+ @Override
+ public String modelKey() {
+ return "dynamic_generator";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ParamsExtractor.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ParamsExtractor.java
new file mode 100644
index 00000000..9caf2686
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/ParamsExtractor.java
@@ -0,0 +1,75 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.api.chat.pojo.ChatResponse;
+import work.slhaf.partner.core.action.entity.MetaActionInfo;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.ExtractorInput;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.ExtractorResult;
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.HistoryAction;
+
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * 负责依据输入内容进行行动单元的参数信息提取
+ */
+@Slf4j
+@AgentSubModule
+public class ParamsExtractor extends AgentRunningSubModule implements ActivateModel {
+
+ @Override
+ public ExtractorResult execute(ExtractorInput input) {
+ String prompt = buildPrompt(input);
+ ChatResponse response = this.singleChat(prompt);
+ ExtractorResult result;
+ try {
+ result = JSONObject.parseObject(response.getMessage(), ExtractorResult.class);
+ } catch (Exception e) {
+ log.error("ParamsExtractor解析结果失败，返回内容：{}", response.getMessage(), e);
+ result = new ExtractorResult();
+ result.setOk(false);
+ result.setParams(new HashMap<>());
+ }
+ return result;
+ }
+
+ private String buildPrompt(ExtractorInput input) {
+ JSONObject prompt = new JSONObject();
+
+ JSONObject actionData = prompt.putObject("[本次行动信息]");
+ MetaActionInfo actionInfo = input.getMetaActionInfo();
+ actionData.put("[行动描述]", actionInfo.getDescription());
+ actionData.put("[行动参数说明]", actionInfo.getParams());
+
+ JSONArray historyData = prompt.putArray("[历史行动执行结果]");
+ List historyActions = input.getHistoryActionResults();
+ for (HistoryAction historyAction : historyActions) {
+ JSONObject historyItem = new JSONObject();
+ historyItem.put("[行动Key]", historyAction.actionKey());
+ historyItem.put("[行动描述]", historyAction.description());
+ historyItem.put("[行动结果]", historyAction.result());
+ historyData.add(historyItem);
+ }
+
+ JSONArray messageData = prompt.putArray("[最近消息列表]");
+ messageData.addAll(input.getRecentMessages());
+
+ return prompt.toString();
+ }
+
+ @Override
+ public String modelKey() {
+ return "params_extractor";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ActionExecutorInput.kt b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ActionExecutorInput.kt
new file mode 100644
index 00000000..f21db99d
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ActionExecutorInput.kt
@@ -0,0 +1,5 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity
+
+import work.slhaf.partner.core.action.entity.ActionData
+
+data class ActionExecutorInput(val actions: Set)
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorInput.java
new file mode 100644
index 00000000..1c2c9af8
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorInput.java
@@ -0,0 +1,24 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Builder;
+import lombok.Data;
+import work.slhaf.partner.api.chat.pojo.Message;
+import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.memory.pojo.EvaluatedSlice;
+
+import java.util.List;
+
+@Data
+@Builder
+public class CorrectorInput {
+ private String tendency;
+ private String source;
+ private String reason;
+ private String description;
+
+ private List history;
+ private ActionData.ActionStatus status;
+
+ private List recentMessages;
+ private List activatedSlices;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorResult.java
new file mode 100644
index 00000000..a2c267e2
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/CorrectorResult.java
@@ -0,0 +1,11 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+import work.slhaf.partner.module.modules.action.interventor.entity.MetaIntervention;
+
+import java.util.List;
+
+@Data
+public class CorrectorResult {
+ private List metaInterventionList;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorInput.java
new file mode 100644
index 00000000..4a99b457
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorInput.java
@@ -0,0 +1,32 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+import work.slhaf.partner.api.chat.pojo.Message;
+import work.slhaf.partner.core.action.entity.MetaActionInfo;
+import work.slhaf.partner.core.memory.pojo.EvaluatedSlice;
+
+import java.util.List;
+
+@Data
+public class ExtractorInput {
+ /**
+ * 目标 MetaActionInfo
+ */
+ private MetaActionInfo metaActionInfo;
+ /**
+ * 可参考的记忆切片
+ */
+ private List evaluatedSlices;
+ /**
+ * 历史行动执行结果
+ */
+ private List historyActionResults;
+ /**
+ * 最近的消息列表
+ */
+ private List recentMessages;
+ /**
+ * 额外的上下文信息（可来自修复器等）
+ */
+ private List additionalContext;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorResult.java
new file mode 100644
index 00000000..aefb5ba1
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/ExtractorResult.java
@@ -0,0 +1,11 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+public class ExtractorResult {
+ private boolean ok;
+ private Map params;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorInput.java
new file mode 100644
index 00000000..67c9094a
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorInput.java
@@ -0,0 +1,13 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+
+import java.util.Map;
+
+@Data
+public class GeneratorInput {
+ private String actionName;
+ private Map params;
+ private String description;
+ private Map paramsDescription;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorResult.java
new file mode 100644
index 00000000..0b8572fb
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/GeneratorResult.java
@@ -0,0 +1,9 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+import work.slhaf.partner.core.action.entity.MetaAction;
+
+@Data
+public class GeneratorResult {
+ private MetaAction tempAction;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/HistoryAction.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/HistoryAction.java
new file mode 100644
index 00000000..cf36f1ed
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/HistoryAction.java
@@ -0,0 +1,4 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+public record HistoryAction(String actionKey, String description, String result) {
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerInput.java
new file mode 100644
index 00000000..dd470a13
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerInput.java
@@ -0,0 +1,17 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+import work.slhaf.partner.api.chat.pojo.Message;
+
+import java.util.List;
+import java.util.Map;
+
+@Data
+public class RepairerInput {
+
+ private String userId;
+ private List recentMessages;
+ private Map params;
+ private String actionDescription;
+ private List historyActionResults;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerResult.java
new file mode 100644
index 00000000..c6e2c999
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/entity/RepairerResult.java
@@ -0,0 +1,30 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.entity;
+
+import lombok.Data;
+
+import java.util.List;
+
+/**
+ * 行动修复结果，包含行动状态和修复后的参数
+ */
+@Data
+public class RepairerResult {
+
+ private RepairerStatus status;
+ private List fixedData;
+
+ public enum RepairerStatus {
+ /**
+ * 成功修复: 携带修复后参数; 此种情况对应 Repairer 通过某种方式获取到了完整的参数(调用额外的行动)
+ */
+ OK,
+ /**
+ * 发送了自对话请求干预行动，这类一般是补充信息或者提供行动指导，后续必须再步入修复进程，但需要设置层级
+ */
+ ACQUIRE,
+ /**
+ * 修复失败(简单修复、自对话通道均出现错误，正常情况不应该出现)
+ */
+ FAILED
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/exception/ActionExecutingFailedException.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/exception/ActionExecutingFailedException.java
new file mode 100644
index 00000000..f10273d1
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/executor/exception/ActionExecutingFailedException.java
@@ -0,0 +1,14 @@
+package work.slhaf.partner.module.modules.action.dispatcher.executor.exception;
+
+import work.slhaf.partner.api.agent.runtime.exception.AgentRuntimeException;
+
+public class ActionExecutingFailedException extends AgentRuntimeException {
+
+ public ActionExecutingFailedException(String message) {
+ super(message);
+ }
+
+ public ActionExecutingFailedException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/scheduler/ActionScheduler.kt b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/scheduler/ActionScheduler.kt
new file mode 100644
index 00000000..2c867e4a
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/dispatcher/scheduler/ActionScheduler.kt
@@ -0,0 +1,400 @@
+package work.slhaf.partner.module.modules.action.dispatcher.scheduler
+
+import com.cronutils.model.CronType
+import com.cronutils.model.definition.CronDefinition
+import com.cronutils.model.definition.CronDefinitionBuilder
+import com.cronutils.model.time.ExecutionTime
+import com.cronutils.parser.CronParser
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import org.slf4j.LoggerFactory
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule
+import work.slhaf.partner.api.agent.factory.module.annotation.Init
+import work.slhaf.partner.api.agent.factory.module.annotation.InjectModule
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule
+import work.slhaf.partner.core.action.ActionCapability
+import work.slhaf.partner.core.action.entity.ActionData
+import work.slhaf.partner.core.action.entity.ScheduledActionData
+import work.slhaf.partner.module.modules.action.dispatcher.executor.ActionExecutor
+import work.slhaf.partner.module.modules.action.dispatcher.executor.entity.ActionExecutorInput
+import java.io.Closeable
+import java.time.Duration
+import java.time.ZonedDateTime
+import java.time.temporal.ChronoUnit
+import java.util.stream.Collectors
+import kotlin.jvm.optionals.getOrNull
+
+@AgentSubModule
+class ActionScheduler : AgentRunningSubModule, Void>() {
+
+ @InjectCapability
+ private lateinit var actionCapability: ActionCapability
+
+ @InjectModule
+ private lateinit var actionExecutor: ActionExecutor
+
+ private lateinit var timeWheel: TimeWheel
+
+ private val schedulerScope =
+ CoroutineScope(Dispatchers.Default + SupervisorJob() + CoroutineName("ActionScheduler"))
+
+ companion object {
+ private val log = LoggerFactory.getLogger(ActionScheduler::class.java)
+ }
+
+ @Init
+ fun init() {
+ val listScheduledActions: () -> Set = {
+ actionCapability.listActions(null, null)
+ .stream()
+ .filter { it is ScheduledActionData }
+ .map { it as ScheduledActionData }
+ .collect(Collectors.toSet())
+ }
+
+ val onTrigger: (Set) -> Unit = { actionExecutor.execute(ActionExecutorInput(it)) }
+
+ timeWheel = TimeWheel(listScheduledActions, onTrigger)
+
+ setupShutdownHook()
+ }
+
+ private fun setupShutdownHook() {
+ Runtime.getRuntime().addShutdownHook(Thread {
+ timeWheel.close()
+ schedulerScope.cancel()
+ })
+ }
+
+ override fun execute(scheduledActionDataSet: Set?): Void? {
+ schedulerScope.launch {
+ scheduledActionDataSet?.run {
+ for (scheduledActionData in scheduledActionDataSet) {
+ log.debug("New action to schedule: {}", scheduledActionData)
+ actionCapability.putAction(scheduledActionData)
+ timeWheel.schedule(scheduledActionData)
+ }
+ }
+ }
+ return null
+ }
+
+ private class TimeWheel(
+ val listScheduledActions: () -> Set,
+ val onTrigger: (toTrigger: Set) -> Unit
+ ) : Closeable {
+
+ private val actionsGroupByHour = Array>(24) { mutableSetOf() }
+ private val wheel = Array>(60 * 60) { mutableSetOf() }
+ private var recordHour: Int = -1
+ private var recordDay: Int = -1
+ private val state = MutableStateFlow(WheelState.SLEEPING)
+
+ private val wheelActionsLock = Mutex()
+ private val timeWheelScope = CoroutineScope(SupervisorJob() + Dispatchers.Default + CoroutineName("TimeWheel"))
+
+ private val cronDefinition: CronDefinition = CronDefinitionBuilder.instanceDefinitionFor(CronType.QUARTZ)
+ private val cronParser: CronParser = CronParser(cronDefinition)
+
+ /**
+ * 根据 primaryActions 建立时间轮，并只加载当天任务，同时启动 tick 线程
+ */
+ init {
+ // 启动时间轮
+ launchWheel()
+ }
+
+ suspend fun schedule(actionData: ScheduledActionData) {
+ if (actionData.status != ActionData.ActionStatus.PREPARE) {
+ return
+ }
+
+ checkThenExecute {
+ val parseToZonedDateTime = parseToZonedDateTime(
+ actionData.scheduleType,
+ actionData.scheduleContent,
+ it
+ ) ?: run {
+ logFailedStatus(actionData)
+ return@checkThenExecute
+ }
+ log.debug("Action next execution time: {}", parseToZonedDateTime)
+
+ val hour = parseToZonedDateTime.hour
+ actionsGroupByHour[hour].add(actionData)
+ log.debug("Action scheduled at {}", hour)
+
+ if (it.hour == hour) {
+ val wheelOffset = parseToZonedDateTime.minute * 60 + parseToZonedDateTime.second
+ wheel[wheelOffset].add(actionData)
+ state.value = WheelState.ACTIVE
+ log.debug("Action scheduled at wheel offset {}", wheelOffset)
+ }
+
+ }
+ }
+
+ private fun launchWheel() {
+
+ fun collectToTrigger(tick: Int, previousTick: Int, triggerHour: Int): Set? {
+ if (tick > previousTick) {
+ val toTrigger = mutableSetOf()
+ for (i in previousTick..tick) {
+ val bucket = wheel[i]
+ if (bucket.isNotEmpty()) {
+ toTrigger.addAll(bucket)
+ val bucketUuids = bucket.asSequence().map { it.uuid }.toHashSet()
+ actionsGroupByHour[triggerHour].removeIf { it.uuid in bucketUuids }
+ bucket.clear() // 避免重复触发
+ }
+ }
+ return toTrigger
+ }
+ return null
+ }
+
+ suspend fun CoroutineScope.wheel(launchingTime: ZonedDateTime, primaryTickAdvanceTime: Long) {
+ val launchingHour = launchingTime.hour
+ var tick = launchingTime.minute * 60 + launchingTime.second
+
+ // 让节拍器从“启动时刻的下一秒”开始（避免立即 step=0）
+ var nextTickNanos = primaryTickAdvanceTime + 1_000_000_000L
+
+ while (isActive) {
+ // 1) 计算落后多少秒：至少 1（正常推进），也可能 >1（追赶）
+ val now0 = System.nanoTime()
+ val lagNanos = now0 - nextTickNanos
+ val step = if (lagNanos < 0) 1 else (lagNanos / 1_000_000_000L).toInt() + 1
+
+ val previousTick = tick
+ tick = (tick + step).coerceAtMost(wheel.lastIndex)
+
+ // 2) 推进节拍器：按“理论秒”前进 step 次
+ nextTickNanos += step.toLong() * 1_000_000_000L
+
+ var shouldBreak = false
+ var toTrigger: Set? = null
+
+ checkThenExecute(false) {
+ if (it.hour != launchingHour) {
+ shouldBreak = true
+ toTrigger = collectToTrigger(wheel.lastIndex, previousTick, launchingHour)
+ log.debug(
+ "Hour changed, previousTick: {}, tick: {}, toTriggerSize: {}",
+ previousTick,
+ tick,
+ toTrigger?.size
+ )
+ return@checkThenExecute
+ }
+
+ toTrigger = collectToTrigger(tick, previousTick, launchingHour)
+
+ if (tick >= wheel.lastIndex || actionsGroupByHour[launchingHour].isEmpty()) {
+ state.value = WheelState.SLEEPING
+ shouldBreak = true
+ return@checkThenExecute
+ }
+ }
+
+ toTrigger?.takeIf { it.isNotEmpty() }?.let {
+ onTrigger(it)
+ log.debug("Executing action at hour {} tick {}", launchingHour, tick)
+ }
+
+ if (shouldBreak) {
+ log.debug("Wheel stopped at tick {}", tick)
+ break
+ }
+
+ // 3) 精确睡到下一次理论 tick（用最新 nanoTime）
+ val now1 = System.nanoTime()
+ val sleepNanos = nextTickNanos - now1
+ if (sleepNanos > 0) {
+ delay(sleepNanos / 1_000_000L) // 毫秒级 delay 足够；剩余 nanos 不必忙等
+ }
+ }
+ }
+
+ suspend fun wait(currentTime: ZonedDateTime) {
+ val nextHour = currentTime.truncatedTo(ChronoUnit.HOURS).plusHours(1)
+ val seconds = Duration.between(
+ currentTime, nextHour
+ ).toMillis()
+ // withTimeoutOrNull 内部已处理 seconds 小于 0 的情况
+ log.debug("Start waiting {} ms at {}, target time: {}", seconds, currentTime, nextHour)
+ withTimeoutOrNull(seconds) {
+ state.first { it == WheelState.ACTIVE }
+ }
+ log.debug("Waiting ended at {}", ZonedDateTime.now())
+ }
+
+ timeWheelScope.launch {
+ while (isActive) {
+ // 判断是否该步入下一小时
+ var shouldWait: Boolean? = null
+ var currentTime: ZonedDateTime? = null
+ var primaryTickAdvanceTime: Long? = null
+ checkThenExecute {
+ currentTime = it
+ shouldWait = actionsGroupByHour[it.hour].isEmpty()
+ // 由于 wheel 的启动时间可能存在延迟，而时内推进由 nanoTime 保证不会漏发，
+ // 正常的时序结束又由 tick 是否触顶、当前时是否存在额外任务触发，
+ // 而启动时无触发保障，此时一并初始化 tick 推进时间，足以应对 check 与 wheel 间的这段时间间隔
+ primaryTickAdvanceTime = System.nanoTime()
+ }
+
+ // 如果该时无任务则等待，插入事件可提前唤醒
+ if (shouldWait!!) {
+ // 计算距离下一小时的时间，等待
+ currentTime?.let { wait(it) }
+ continue
+ }
+
+ // 唤醒进行时间轮循环
+ wheel(currentTime!!, primaryTickAdvanceTime!!)
+ }
+ }
+ }
+
+ suspend fun checkThenExecute(finallyToExecute: Boolean = true, then: (currentTime: ZonedDateTime) -> Unit) =
+ wheelActionsLock.withLock {
+ fun loadActions(
+ source: Set,
+ now: ZonedDateTime,
+ load: (latestExecutingTime: ZonedDateTime, actionData: ScheduledActionData) -> Unit,
+ repair: () -> Unit
+ ) {
+ val runLoading = {
+ for (actionData in source) {
+ val nextExecutingTime =
+ parseToZonedDateTime(
+ actionData.scheduleType,
+ actionData.scheduleContent,
+ now
+ ) ?: run {
+ logFailedStatus(actionData)
+ continue
+ }
+
+ load(nextExecutingTime, actionData)
+ }
+ }
+
+ repair()
+ runLoading()
+ }
+
+ fun loadHourActions(currentTime: ZonedDateTime) {
+ val load: (ZonedDateTime, ScheduledActionData) -> Unit = { latestExecutionTime, actionData ->
+ val secondsTime = latestExecutionTime.minute * 60 + latestExecutionTime.second
+ wheel[secondsTime].add(actionData)
+ log.debug("Action loaded to hour: {}", actionData)
+ }
+
+ val repair: () -> Unit = {
+ for (set in wheel) {
+ set.clear()
+ }
+ }
+
+ loadActions(actionsGroupByHour[currentTime.hour], currentTime, load, repair)
+ }
+
+ fun loadDayActions(currentTime: ZonedDateTime) {
+ val load: (ZonedDateTime, ScheduledActionData) -> Unit = { latestExecutingTime, actionData ->
+ actionsGroupByHour[latestExecutingTime.hour].add(actionData)
+ log.debug("Action loaded to day: {}", actionData)
+ }
+
+ val repair: () -> Unit = {
+ for (set in actionsGroupByHour) {
+ set.clear()
+ }
+ }
+
+ loadActions(listScheduledActions(), currentTime, load, repair)
+ }
+
+ fun refreshIfNeeded(now: ZonedDateTime) {
+ val d = now.dayOfMonth
+ val h = now.hour
+ if (d != recordDay) {
+ recordDay = d
+ recordHour = h
+ loadDayActions(now)
+ loadHourActions(now)
+ } else if (h != recordHour) {
+ recordHour = h
+ loadHourActions(now)
+ }
+ }
+
+ val now = ZonedDateTime.now()
+
+ if (finallyToExecute) {
+ refreshIfNeeded(now)
+ then(now)
+ } else {
+ then(now)
+ refreshIfNeeded(now)
+ }
+ }
+
+ private fun parseToZonedDateTime(
+ scheduleType: ScheduledActionData.ScheduleType,
+ scheduleContent: String,
+ now: ZonedDateTime
+ ): ZonedDateTime? {
+ return when (scheduleType) {
+ ScheduledActionData.ScheduleType.CYCLE
+ -> {
+ val cron = try {
+ cronParser.parse(scheduleContent).validate()
+ } catch (_: Exception) {
+ return null
+ }
+ val executionTime = ExecutionTime.forCron(cron)
+ executionTime.nextExecution(now).getOrNull()
+ }
+
+ ScheduledActionData.ScheduleType.ONCE -> {
+ val executionTime = try {
+ ZonedDateTime.parse(scheduleContent)
+ } catch (_: Exception) {
+ return null
+ }
+ if (executionTime.plusSeconds(1).isBefore(now) || executionTime.dayOfMonth != now.dayOfMonth)
+ null
+ else
+ executionTime
+ }
+
+ }
+
+ }
+
+ private fun logFailedStatus(actionData: ScheduledActionData) {
+ log.warn(
+ "行动未加载，uuid: {}, source: {}, tendency: {}, scheduleContent: {}",
+ actionData.uuid,
+ actionData.source,
+ actionData.tendency,
+ actionData.scheduleContent,
+ )
+ }
+
+ override fun close() {
+ timeWheelScope.cancel()
+ }
+
+ private enum class WheelState {
+ ACTIVE,
+ SLEEPING,
+ }
+ }
+}
\ No newline at end of file
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/ActionInterventor.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/ActionInterventor.java
new file mode 100644
index 00000000..f81f1e88
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/ActionInterventor.java
@@ -0,0 +1,251 @@
+package work.slhaf.partner.module.modules.action.interventor;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import lombok.val;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentModule;
+import work.slhaf.partner.api.agent.factory.module.annotation.InjectModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore;
+import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.action.entity.PhaserRecord;
+import work.slhaf.partner.core.cognation.CognationCapability;
+import work.slhaf.partner.core.memory.MemoryCapability;
+import work.slhaf.partner.module.common.module.PreRunningModule;
+import work.slhaf.partner.module.modules.action.interventor.entity.InterventionType;
+import work.slhaf.partner.module.modules.action.interventor.entity.MetaIntervention;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.InterventionEvaluator;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorInput;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorResult;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorResult.EvaluatedInterventionData;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.InterventionRecognizer;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.entity.RecognizerInput;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.entity.RecognizerResult;
+import work.slhaf.partner.runtime.interaction.data.context.PartnerRunningFlowContext;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
+
+/**
+ * 负责识别潜在的行动干预信息，作用于正在进行或已存在的行动池中内容
+ */
+@AgentModule(name = "action_identifier", order = 2)
+public class ActionInterventor extends PreRunningModule implements ActivateModel {
+
+ @InjectModule
+ private InterventionRecognizer interventionRecognizer;
+ @InjectModule
+ private InterventionEvaluator interventionEvaluator;
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+ @InjectCapability
+ private CognationCapability cognationCapability;
+ @InjectCapability
+ private MemoryCapability memoryCapability;
+
+ private final AssemblyHelper assemblyHelper = new AssemblyHelper();
+ private final PromptHelper promptHelper = new PromptHelper();
+
+ /**
+ * 键: 本次调用uuid；
+ * 值：本次调用对应的prompt；
+ */
+ private final Map> interventionPrompt = new HashMap<>();
+
+ @Override
+ protected void doExecute(PartnerRunningFlowContext context) {
+ // 综合当前正在进行的行动链信息、用户交互历史、激活的记忆切片，尝试识别出是否存在行动干预意图
+ // 首先通过recognizer进行快速意图识别，识别成功则步入评估阶段，评估成功则直接作用于目标行动链
+ // 进行快速意图识别时必须结合近期对话与进行中行动链情况
+
+ // 干预意图识别
+ String uuid = context.getUuid();
+ String userId = context.getUserId();
+ RecognizerResult recognizerResult = interventionRecognizer
+ .execute(assemblyHelper.buildRecognizerInput(userId, context.getInput())); // 此处的输入内容携带了所有 PhaserRecord
+ if (!recognizerResult.isOk()) {
+ promptHelper.setupNoInterventionPrompt(uuid);
+ return;
+ }
+
+ // 干预意图评估
+ EvaluatorResult evaluatorResult = interventionEvaluator
+ .execute(assemblyHelper.buildEvaluatorInput(recognizerResult, userId));
+ List executingDataList = evaluatorResult.getExecutingDataList();
+ List preparedDataList = evaluatorResult.getPreparedDataList();
+
+ // 意图评估结果处理
+ if (evaluatorResult.isOk()) {
+ // 对存在‘异常ActionKey’的评估结果列表进行过滤
+ invalidActionKeysFilter(executingDataList);
+ invalidActionKeysFilter(preparedDataList);
+
+ // 同步写入prompt，异步处理干预行为，‘异步’在处理流程中体现
+ promptHelper.setupInterventionPrompt(uuid, executingDataList, preparedDataList);
+ handleInterventions(executingDataList, recognizerResult.getExecutingInterventions());
+ handleInterventions(preparedDataList, recognizerResult.getPreparedInterventions());
+ } else {
+ promptHelper.setupInterventionIgnoredPrompt(uuid, executingDataList, preparedDataList);
+ }
+
+ }
+
+ private void handleInterventions(List interventionDataList, Map interventionDataMap) {
+ val executor = actionCapability.getExecutor(ActionCore.ExecutorType.PLATFORM);
+ executor.execute(() -> {
+ for (EvaluatedInterventionData interventionData : interventionDataList) {
+ // 此处拿到的为 ActionData 或者 PhaserRecord, 来自 Recognizer 的封装
+ val data = interventionDataMap.get(interventionData.getTendency());
+ actionCapability.handleInterventions(interventionData.getMetaInterventionList(), data);
+ }
+ });
+ }
+
+ private void invalidActionKeysFilter(List interventions) {
+
+ List toRemove = new ArrayList<>();
+
+ for (EvaluatedInterventionData intervention : interventions) {
+ List interventionData = intervention.getMetaInterventionList();
+ List actions = new ArrayList<>();
+ for (MetaIntervention metaData : interventionData) {
+ actions.addAll(metaData.getActions());
+ }
+ // 如果存在异常行动key，则可视为该评估结果存在问题，直接忽略该结果
+ if (!actionCapability.checkExists(actions.toArray(String[]::new))) {
+ toRemove.add(intervention);
+ }
+
+ // 针对 REBUILD 类型进行特殊校验, REBUILD 类型必须满足所有 MetaIntervention 的类型均为 REBUILD
+ if (!checkRebuildType(interventionData)) {
+ toRemove.add(intervention);
+ }
+ }
+
+ interventions.removeAll(toRemove);
+ }
+
+ private boolean checkRebuildType(List interventionData) {
+ boolean hasRebuild = false;
+ for (MetaIntervention meta : interventionData) {
+ if (meta.getType() == InterventionType.REBUILD) {
+ hasRebuild = true;
+ } else if (hasRebuild) {
+ // 已经存在REBUILD类型，但又发现了非REBUILD类型，不合法
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ @Override
+ public String modelKey() {
+ return "action_identifier";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+
+ @Override
+ protected Map getPromptDataMap(PartnerRunningFlowContext context) {
+ return interventionPrompt.remove(context.getUuid());
+ }
+
+ @Override
+ protected String moduleName() {
+ return "[行动干预识别模块]";
+ }
+
+ private final class AssemblyHelper {
+ private AssemblyHelper() {
+ }
+
+ private RecognizerInput buildRecognizerInput(String userId, String input) {
+ RecognizerInput recognizerInput = new RecognizerInput();
+ recognizerInput.setInput(input);
+ recognizerInput.setUserDialogMapStr(memoryCapability.getUserDialogMapStr(userId));
+ // 参考的对话列表大小或需调整
+ recognizerInput.setRecentMessages(cognationCapability.getChatMessages());
+ recognizerInput.setExecutingActions(actionCapability.listPhaserRecords().stream().map(PhaserRecord::actionData).toList());
+ recognizerInput.setPreparedActions(actionCapability.listActions(ActionData.ActionStatus.PREPARE, userId).stream().toList());
+ return recognizerInput;
+ }
+
+ private EvaluatorInput buildEvaluatorInput(RecognizerResult recognizerResult, String userId) {
+ EvaluatorInput input = new EvaluatorInput();
+ input.setExecutingInterventions(recognizerResult.getExecutingInterventions());
+ input.setPreparedInterventions(recognizerResult.getPreparedInterventions());
+ input.setRecentMessages(cognationCapability.getChatMessages());
+ input.setActivatedSlices(memoryCapability.getActivatedSlices(userId));
+ return input;
+ }
+ }
+
+ private final class PromptHelper {
+ private PromptHelper() {
+ }
+
+ private void setupInterventionIgnoredPrompt(String uuid, List executingDataList, List preparedDataList) {
+ List total = Stream.concat(executingDataList.stream(), preparedDataList.stream()).toList();
+
+ JSONArray reasons = new JSONArray();
+
+ for (EvaluatedInterventionData data : total) {
+ JSONObject reason = reasons.addObject();
+ reason.put("[干预倾向]", data.getTendency());
+ reason.put("[未采用原因]", data.getDescription());
+ }
+
+ synchronized (interventionPrompt) {
+ interventionPrompt.put(uuid, Map.of(
+ "[识别状态] <是否识别到干预已存在行动的意图>", "识别到，但都未采用",
+ "[忽略原因] <各个意图被忽略的原因>", reasons.toString(),
+ "[干预行动] <将对已存在行动做出的行为>", "无行为"));
+ }
+ }
+
+ private void setupInterventionPrompt(String uuid, List executingDataList,
+ List preparedDataList) {
+ JSONArray contents = new JSONArray();
+ List temp = Stream.concat(executingDataList.stream(), preparedDataList.stream()).toList();
+
+ for (EvaluatedInterventionData data : temp) {
+ if (!data.isOk()) {
+ continue;
+ }
+ String tendency = data.getTendency();
+ JSONObject newElement = contents.addObject();
+ newElement.put("[干预倾向]", tendency);
+ JSONArray changes = newElement.putArray("[行动链变动情况]");
+
+ for (MetaIntervention intervention : data.getMetaInterventionList()) {
+ JSONObject change = changes.addObject();
+ change.put("[干预类型]", intervention.getType());
+ change.put("[干预序号]", intervention.getOrder());
+ change.putArray("[干预内容]").addAll(intervention.getActions());
+ }
+ }
+
+ synchronized (interventionPrompt) {
+ interventionPrompt.put(uuid, Map.of(
+ "[识别状态] <是否识别到干预已存在行动的意图>", "识别到，将采用",
+ "[干预内容] <将对已存在行动做出的行为>", contents.toString()));
+ }
+ }
+
+ private void setupNoInterventionPrompt(String uuid) {
+ interventionPrompt.put(uuid, Map.of(
+ "[识别状态] <是否识别到干预已存在行动的意图>", "未识别到干预意图",
+ "[干预行动] <将对已存在行动做出的行为>", "无行动"));
+ }
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/InterventionType.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/InterventionType.java
new file mode 100644
index 00000000..19164c1a
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/InterventionType.java
@@ -0,0 +1,28 @@
+package work.slhaf.partner.module.modules.action.interventor.entity;
+
+public enum InterventionType {
+ /**
+ * 追加行动: 追加至指定行动链序列之后才执行
+ */
+ APPEND,
+
+ /**
+ * 插入行动: 指定行动链序列执行过程中即时新增并执行
+ */
+ INSERT,
+
+ /**
+ * 重建行动: 重建指定行动链序列之后的所有行动内容
+ */
+ REBUILD,
+
+ /**
+ * 删除行动: 删除指定行动链序列上的指定行动单元
+ */
+ DELETE,
+
+ /**
+ * 取消行动链: 中断并取消指定行动链的执行
+ */
+ CANCEL
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/MetaIntervention.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/MetaIntervention.java
new file mode 100644
index 00000000..0984548d
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/entity/MetaIntervention.java
@@ -0,0 +1,21 @@
+package work.slhaf.partner.module.modules.action.interventor.entity;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class MetaIntervention {
+ /**
+ * 干预数据类型
+ */
+ private InterventionType type;
+ /**
+ * 干预数据对应的行动链序列
+ */
+ private int order;
+ /**
+ * 干预数据所需的行动key列表
+ */
+ private List actions;
+}
\ No newline at end of file
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/InterventionEvaluator.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/InterventionEvaluator.java
new file mode 100644
index 00000000..f0675e9f
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/InterventionEvaluator.java
@@ -0,0 +1,99 @@
+package work.slhaf.partner.module.modules.action.interventor.evaluator;
+
+import com.alibaba.fastjson2.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.api.chat.pojo.ChatResponse;
+import work.slhaf.partner.api.chat.pojo.Message;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore.ExecutorType;
+import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.memory.pojo.EvaluatedSlice;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorInput;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorResult;
+import work.slhaf.partner.module.modules.action.interventor.evaluator.entity.EvaluatorResult.EvaluatedInterventionData;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+
+@Slf4j
+@AgentSubModule
+public class InterventionEvaluator extends AgentRunningSubModule
+ implements ActivateModel {
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+
+ /**
+ * 基于干预意图、记忆切片、交互上下文、已有行动程序综合评估，尝试评估并选取出合适的行动程序，交付给 ActionInterventor
+ */
+ @Override
+ public EvaluatorResult execute(EvaluatorInput input) {
+ // 获取必须数据
+ ExecutorService executor = actionCapability.getExecutor(ExecutorType.VIRTUAL);
+ Map executingInterventions = input.getExecutingInterventions();
+ Map preparedInterventions = input.getPreparedInterventions();
+ CountDownLatch latch = new CountDownLatch(executingInterventions.size() + preparedInterventions.size());
+
+ // 创建结果容器
+ EvaluatorResult result = new EvaluatorResult();
+ List executingDataList = result.getExecutingDataList();
+ List preparedDataList = result.getPreparedDataList();
+
+ // 并发评估
+ evaluateIntervention(executingDataList, executingInterventions, input, executor, latch);
+ evaluateIntervention(preparedDataList, preparedInterventions, input, executor, latch);
+
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ log.warn("CountDownLatch阻塞已中断");
+ }
+
+ return result;
+ }
+
+ private void evaluateIntervention(List evaluatedDataList, Map interventionMap, EvaluatorInput input, ExecutorService executor, CountDownLatch latch) {
+ interventionMap.forEach((tendency, actionData) -> executor.execute(() -> {
+ try {
+ String prompt = buildPrompt(input.getRecentMessages(), input.getActivatedSlices(), actionData, tendency);
+
+ ChatResponse response = this.singleChat(prompt);
+ EvaluatedInterventionData evaluatedData = JSONObject.parseObject(response.getMessage(),
+ EvaluatedInterventionData.class);
+ synchronized (evaluatedDataList) {
+ evaluatedDataList.add(evaluatedData);
+ }
+ } catch (Exception e) {
+ log.error("干预意图评估出错: {}", tendency, e);
+ } finally {
+ latch.countDown();
+ }
+ }));
+ }
+
+ private String buildPrompt(List recentMessages, List activatedSlices,
+ ActionData actionData, String tendency) {
+ JSONObject json = new JSONObject();
+ json.put("干预倾向", tendency);
+ json.putArray("近期对话").addAll(recentMessages);
+ json.putArray("参考记忆").addAll(activatedSlices);
+ json.put("将干预的行动", JSONObject.toJSONString(actionData));
+ return json.toJSONString();
+ }
+
+ @Override
+ public String modelKey() {
+ return "intervention_evaluator";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorInput.java
new file mode 100644
index 00000000..d25d8729
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorInput.java
@@ -0,0 +1,17 @@
+package work.slhaf.partner.module.modules.action.interventor.evaluator.entity;
+
+import lombok.Data;
+import work.slhaf.partner.api.chat.pojo.Message;
+import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.core.memory.pojo.EvaluatedSlice;
+
+import java.util.List;
+import java.util.Map;
+
+@Data
+public class EvaluatorInput {
+ private Map executingInterventions;
+ private Map preparedInterventions;
+ private List activatedSlices;
+ private List recentMessages;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorResult.java
new file mode 100644
index 00000000..bd249426
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/evaluator/entity/EvaluatorResult.java
@@ -0,0 +1,33 @@
+package work.slhaf.partner.module.modules.action.interventor.evaluator.entity;
+
+import lombok.Data;
+import work.slhaf.partner.module.modules.action.interventor.entity.MetaIntervention;
+
+import java.util.List;
+
+/**
+ * 干预倾向评估结果，包含评估通过的倾向文本、对行动链的行为、指定操作的行动单元key、未通过的原因
+ */
+@Data
+public class EvaluatorResult {
+ /**
+ * 是否存在通过的干预倾向
+ */
+ private boolean ok;
+ private List executingDataList;
+ private List preparedDataList;
+
+ @Data
+ public static class EvaluatedInterventionData {
+ /**
+ * 是否通过
+ */
+ private boolean ok;
+ private String tendency;
+ /**
+ * 描述信息(包括通过、失败原因)
+ */
+ private String description;
+ private List metaInterventionList;
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/InterventionRecognizer.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/InterventionRecognizer.java
new file mode 100644
index 00000000..b809f35d
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/InterventionRecognizer.java
@@ -0,0 +1,102 @@
+package work.slhaf.partner.module.modules.action.interventor.recognizer;
+
+import com.alibaba.fastjson2.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
+import work.slhaf.partner.api.agent.factory.module.annotation.AgentSubModule;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.ActivateModel;
+import work.slhaf.partner.api.agent.runtime.interaction.flow.abstracts.AgentRunningSubModule;
+import work.slhaf.partner.api.chat.pojo.ChatResponse;
+import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore;
+import work.slhaf.partner.core.action.entity.ActionData;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.entity.MetaRecognizerResult;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.entity.RecognizerInput;
+import work.slhaf.partner.module.modules.action.interventor.recognizer.entity.RecognizerResult;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+
+@Slf4j
+@AgentSubModule
+public class InterventionRecognizer extends AgentRunningSubModule implements ActivateModel {
+
+ @InjectCapability
+ private ActionCapability actionCapability;
+
+ @Override
+ public RecognizerResult execute(RecognizerInput input) {
+ // 获取必须数据
+ ExecutorService executor = actionCapability.getExecutor(ActionCore.ExecutorType.VIRTUAL);
+ List executingActions = input.getExecutingActions();
+ List preparedActions = input.getPreparedActions();
+ CountDownLatch countDownLatch = new CountDownLatch(executingActions.size() + preparedActions.size());
+
+ // 创建结果容器
+ RecognizerResult recognizerResult = new RecognizerResult();
+ Map executingInterventions = recognizerResult.getExecutingInterventions();
+ Map preparedInterventions = recognizerResult.getPreparedInterventions();
+
+ // 执行识别操作
+ recognizeIntervention(executingInterventions, executingActions, executor, input, countDownLatch);
+ recognizeIntervention(preparedInterventions, preparedActions, executor, input, countDownLatch);
+
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ log.warn("CountDownLatch阻塞已中断");
+ }
+ return recognizerResult;
+ }
+
+ private void recognizeIntervention(Map interventionsMap, List actions, ExecutorService executor, RecognizerInput input, CountDownLatch latch) {
+ for (ActionData data : actions) {
+ executor.execute(() -> {
+ try {
+ String prompt = buildPrompt(data, input);
+ ChatResponse response = this.singleChat(prompt);
+ MetaRecognizerResult result = JSONObject.parseObject(response.getMessage(), MetaRecognizerResult.class);
+ if (result.isOk()) {
+ synchronized (interventionsMap) {
+ interventionsMap.put(result.getIntervention(), data);
+ }
+ }
+ } catch (Exception e) {
+ log.error("LLM干预意图提取出错", e);
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+ }
+
+ private String buildPrompt(ActionData actionData, RecognizerInput input) {
+ JSONObject json = new JSONObject();
+
+ JSONObject actionInfo = json.putObject("行动信息");
+ actionInfo.put("行动倾向", actionData.getTendency());
+ actionInfo.put("行动原因", actionData.getReason());
+ actionInfo.put("行动描述", actionData.getDescription());
+ actionInfo.put("行动状态", actionData.getStatus());
+ actionInfo.put("行动来源", actionData.getSource());
+
+ JSONObject interactionInfo = json.putObject("交互信息");
+ interactionInfo.put("用户输入", input.getInput());
+ interactionInfo.put("当前对话", input.getRecentMessages());
+ interactionInfo.put("近期对话", input.getUserDialogMapStr());
+
+ return json.toString();
+ }
+
+ @Override
+ public String modelKey() {
+ return "intervention_recognizer";
+ }
+
+ @Override
+ public boolean withBasicPrompt() {
+ return false;
+ }
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/MetaRecognizerResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/MetaRecognizerResult.java
new file mode 100644
index 00000000..4c81233b
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/MetaRecognizerResult.java
@@ -0,0 +1,9 @@
+package work.slhaf.partner.module.modules.action.interventor.recognizer.entity;
+
+import lombok.Data;
+
+@Data
+public class MetaRecognizerResult {
+ private boolean ok;
+ private String intervention;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerInput.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerInput.java
new file mode 100644
index 00000000..9deca7cf
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerInput.java
@@ -0,0 +1,22 @@
+package work.slhaf.partner.module.modules.action.interventor.recognizer.entity;
+
+import lombok.Data;
+import work.slhaf.partner.api.chat.pojo.Message;
+import work.slhaf.partner.core.action.entity.ActionData;
+
+import java.util.List;
+
+@Data
+public class RecognizerInput {
+ private String input;
+ private List recentMessages;
+ /**
+ * 当前用户对应的近两日对话缓存
+ */
+ private String userDialogMapStr;
+ /**
+ * 正在执行的行动-Phaser记录列表，在Recognizer中结合本次输入并发评估(考虑到不同行动链之间对LLM的影响)
+ */
+ private List executingActions;
+ private List preparedActions;
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerResult.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerResult.java
new file mode 100644
index 00000000..e05511ec
--- /dev/null
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/interventor/recognizer/entity/RecognizerResult.java
@@ -0,0 +1,29 @@
+package work.slhaf.partner.module.modules.action.interventor.recognizer.entity;
+
+import lombok.Data;
+import work.slhaf.partner.core.action.entity.ActionData;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Data
+public class RecognizerResult {
+
+ private boolean ok;
+
+ /**
+ * 将被干预的‘执行中行动’
+ * key: 干预倾向
+ *
+ * value: 干预倾向将作用的行动数据
+ */
+ private Map executingInterventions = new HashMap<>();
+
+ /**
+ * 将被干预的‘等待中行动’
+ * key: 干预倾向
+ *
+ * value: 干预倾向将作用的行动数据
+ */
+ private Map preparedInterventions = new HashMap<>();
+}
diff --git a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/planner/ActionPlanner.java b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/planner/ActionPlanner.java
index abfbe049..4456406a 100644
--- a/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/planner/ActionPlanner.java
+++ b/Partner-Main/src/main/java/work/slhaf/partner/module/modules/action/planner/ActionPlanner.java
@@ -1,15 +1,19 @@
package work.slhaf.partner.module.modules.action.planner;
+import lombok.extern.slf4j.Slf4j;
+import lombok.val;
+import org.jetbrains.annotations.NotNull;
import work.slhaf.partner.api.agent.factory.capability.annotation.InjectCapability;
-import work.slhaf.partner.api.agent.factory.module.annotation.AfterExecute;
import work.slhaf.partner.api.agent.factory.module.annotation.AgentModule;
import work.slhaf.partner.api.agent.factory.module.annotation.Init;
import work.slhaf.partner.api.agent.factory.module.annotation.InjectModule;
import work.slhaf.partner.api.chat.pojo.Message;
-import work.slhaf.partner.common.thread.InteractionThreadPoolExecutor;
import work.slhaf.partner.common.vector.VectorClient;
import work.slhaf.partner.core.action.ActionCapability;
+import work.slhaf.partner.core.action.ActionCore;
import work.slhaf.partner.core.action.entity.*;
+import work.slhaf.partner.core.action.entity.cache.CacheAdjustData;
+import work.slhaf.partner.core.action.entity.cache.CacheAdjustMetaData;
import work.slhaf.partner.core.cognation.CognationCapability;
import work.slhaf.partner.core.memory.MemoryCapability;
import work.slhaf.partner.core.perceive.PerceiveCapability;
@@ -25,15 +29,15 @@ import work.slhaf.partner.module.modules.action.planner.extractor.entity.Extract
import work.slhaf.partner.module.modules.action.planner.extractor.entity.ExtractorResult;
import work.slhaf.partner.runtime.interaction.data.context.PartnerRunningFlowContext;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.UUID;
+import java.util.*;
import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
/**
* 负责针对本次输入生成基础的行动计划，在主模型传达意愿后，执行行动或者放入计划池
*/
+@Slf4j
@AgentModule(name = "action_planner", order = 2)
public class ActionPlanner extends PreRunningModule {
@@ -53,21 +57,25 @@ public class ActionPlanner extends PreRunningModule {
@InjectModule
private ActionConfirmer actionConfirmer;
- private InteractionThreadPoolExecutor executor;
- private ActionAssemblyHelper assemblyHelper;
+ private ExecutorService executor;
+
+ private final ActionAssemblyHelper assemblyHelper = new ActionAssemblyHelper();
@Init
public void init() {
- executor = InteractionThreadPoolExecutor.getInstance();
- assemblyHelper = new ActionAssemblyHelper();
+ executor = actionCapability.getExecutor(ActionCore.ExecutorType.VIRTUAL);
}
@Override
protected void doExecute(PartnerRunningFlowContext context) {
- List> tasks = new ArrayList<>();
- addConfirmTask(tasks, context);
- addNewActionTask(tasks, context);
- executor.invokeAll(tasks);
+ try {
+ List> tasks = new ArrayList<>();
+ addConfirmTask(tasks, context);
+ addNewActionTask(tasks, context);
+ executor.invokeAll(tasks);
+ } catch (Exception e) {
+ log.error("执行异常", e);
+ }
}
/**
@@ -84,14 +92,15 @@ public class ActionPlanner extends PreRunningModule {
return null;
}
EvaluatorInput evaluatorInput = assemblyHelper.buildEvaluatorInput(extractorResult, context.getUserId());
- List