实现 Multi-Agent 和 SubAgent 并发，让 Agent 团队跑得飞起

// src/main/java/com/paicli/agent/AgentRole.java
public enum AgentRole {
    PLANNER("规划者", "负责分析用户任务，制定执行计划，将复杂任务拆解为可执行的子任务"),
    WORKER("执行者", "负责执行具体任务步骤，调用工具完成文件操作、命令执行等操作"),
    REVIEWER("检查者", "负责检查执行结果的质量和正确性，提供改进建议");
}

// src/main/java/com/paicli/agent/AgentMessage.java
public record AgentMessage(
        String fromAgent,
        AgentRole fromRole,
        String content,
        Type type
) {
    public enum Type {
        TASK,
        RESULT,
        PARTIAL,
        FEEDBACK,
        APPROVAL,
        REJECTION,
        ERROR
    }
}

// src/main/java/com/paicli/agent/SubAgent.java
private boolean shouldUseTools() {
    return role == AgentRole.WORKER;
}

// src/main/java/com/paicli/agent/SubAgent.java
if (response.hasToolCalls() && !shouldUseTools()) {
    streamRenderer.finish();
    String reason = "本角色不提供工具，工具调用未执行";
    conversationLedger.appendMessage("team", name, "llm_response", LlmClient.Message.assistant(
            response.reasoningContent(), response.content(), response.toolCalls()));
    out.println(AnsiStyle.subtle(reason));
    return AgentMessage.error(name, role, reason);
}

// src/main/java/com/paicli/agent/SubAgent.java
private String getSystemPrompt() {
    return promptAssembler.assemble(promptMode(), PromptContext.builder()
            .projectMemoryContext(buildProjectMemoryContext())
            .externalContext(buildExternalContext())
            .skillIndex(shouldUseTools() ? buildSkillIndex() : "")
            .toolsEnabled(shouldUseTools() && llmClient.supportsTools())
            .build());
}

【系统消息】

基础规则、执行者职责、审批规则、上下文管理和输出要求……

## Project Context
项目规范（来自 PAI.md）
- 项目使用 Java 17。
- 修改配置后运行对应测试。

## Skills
……执行者可用的技能目录……

## MCP Resources 索引（仅 URI / 描述，不含正文）
- @docs:demo://config-guide — 项目配置说明

【本轮任务消息】

<paicli-turn-context>
当前日期: 2026-11-09
当前时区: Asia/Taipei
</paicli-turn-context>

总任务上下文：检查项目配置是否符合团队规范。

当前任务：读取配置文件，核对 Java 版本。

// src/main/java/com/paicli/agent/TeamStructuredReply.java
private static final ObjectMapper STRICT_MAPPER = new ObjectMapper()
        .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
        .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);

static JsonNode parseObject(String reply) throws IOException {
    if (reply == null || reply.isBlank()) {
        throw new IOException("回复为空");
    }
    JsonNode root;
    try {
        root = STRICT_MAPPER.readTree(unwrapWholeFence(reply.trim()));
    } catch (IOException e) {
        throw new IOException("回复不是单个合法 JSON 对象", e);
    }
    if (root == null || !root.isObject()) {
        throw new IOException("回复必须是 JSON 对象");
    }
    return root;
}

// src/main/java/com/paicli/agent/TeamPlanParser.java
private static void requireAcyclic(List<AgentOrchestrator.ExecutionStep> steps) throws IOException {
    Map<String, Integer> pendingDeps = new HashMap<>();
    Map<String, List<String>> dependents = new HashMap<>();
    for (AgentOrchestrator.ExecutionStep step : steps) {
        pendingDeps.put(step.id(), step.dependencies().size());
        for (String dep : step.dependencies()) {
            dependents.computeIfAbsent(dep, key -> new ArrayList<>()).add(step.id());
        }
    }
    Deque<String> ready = new ArrayDeque<>();
    pendingDeps.forEach((id, count) -> {
        if (count == 0) {
            ready.add(id);
        }
    });
    int visited = 0;
    while (!ready.isEmpty()) {
        String id = ready.poll();
        visited++;
        for (String dependent : dependents.getOrDefault(id, List.of())) {
            if (pendingDeps.merge(dependent, -1, Integer::sum) == 0) {
                ready.add(dependent);
            }
        }
    }
    if (visited != steps.size()) {
        throw new IOException("计划中存在循环依赖");
    }
}

