ReAct、Plan-and-Execute、Multi-Agent全都支持并行了，Agent执行效率直接拉满

PaiCLI 已经更新到第7期了，ReAct、Plan-and-Execute、Memory、RAG、Multi-Agent、HITL，该有的都有了。

但有一个问题一直没解决——串行。

我们让 PaiCLI 帮忙读三个文件，它会老老实实读完第一个，再读第二个，再读第三个。三个文件之间没有任何依赖关系，完全可以同时读，但 Agent 偏偏要排队。

Plan-and-Execute 模式下更明显。五个任务拆出来，前两个互相不依赖，第三个依赖前两个的结果。按道理前两个应该同时跑，但现在是第一个跑完才轮到第二个。

Multi-Agent 也一样，两个 Worker 都闲着，但编排器只分配给其中一个，另一个干等。

今天，我们就把 PaiCLI 从串行改造成并行。改完之后，三条执行路径——ReAct、Plan-and-Execute、Multi-Agent——全部支持并行执行，效率直接拉满。

01、并行的切入点有哪些？

PaiCLI 里有三个可以并行的场景。

第一个是工具调用的并行。大模型在一次响应里返回多个 tool_calls，这几个工具之间没有依赖，可以同时执行。

比如 LLM 说“我要同时读 pom.xml 和 README.md”，返回两个 read_file 调用，那就没必要等第一个读完再读第二个。

第二个是 Plan-and-Execute 模式下的任务并行。DAG 里同一批次的任务没有依赖关系，可以同时丢给 LLM 执行。

第三个是 Multi-Agent 模式下的 Worker 并行。编排器发现一批独立的步骤，可以同时分配给多个 Worker 跑。

02、工具并行执行

先看改造前的 ToolRegistry：

public String executeTool(String name, String argumentsJson) {
    Tool tool = tools.get(name);
    if (tool == null) {
        return "未知工具: " + name;
    }
    // 解析参数，执行工具
    JsonNode args = mapper.readTree(argumentsJson);
    Map argMap = new HashMap<>();
    args.fields().forEachRemaining(entry ->
            argMap.put(entry.getKey(), entry.getValue().asText()));
    return tool.executor().execute(argMap);
}

一次调用执行一个工具，返回一个字符串。Agent 拿到多个 tool_calls 就循环调用这个方法，串行。

现在要加一个批量执行的方法 executeTools，接收一组工具调用，并行跑完再返回结果。

先定义数据结构。

ToolInvocation 表示一次工具调用请求，ToolExecutionResult 表示执行结果：

public record ToolInvocation(String id, String name, String argumentsJson) {}

public record ToolExecutionResult(String id, String name, String argumentsJson,
                                  String result, long elapsedMillis, boolean timedOut) {
    static ToolExecutionResult completed(ToolInvocation inv, String result, long elapsed) {
        return new ToolExecutionResult(inv.id(), inv.name(), inv.argumentsJson(),
                result, elapsed, false);
    }

    static ToolExecutionResult timedOut(ToolInvocation inv, long timeoutSeconds) {
        return new ToolExecutionResult(inv.id(), inv.name(), inv.argumentsJson(),
                "工具执行超时（" + timeoutSeconds + "秒），已取消",
                timeoutSeconds * 1000, true);
    }

    static ToolExecutionResult failed(ToolInvocation inv, String message) {
        return completed(inv, "工具执行失败: " + message, 0);
    }
}

timedOut 字段标记是否超时，方便后续 LLM 根据超时结果决定是否重试。

核心的 executeTools 方法：

private static final int MAX_PARALLEL_TOOLS = 4;
private static final int DEFAULT_TOOL_BATCH_TIMEOUT_SECONDS = 90;

public List executeTools(List invocations) {
    if (invocations == null || invocations.isEmpty()) {
        return List.of();
    }
    // 只有一个工具时，不开线程池，直接执行
    if (invocations.size() == 1) {
        ToolInvocation invocation = invocations.get(0);
        long startedAt = System.nanoTime();
        String result = executeTool(invocation.name(), invocation.argumentsJson());
        return List.of(ToolExecutionResult.completed(invocation, result,
                elapsedMillis(startedAt)));
    }

    int parallelism = Math.min(invocations.size(), MAX_PARALLEL_TOOLS);
    ExecutorService executor = Executors.newFixedThreadPool(parallelism, r -> {
        Thread thread = new Thread(r, "paicli-tool-executor");
        thread.setDaemon(true);
        return thread;
    });

    try {
        List> tasks = invocations.stream()
                .>map(invocation -> () -> {
                    long startedAt = System.nanoTime();
                    String result = executeTool(invocation.name(),
                            invocation.argumentsJson());
                    return ToolExecutionResult.completed(invocation, result,
                            elapsedMillis(startedAt));
                })
                .toList();

        List> futures =
                executor.invokeAll(tasks, toolBatchTimeoutSeconds, TimeUnit.SECONDS);

        List results = new ArrayList<>();
        for (int i = 0; i < futures.size(); i++) {
            Future future = futures.get(i);
            ToolInvocation invocation = invocations.get(i);
            if (future.isCancelled()) {
                results.add(ToolExecutionResult.timedOut(invocation,
                        toolBatchTimeoutSeconds));
                continue;
            }
            results.add(future.get());
        }
        return results;
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        return invocations.stream()
                .map(inv -> ToolExecutionResult.failed(inv, "工具批次执行被中断"))
                .toList();
    } finally {
        executor.shutdownNow();
    }
}