# 我用 SubAgent 做了一个 AI 自动修复闭环：流式修代码、自动构建、失败重试

最近用 AI 零代码项目生成前端页面有时候会出现问题，但是直接把原始报错信息返回给前端，再让用户手动点击按钮才能开始修复，用户体验不是很好，那么如何优化呢？我想到了 SubAgent 来进行修复！

如果把这类修复工作一直交给主 Agent 来做，会带来两个明显问题：

1.  修复过程会不断污染主 Agent 的上下文。
    
2.  错误日志、修复尝试和中间结果会迅速挤占上下文窗口，影响主 Agent 后续继续处理其他任务。
    
3.  并且越接近上下文窗口的上线，AI 的注意力就会“涣散”生成的效果也就越差。
    

![](https://cdn.hashnode.com/uploads/covers/6509a5560f2f190521cc105c/9bbd69ee-2836-4f93-9a79-2d44b624611e.png align="center")

## 源码

欢迎各位大佬 Star ⭐

源码地址：[https://github.com/lieeew/leikooo-code-mother](https://github.com/lieeew/leikooo-code-mother)

## 整体流程

完整流程如下：

1.  用户点击“修复错误”。
    
2.  后端启动 `SubAgent` 的 SSE 修复流。
    
3.  查询当前激活版本号，定位当前版本目录。
    
4.  读取当前版本 `metadata.json` 中的 `errorLog`。
    
5.  `SubAgent` 基于 `errorLog` 流式修复代码，前端实时展示修复内容。
    
6.  AI 输出结束后，后端执行 `build` 和构建产物校验。
    
7.  后端更新当前版本的 `metadata.json` 和数据库状态。
    
8.  前端根据 SSE 展示本轮构建结果。
    
9.  如果构建失败，则进入下一轮修复，最多重试 3 次。
    
10.  如果最终修复成功，前端刷新版本列表和预览。
     
11.  修复成功后，前端把 `SubAgent` 的 `[Fix Summary]` 回传给主 Agent，让主 Agent 感知这次修复结果。
     

## 为什么不用主 Agent 直接修

主 Agent 负责的是整条业务链路，它需要理解页面、任务上下文、用户目标和历史操作。如果让它直接参与一轮又一轮的报错修复，短时间内确实能解决问题，但长期来看成本很高：

1.  大量错误日志和修复过程会污染主上下文。
    
2.  主 Agent 的注意力会被局部报错吸走，后续做别的任务时上下文质量下降。
    
3.  当上下文窗口逐渐逼近上限时，模型输出稳定性也会明显变差。
    

因此更合适的做法是把“局部、重复、面向报错”的工作隔离给 `SubAgent`，让主 Agent 只消费最终结果，而不是参与整个修复过程。

## 为什么在 Reactor 里必须切线程

![](https://cdn.hashnode.com/uploads/covers/6509a5560f2f190521cc105c/ae2501cb-cafe-4b11-97d5-8bcd55881ac9.png align="center")

这里有一个很容易踩的坑：我们使用的是 Reactor 的非阻塞线程模型，并且整个修复过程要通过 SSE 持续向前端推送状态。

在这种模式下，处理请求和推进事件循环的线程数量是有限的。如果把 `build`、文件读写、等待外部进程结束这类阻塞操作直接放到 Reactor 线程里执行，线程就会被长期占住，其他请求和 SSE 推送也会受到影响，系统吞吐量和响应性都会明显下降，用户体验直接爆炸。

Tomcat 传统上更接近“一请求一线程”模型。某个请求阻塞了，主要影响的是这个请求本身，只要线程池里还有空闲线程，其他请求通常还能继续跑。但 Reactor 是“少量线程处理大量连接”的事件驱动模型，核心线程的职责是分发和推进事件，一旦把阻塞逻辑塞进去，影响的就不只是当前请求，而是整个事件循环。

所以这里所有阻塞操作都要切到 `boundedElastic` 线程池执行：

```java
Mono.fromCallable(this::doBlockingWork)
    .subscribeOn(Schedulers.boundedElastic());
```

这个原则在本文里贯穿始终。像读取 `metadata.json`、执行 `build`、更新状态这类操作，都会放到 `boundedElastic` 上处理。

## 关键实现

![](https://cdn.hashnode.com/uploads/covers/6509a5560f2f190521cc105c/9f548e0b-2dcf-4b51-bc7b-e64a8ca86db7.png align="center")

### 1\. 对外暴露一个 SSE 修复接口

前端点击“修复错误”后，后端会启动一个 SSE 流，把修复中的阶段信息、AI 输出和构建结果持续推给前端。

```java
@RestController
@RequestMapping("/app")
public class AppController {

    @GetMapping(value = "/sub-agent/fix", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> fixAppWithSubAgent(
            @RequestParam(name = "appId") Long appId) {
        ThrowUtils.throwIf(appId == null, ErrorCode.PARAMS_ERROR);
        UserVO user = userService.getUserLogin();
        return subAgentService.executeFixLoop(appId.toString(), user);
    }
}
```

对应的服务接口非常简单：

```java
public interface SubAgentService {

    Flux<ServerSentEvent<String>> executeFixLoop(String appId, UserVO user);
}
```

### 2\. 给 SubAgent 单独的 ChatClient 和会话隔离

为了避免修复流程污染主 Agent 的聊天记录，我单独给 `SubAgent` 配了一个 `ChatClient`，并使用独立的内存 `ChatMemory`。这里有三个关键点：

1.  `SubAgent` 的聊天记录不是核心资产，使用内存存储即可。
    
2.  每一次修复任务都要生成独立的 `conversationId`，不能复用默认值，也不能直接复用 `appId`。
    
3.  整个修复循环结束后，再统一清理这次会话的聊天记录。
    

如果不显式指定 `conversationId`，默认值很容易导致不同修复任务之间出现上下文污染；如果直接用 `appId`，连续两次修复同一个应用时，历史聊天记录又会串起来，导致上下文重复累积。

因此更合理的做法是为每次修复生成一个唯一会话 ID，例如：

```java
private String generateSubAgentConversationId(String appId) {
    return "subfix:" + appId + ":" +
            UuidV7Generator.bytesToUuid(UuidV7Generator.generate());
}
```

然后把这个 `conversationId` 显式传给 `ChatClient`，这里的参数就指定了 CONVERSATION\_ID

```java
advisorSpec.param(CONVERSATION_ID, conversationId)
```

完整的代码如下：

```java
.advisors(advisorSpec -> {
    advisorSpec.param(GEN_APP_INFO, ChatContext.of(appId, userId));
    advisorSpec.param(CONVERSATION_ID, conversationId);
})
.toolContext(Map.of(GEN_APP_INFO, ToolsContext.of(appId, userId)))
```

`ChatClient` 的配置本身并不复杂，重点是把它和主 Agent 的上下文彻底隔离开：

```java
@Configuration
public class ChatClientConfig {

    @Bean("subAgentChatMemoryRepository")
    public ChatMemoryRepository subAgentChatMemoryRepository() {
        return new InMemoryChatMemoryRepository();
    }

    @Bean("fixChatClient")
    public ChatClient fixChatClient(
            ChatModel primaryChatModel,
            ExecuteToolAdvisor executeToolAdvisor,
            SystemMessageFirstAdvisor systemMessageFirstAdvisor,
            FileTools fileTools,
            TodolistTools todolistTools,
            ToolAdvisor toolAdvisor,
            @Qualifier("subAgentChatMemoryRepository")
            ChatMemoryRepository subAgentChatMemoryRepository) {

        TodolistCacheCleanupAdvisor todolistCacheCleanupAdvisor =
                new TodolistCacheCleanupAdvisor();

        return ChatClient.builder(primaryChatModel)
                .defaultAdvisors(
                        executeToolAdvisor,
                        systemMessageFirstAdvisor,
                        toolAdvisor,
                        todolistCacheCleanupAdvisor,
                        MessageChatMemoryAdvisor.builder(
                                MessageWindowChatMemory.builder()
                                        .chatMemoryRepository(subAgentChatMemoryRepository)
                                        .maxMessages(100)
                                        .build()
                        ).build()
                )
                .defaultTools(fileTools, todolistTools)
                .build();
    }
}
```

### 3\. 用 Flux.concat() 串起“修复 -> 构建 -> 决策下一轮”

这里的执行顺序是严格的：

1.  先让 AI 修代码。
    
2.  再执行 `build` 和校验。
    
3.  根据结果决定是结束，还是进入下一轮修复。
    

所以这里必须使用 `Flux.concat()`，而不是 `Flux.merge()`。

`Flux.concat()` 会按顺序订阅上游，前一个流只有在发送 `onComplete` 之后，后一个流才会开始执行；这正好符合“修复完成后再构建”的要求。

相反，`Flux.merge()` 会并发订阅多个流，数据会按到达顺序交错发给下游。它适合并行汇聚多个事件源，但不适合这种必须按阶段推进的修复链路。

核心逻辑如下：

```java
private Flux<ServerSentEvent<String>> doFixLoopReactive(FixLoopContext context, int attempt) {
    if (attempt > MAX_ATTEMPTS) {
        return Flux.just(doneEvent(
                new DoneEventPayload(false, MAX_ATTEMPTS, "", context.aiContentAsString())));
    }

    return Flux.concat(
            streamAiFixReactive(context, attempt),
            continueAfterAi(context, attempt)
    ).onErrorResume(e -> handleAiFailure(context, attempt, e));
}
```

其中，AI 修复阶段会先读取当前 `errorLog`，然后把模型输出按 SSE 方式流式推给前端：

```java
private Flux<ServerSentEvent<String>> streamAiFixReactive(FixLoopContext context, int attempt) {
    return Mono.fromCallable(() -> getErrorLogs(context))
            .subscribeOn(Schedulers.boundedElastic())
            .flatMapMany(errorLog -> {
                GenAppDto dto = new GenAppDto(
                        errorLog, context.appId, context.user, context.conversationId);
                return Flux.concat(
                        Flux.just(phaseEvent(FixPhaseEnum.FIXING, attempt)),
                        aiChatClient.fixCode(dto)
                                .timeout(Duration.ofMinutes(AI_TIMEOUT_MINUTES))
                                .doOnNext(context.aiContent::append)
                                .map(this::dataEvent)
                );
            });
}
```

### 4\. 构建失败就继续下一轮，最多 3 次

AI 输出结束后，会立即进入 `build` 和产物校验阶段。这里同样要把阻塞操作放到 `boundedElastic` 线程池里执行：

```java
private Flux<ServerSentEvent<String>> continueAfterAi(FixLoopContext context, int attempt) {
    return Mono.fromCallable(() -> {
                BuildValidationResult buildResult = buildAndValidate(context.appId);
                updateMetadataFile(context, buildResult);
                syncVersionStatus(context, buildResult);
                return buildResult;
            })
            .subscribeOn(Schedulers.boundedElastic())
            .flatMapMany(buildResult -> afterBuild(context, attempt, buildResult));
}
```

如果构建成功，直接发送 `done` 事件； 如果失败但还没到上限，就继续下一轮； 如果达到最大次数仍然失败，则返回最终失败结果：

```java
private Flux<ServerSentEvent<String>> afterBuild(
        FixLoopContext context, int attempt, BuildValidationResult buildResult) {

    Flux<ServerSentEvent<String>> buildEvents = buildEvents(buildResult, attempt);

    if (buildResult.success()) {
        return Flux.concat(buildEvents, Flux.just(doneEvent(successPayload(context, attempt))));
    }

    if (attempt >= MAX_ATTEMPTS) {
        return Flux.concat(buildEvents, Flux.just(doneEvent(failurePayload(context, attempt))));
    }

    return Flux.concat(buildEvents, doFixLoopReactive(context, attempt + 1));
}
```

### 5\. 修复循环结束后统一清理聊天记录

这里还有一个细节值得单独说一下：聊天记录不是在每一轮重试后立刻清，而是在整个修复循环彻底结束后统一清理。

原因如下： 1、如果每次重试都把上下文清掉，下一轮修复就拿不到上一轮已经产生的修复信息。 2、如果一直不清理，长时间积累又会增加内存压力，甚至带来 OOM 风险。

所以更平衡的做法是：在一次完整修复任务结束后，通过 `doFinally` 做统一清理。

```java
@Override
public Flux<ServerSentEvent<String>> executeFixLoop(String appId, UserVO user) {
    Long appIdLong = Long.parseLong(appId);

    return Mono.fromCallable(() -> appVersionService.getCurrentVersionNum(appIdLong))
            .subscribeOn(Schedulers.boundedElastic())
            .defaultIfEmpty(0)
            .flatMapMany(currentVersionNum -> {
                if (currentVersionNum == null || currentVersionNum <= 0) {
                    return missingVersionEvents();
                }

                FixLoopContext context = new FixLoopContext(
                        appId,
                        currentVersionNum,
                        user,
                        generateSubAgentConversationId(appId)
                );

                return doFixLoopReactive(context, 1)
                        .doFinally(signalType -> cleanupConversationMemory(context));
            });
}
```

清理逻辑本身很直接：

```java
private void cleanupConversationMemory(FixLoopContext context) {
    try {
        subAgentChatMemoryRepository.deleteByConversationId(context.conversationId);
        log.info("[SubAgent] Cleared chat memory, appId: {}, conversationId: {}",
                context.appId, context.conversationId);
    } catch (Exception e) {
        log.warn("[SubAgent] Failed to clear chat memory, appId: {}, conversationId: {}",
                context.appId, context.conversationId, e);
    }
}
```

## 前端效果

前端这部分主要做了三件事：

1.  在用户触发修复后，及时展示当前状态和错误提示。
    
2.  把 `SubAgent` 的流式输出实时渲染出来，方便观察修复过程。
    
3.  在修复成功后，把 `[Fix Summary]` 回传给主 Agent，完成主从 Agent 之间的信息闭环。
    

## 总结

至此 SubAgent 的实现告一段落，大概有以下几点：

1.  `SubAgent` 负责局部修复，避免污染主 Agent 上下文。
    
2.  SSE 让修复过程可观察，前端能实时感知阶段变化。
    
3.  `build` 和产物校验把“修没修好”从主观判断变成了可验证结果。
    
4.  自动重试让整个链路具备了基础的自愈能力。
    

下一期准备使用使用 Docker 隔离 AI 生成的代码，让 AI 生成的代码先在 Docker 里面跑 build => 成功生成 dist 文件 => 复制出来 => 前端进行展示