// src/main/java/com/paicli/agent/AgentOrchestrator.java
this.planner = new SubAgent("planner", AgentRole.PLANNER, llmClient, toolRegistry);
this.workers = List.of(
        new SubAgent("worker-1", AgentRole.WORKER, llmClient, toolRegistry),
        new SubAgent("worker-2", AgentRole.WORKER, llmClient, toolRegistry)
);
this.reviewer = new SubAgent("reviewer", AgentRole.REVIEWER, llmClient, toolRegistry);

// src/main/java/com/paicli/agent/AgentOrchestrator.java
List<ExecutionStep> getExecutableSteps(List<ExecutionStep> steps) {
    Map<String, StepStatus> statusMap = new HashMap<>();
    for (ExecutionStep step : steps) {
        statusMap.put(step.id(), step.status());
    }

    return steps.stream()
            .filter(step -> step.status() == StepStatus.PENDING)
            .filter(step -> step.dependencies().stream()
                    .allMatch(dep -> statusMap.get(dep) == StepStatus.COMPLETED))
            .toList();
}

// src/main/java/com/paicli/agent/AgentOrchestrator.java
private void runBatchParallel(List<ExecutionStep> batch, List<ExecutionStep> steps,
                              TurnToolPolicy turnToolPolicy,
                              Map<String, TurnToolPolicy.TrustedUrlContext> stepTrustedUrls,
                              TeamObservation observation, int batchOrdinal) {
    int parallelism = Math.min(batch.size(), workers.size());
    ExecutorService executor = Executors.newFixedThreadPool(parallelism, r -> {
        Thread t = new Thread(r, "paicli-multi-agent");
        t.setDaemon(true);
        return t;
    });
    BlockingQueue<SubAgent> workerPool = new LinkedBlockingQueue<>(workers);
    Map<String, ByteArrayOutputStream> buffers = new ConcurrentHashMap<>();
    List<Future<?>> futures = new ArrayList<>();

    for (ExecutionStep step : batch) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        buffers.put(step.id(), baos);
        PrintStream stepOut = new PrintStream(baos, true, StandardCharsets.UTF_8);
        // ... 准备 stepPolicy 与 context
        futures.add(executor.submit(() -> {
            SubAgent worker = null;
            SubAgent localReviewer = new SubAgent(
                    "reviewer-" + step.id(), AgentRole.REVIEWER, llmClient, toolRegistry);
            localReviewer.setExternalContextSupplier(externalContextSupplier);
            localReviewer.setConversationLedger(conversationLedger);
            try {
                worker = workerPool.take();
                runStep(step, steps, worker, localReviewer, context, stepOut,
                        stepPolicy, stepTrustedUrls, stepObservation);
            } finally {
                if (worker != null) {
                    worker.clearHistory();
                    workerPool.offer(worker);
                }
                stepOut.flush();
            }
            return null;
        }));
    }
    // 等待整批完成并按 step_id 顺序刷出缓冲
}

// src/main/java/com/paicli/agent/AgentOrchestrator.java
private String buildStepContext(List<ExecutionStep> steps, ExecutionStep currentStep,
                                List<TurnToolPolicy.TrustedUrlContext> dependencyUrls) {
    StringBuilder context = new StringBuilder();
    context.append("总任务上下文：\n");

    for (ExecutionStep step : steps) {
        if (step.status() == StepStatus.COMPLETED && currentStep.dependencies().contains(step.id())) {
            context.append("已完成的依赖步骤 [").append(step.id()).append("]: ")
                    .append(step.description()).append("\n");
            if (step.result() != null && !step.result().isBlank()) {
                String preview = step.result().length() > 500
                        ? step.result().substring(0, 500) + "..."
                        : step.result();
                context.append("结果：").append(preview).append("\n");
            }
            context.append("\n");
        }
    }

    Set<String> trustedDependencyUrls = dependencyUrls.stream()
            .flatMap(contextItem -> contextItem.urls().stream())
            .collect(Collectors.toCollection(LinkedHashSet::new));
    if (!trustedDependencyUrls.isEmpty()) {
        context.append("依赖分支经 web_search 验证的 URL（可供当前步骤抓取/导航）：\n");
        trustedDependencyUrls.forEach(url -> context.append("- ").append(url).append("\n"));
        context.append("\n");
    }

    return context.toString();
}

// src/main/java/com/paicli/agent/TeamWorkerReport.java
record TeamWorkerReport(String result, String blockedReason) {

    static TeamWorkerReport parse(String reply) throws IOException {
        JsonNode json = TeamStructuredReply.parseObject(reply);
        String result = textField(json, "result");
        String reason = textField(json, "reason");
        return switch (json.path("status").asText()) {
            case "completed" -> {
                if (result == null || result.isBlank() || reason != null && !reason.isBlank()) {
                    throw new IOException("completed 需要非空 result，不能同时报告阻塞原因");
                }
                yield new TeamWorkerReport(result, null);
            }
            case "blocked" -> {
                if (reason == null || reason.isBlank()) throw new IOException("blocked 需要说明 reason");
                yield new TeamWorkerReport(result, reason);
            }
            default -> throw new IOException("status 必须是 completed 或 blocked");
        };
    }
}

// src/main/java/com/paicli/agent/TeamReviewVerdict.java
record TeamReviewVerdict(boolean approved, String feedback) {

    static final String DEFAULT_FEEDBACK = "审查未通过，请改进执行结果";

    static TeamReviewVerdict parse(String reviewContent) {
        JsonNode root;
        try {
            root = TeamStructuredReply.parseObject(reviewContent);
        } catch (IOException e) {
            return rejected("审查结论无法解析（" + e.getMessage() + "），按未通过处理");
        }
        JsonNode approvedNode = root.get("approved");
        if (approvedNode == null || !approvedNode.isBoolean()) {
            return rejected("审查结论缺少布尔类型的 approved 字段，按未通过处理");
        }
        return new TeamReviewVerdict(approvedNode.booleanValue(), feedbackOf(root));
    }
}

// src/main/java/com/paicli/agent/AgentOrchestrator.java
for (int attempt = 1; attempt <= 1 + MAX_RETRIES_PER_STEP; attempt++) {
    if (cancelStepIfRequested(step, steps, observation, out)) return;
    String workerContext = context;
    if (attempt > 1) {
        out.println("⚠️ 步骤 [" + step.id() + "] 重新执行，反馈：" + feedback);
        workerContext += "\n\n之前的执行结果被审查拒绝，原因：\n" + feedback;
    }
    ActivationObservation workerCall = observation == null ? null : observation.worker(worker, attempt);
    AgentMessage result = worker.executeWithContext(task, workerContext, out, stepPolicy, workerCall);
    // ... 校验 TeamWorkerReport 并调用 reviewer.review
    try {
        review = reviewer.review(step.description(), retainedResult, out, reviewCall);
    } finally {
        reviewer.clearHistory();
    }
    // ... 解析 TeamReviewVerdict，通过则 markStepCompleted 并 return
}
String reason = "重试 " + MAX_RETRIES_PER_STEP + " 次后仍未通过审查：" + feedback;
updateStep(steps, step.id(), step.withFailed(reason, retainedResult));

// src/main/java/com/paicli/agent/SubAgent.java
public void clearHistory() {
    autoCompactionManager.clear(conversationHistory);
    LlmClient.Message systemMsg = conversationHistory.get(0);
    conversationLedger.appendEvent(
            "history_clear",
            "team",
            name,
            "task_boundary",
            Map.of("discardedViewMessages", Math.max(0, conversationHistory.size() - 1)));
    conversationHistory.clear();
    conversationHistory.add(systemMsg);
    observationHistoryGeneration++;
}

// src/main/java/com/paicli/agent/AgentOrchestrator.java
for (ExecutionStep step : steps) {
    if (step.status() == StepStatus.PENDING) {
        observation.blocked(step);
        out.println("⏭️ 步骤 [" + step.id() + "] 因前置步骤未完成被跳过: " + step.description());
    }
}

mvn clean package
java -jar target/paicli-1.0-SNAPSHOT.jar

👤 你: /team 创建一个 Spring Boot 项目，写一个 HelloController，然后验证项目结构

📋 第一阶段：规划
🧑‍💼 规划者正在分析任务...

📋 执行计划
  ⏳ [step_1] 创建 Spring Boot 项目基础结构，包括目录和 pom.xml 配置文件 (依赖: 无)
  ⏳ [step_2] 创建 Spring Boot 主启动类 DemoApplication.java (依赖: step_1)
  ⏳ [step_3] 创建 HelloController.java，实现返回 'Hello, Spring Boot!' 的接口 (依赖: step_1)
  ⏳ [step_4] 验证项目结构是否完整，并执行编译命令确保无语法错误 (依赖: step_2, step_3)

⚡ 第二阶段：执行
🛠️ worker-1 执行步骤 [step_1]: 创建 Spring Boot 项目基础结构，包括目录和 pom.xml 配置文件
🔍 reviewer 正在审查步骤 [step_1] 的结果...
✅ 步骤 [step_1] 审查通过
🛠️ worker-2 执行步骤 [step_2]: 创建 Spring Boot 主启动类 DemoApplication.java
🔍 reviewer 正在审查步骤 [step_2] 的结果...
✅ 步骤 [step_2] 审查通过
🛠️ worker-1 执行步骤 [step_3]: 创建 HelloController.java，实现返回 'Hello, Spring Boot!' 的接口
🔍 reviewer 正在审查步骤 [step_3] 的结果...
✅ 步骤 [step_3] 审查通过
🛠️ worker-2 执行步骤 [step_4]: 验证项目结构是否完整，并执行编译命令确保无语法错误
🔍 reviewer 正在审查步骤 [step_4] 的结果...
✅ 步骤 [step_4] 审查通过

mvn test -Dtest=AgentRoleTest,AgentMessageTest,SubAgentTest,TeamPlanParserTest,TeamReviewVerdictTest,AgentOrchestratorTest,TeamExecutionObservationTest,TeamRequestAuditTest -DskipTests=false

public enum AgentRole {
    PLANNER("规划者", "负责分析用户任务，制定执行计划，将复杂任务拆解为可执行的子任务"),
    WORKER("执行者", "负责执行具体任务步骤，调用工具完成文件操作、命令执行等操作"),
    REVIEWER("检查者", "负责检查执行结果的质量和正确性，提供改进建议");
}

public record AgentMessage(
        String fromAgent,
        AgentRole fromRole,
        String content,
        Type type
) {
    public enum Type {
        TASK, RESULT, FEEDBACK, APPROVAL, REJECTION, ERROR
    }
}

public AgentOrchestrator(GLMClient llmClient, ToolRegistry toolRegistry, MemoryManager memoryManager) {
    this.llmClient = llmClient;
    this.toolRegistry = toolRegistry;
    this.planner = new SubAgent("planner", AgentRole.PLANNER, llmClient, toolRegistry);
    this.workers = List.of(
            new SubAgent("worker-1", AgentRole.WORKER, llmClient, toolRegistry),
            new SubAgent("worker-2", AgentRole.WORKER, llmClient, toolRegistry)
    );
    this.reviewer = new SubAgent("reviewer", AgentRole.REVIEWER, llmClient, toolRegistry);
    this.memoryManager = memoryManager;
}

public String run(String userInput) {
    // 1. 规划
    AgentMessage planResult = planner.execute(planMessage);

    // 2. 解析计划
    List<ExecutionStep> steps = parsePlan(planResult.content());

    // 3. 执行 + 审查
    while (true) {
        List<ExecutionStep> executable = getExecutableSteps(steps);
        if (executable.isEmpty()) break;
        // 单步串行，多步并行
    }

    // 4. 处理残留 + 汇总
    String finalResult = buildFinalResult(steps);
    return finalResult;
}

List<ExecutionStep> getExecutableSteps(List<ExecutionStep> steps) {
    Map<String, StepStatus> statusMap = new HashMap<>();
    for (ExecutionStep step : steps) {
        statusMap.put(step.id(), step.status());
    }
    return steps.stream()
            .filter(step -> step.status() == StepStatus.PENDING)
            .filter(step -> step.dependencies().stream()
                    .allMatch(dep -> statusMap.get(dep) == StepStatus.COMPLETED))
            .toList();
}

private void runBatchParallel(List<ExecutionStep> batch, ...) {
    ExecutorService executor = Executors.newFixedThreadPool(parallelism);
    BlockingQueue<SubAgent> workerPool = new LinkedBlockingQueue<>(workers);
    Map<String, ByteArrayOutputStream> buffers = new ConcurrentHashMap<>();

    for (ExecutionStep step : batch) {
        // 每步一个 buffer，并行执行
        futures.add(executor.submit(() -> {
            SubAgent worker = workerPool.take(); // 从池子取 Worker
            // 执行 + 审查
            workerPool.offer(worker); // 用完放回
        }));
    }

    // 按 step_id 顺序 flush
    for (ExecutionStep step : batch) {
        System.out.print(buffers.get(step.id()).toString());
    }
}

private static final String PLANNER_PROMPT = """
        你是一个任务规划专家。你的职责是分析用户的需求，将其拆解为清晰的执行步骤。

        请按以下 JSON 格式输出执行计划：
        {
            "summary": "任务摘要",
            "steps": [
                {
                    "id": "step_1",
                    "description": "步骤描述，要具体明确",
                    "type": "FILE_READ | FILE_WRITE | COMMAND | ANALYSIS | VERIFICATION",
                    "dependencies": []
                }
            ]
        }
        """;

private static final String WORKER_PROMPT = """
        你是一个任务执行专家。你的职责是根据给定的任务步骤，调用工具完成具体操作。

        可用工具：
        1. read_file - 读取文件内容
        2. write_file - 写入文件内容
        3. list_dir - 列出目录内容
        4. execute_command - 执行命令
        5. create_project - 创建项目
        6. search_code - 语义检索代码库

        如果任务涉及理解代码库，请优先使用 search_code 工具。
        """;

private static final String REVIEWER_PROMPT = """
        你是一个质量检查专家。你的职责是检查执行结果是否正确、完整和高质量。

        请以 JSON 格式输出检查结果：
        {
            "approved": true 或 false,
            "summary": "检查摘要",
            "issues": ["问题1", "问题2"],
            "suggestions": ["建议1", "建议2"]
        }
        """;

private boolean shouldUseTools() {
    return role == AgentRole.WORKER;
}

public void clearHistory() {
    GLMClient.Message systemMsg = conversationHistory.get(0);
    conversationHistory.clear();
    conversationHistory.add(systemMsg);
}

AgentMessage reviewResult = reviewer.review(step.description(), result.content(), out);

boolean approved = parseReviewApproval(reviewResult.content());

boolean parseReviewApproval(String reviewContent) {
    if (reviewContent == null || reviewContent.isEmpty()) {
        return false; // 空内容，默认不通过
    }
    try {
        JsonNode root = mapper.readTree(cleaned);
        JsonNode approvedNode = root.path("approved");
        if (approvedNode.isMissingNode() || approvedNode.isNull()) {
            return false; // 缺少 approved 字段，默认不通过
        }
        return approvedNode.asBoolean(false);
    } catch (Exception e) {
        // JSON 解析失败：必须同时有肯定关键词且无否定关键词才视为通过
        String lower = reviewContent.toLowerCase();
        boolean hasNegative = lower.contains("未通过") || lower.contains("不通过");
        boolean hasPositive = lower.contains("通过") || lower.contains("合格");
        if (hasNegative) return false;
        if (!hasPositive) return false; // 既无肯定也无否定，保守判不通过
        return true;
    }
}

while (!approved && retries < MAX_RETRIES_PER_STEP) {
    retries++;
    String feedbackContext = context + "\n\n之前的执行结果被审查拒绝，原因：\n" + issues;
    AgentMessage retryResult = worker.executeWithContext(taskMsg, feedbackContext, out);
    AgentMessage retryReview = reviewer.review(step.description(), retryResult.content(), out);
    approved = parseReviewApproval(retryReview.content());
}

private String buildStepContext(List<ExecutionStep> steps, ExecutionStep currentStep) {
    StringBuilder context = new StringBuilder();
    context.append("总任务上下文：\n");
    for (ExecutionStep step : steps) {
        if (step.status() == StepStatus.COMPLETED && currentStep.dependencies().contains(step.id())) {
            context.append("已完成的依赖步骤 [").append(step.id()).append("]: ")
                    .append(step.description()).append("\n");
            String preview = step.result().length() > 500
                    ? step.result().substring(0, 500) + "..."
                    : step.result();
            context.append("结果：").append(preview).append("\n");
        }
    }
    return context.toString();
}

mvn clean package

java -jar target/paicli-1.0-SNAPSHOT.jar

👤 你: /team 

👤 你: 创建一个 Spring Boot 项目，写一个 HelloController，然后验证项目结构

👤 你: /team 创建一个名为 demoapp 的 Java 项目，然后读取 pom.xml，最后验证项目结构

📋 第一阶段：规划
🧑‍💼 规划者正在分析任务...

📋 执行计划
  ⏳ [step_1] 创建 demoapp 项目结构 (依赖: 无)
  ⏳ [step_2] 读取 demoapp/pom.xml 内容 (依赖: step_1)
  ⏳ [step_3] 验证项目结构与 Maven 配置 (依赖: step_2)

⚡ 第二阶段：执行
🛠️ worker-1 执行步骤 [step_1]: 创建 demoapp 项目结构
🔍 reviewer 正在审查步骤 [step_1] 的结果...
✅ 步骤 [step_1] 审查通过
🛠️ worker-2 执行步骤 [step_2]: 读取 demoapp/pom.xml 内容
🔍 reviewer 正在审查步骤 [step_2] 的结果...
✅ 步骤 [step_2] 审查通过
🛠️ worker-1 执行步骤 [step_3]: 验证项目结构与 Maven 配置
🔍 reviewer 正在审查步骤 [step_3] 的结果...
✅ 步骤 [step_3] 审查通过