用 SubAgent 搭建 AI 自动修复闭环：流式修代码、自动构建、失败重试的 TaoToken 配置实战

1. 为什么要把“修代码”这件事从主 Agent 手里拿走先说结论SubAgent 驱动的 AI 自动修复闭环本质是把“局部、重复、面向报错”的脏活隔离出去让主 Agent 只消费最终结果。它适合谁适合已经在用 AI 生成前端页面、但被“生成完报错、用户手动点修复、主对话越修越乱”折磨的开发者。能做什么一句话用户点一次修复后端启动 SSE 流SubAgent 流式改代码改完自动 build构建失败就带着新错误日志再修一轮最多重试 3 次成功或彻底失败都通过 SSE 把结果推回前端。我踩过的坑是一开始图省事直接让主 Agent 去修。结果错误日志、修复尝试、中间产物全塞进主上下文几轮下来上下文窗口逼近上限模型输出开始“涣散”后面正常任务也跟着变差。更麻烦的是Reactor 的非阻塞线程模型下如果把 build、文件读写、等待外部进程这类阻塞操作直接丢进事件循环线程整个 SSE 推送都会被拖住吞吐量肉眼可见地掉。所以这篇不讲空泛概念直接给你一套可复制的闭环骨架TaoToken 统一 Key/API 通道怎么配、SubAgent 的 ChatClient 怎么隔离、Flux.concat()怎么串起“修复→构建→决策下一轮”、失败重试怎么验证。你照着改就能跑起来。2. TaoToken 前置统一 Key 与 API 通道在动手写 SubAgent 之前先把模型通道统一掉。TaoToken 在这里扮演的角色是“统一入口”你不需要在代码里散落多家厂商的 Key 和 Base URL而是通过一个统一 Key 走 API 通道SubAgent 和主 Agent 共用同一套配置切换模型时只改配置不改业务代码。官网入口在这里注册和查看套餐都从这进https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基地址注意这个不带 UTM直接用于代码配置https://taotoken.net/api你需要先拿到 API Key去控制台的 API Keys 页面创建https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content如果你后面要跑长期编码或 Agent 任务建议直接看 Coding Plan额度模型更适合这种“多轮重试”的消耗场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content接入文档在这里配置字段有疑问时对照查https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content注意Key 只放在服务端配置或环境变量里不要写进前端代码也不要提交到 Git 仓库。SubAgent 的修复流是后端发起的前端只消费 SSE不接触 Key。3. 可复制配置settings.json 与 config.toml 骨架不同工具链读的配置文件不一样这里给两份骨架。核心思路一致把 Base URL 指向 TaoToken 的 API 地址把 Key 用环境变量注入模型名按你实际使用的填。3.1 settings.json 配置骨架适用于读取 JSON 配置的客户端。把下面内容放进你的settings.json注意env里的变量名按你项目实际约定调整{ env: { TAOTOKEN_API_KEY: sk-你的Key, TAOTOKEN_BASE_URL: https://taotoken.net/api }, model: { provider: taotoken, baseUrl: https://taotoken.net/api, apiKeyEnv: TAOTOKEN_API_KEY, modelName: claude-sonnet-4-5, timeoutSeconds: 300, maxRetries: 3 }, subAgent: { enabled: true, conversationPrefix: subfix:, maxMessages: 100, maxAttempts: 3 } }几个字段值得单独说timeoutSeconds给到 300 是因为流式修复一轮可能跑几分钟超时太短会误判失败maxRetries是网络层重试和业务层的“构建失败重试 3 次”是两回事别混subAgent.maxAttempts对应修复闭环的最大轮数。3.2 config.toml 配置骨架适用于 TOML 风格的客户端比如一些 CLI 工具和 Agent 框架[provider.taotoken] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY model claude-sonnet-4-5 timeout_seconds 300 [sub_agent] enabled true conversation_prefix subfix: max_messages 100 max_attempts 3 build_command npm run build dist_dir dist [reactor] blocking_scheduler boundedElastic sse_timeout_minutes 5build_command和dist_dir是给构建校验阶段用的SubAgent 改完代码后后端执行这条命令然后检查dist目录是否产出了有效文件。这一步把“修没修好”从主观判断变成可验证结果是整个闭环能自动重试的前提。提示两份配置里的模型名只是示例按你账号实际可用的模型填。配置改完先别急着接 SubAgent用下一节的验证请求确认通道通了再往下走。4. 验证请求先确认通道通了再谈闭环配置写完第一步不是直接跑修复循环而是发一个最小请求确认 TaoToken 通道可用。用 curl 验证最直接curl -X POST https://taotoken.net/api/v1/messages \ -H Content-Type: application/json \ -H x-api-key: $TAOTOKEN_API_KEY \ -H anthropic-version: 2023-06-01 \ -d { model: claude-sonnet-4-5, max_tokens: 128, stream: true, messages: [ {role: user, content: 只回复两个字通了} ] }stream: true是关键因为 SubAgent 的修复过程就是靠 SSE 流式推给前端的。如果这里能持续收到data:开头的事件块说明流式通道没问题。返回类似下面这种分块输出就对了event: message_start data: {type:message_start,message:{id:msg_xxx,role:assistant}} event: content_block_delta data: {type:content_block_delta,delta:{type:text_delta,text:通}} event: content_block_delta data: {type:content_block_delta,delta:{type:text_delta,text:了}} event: message_stop data: {type:message_stop}通道确认后再验证 SubAgent 的会话隔离逻辑。核心是每次修复生成独立的conversationId不能复用默认值也不能直接用appIdprivate String generateSubAgentConversationId(String appId) { return subfix: appId : UuidV7Generator.bytesToUuid(UuidV7Generator.generate()); }为什么不能直接用appId因为连续两次修复同一个应用时历史聊天记录会串起来上下文重复累积第二轮修复会带着上一轮的残留信息跑结果不可控。加上 UUID 后缀后每次修复都是干净会话。然后把这个 ID 显式传给 ChatClient.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)))SubAgent 的 ChatClient 单独配一个内存 ChatMemory和主 Agent 彻底隔离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(); }maxMessages(100)是窗口上限防止单轮修复把内存撑爆。SubAgent 的聊天记录不是核心资产用内存存储即可整个修复循环结束后统一清理。5. 闭环核心Flux.concat 串起修复、构建与重试执行顺序是严格的先让 AI 修代码再执行 build 和校验根据结果决定结束还是进入下一轮。所以必须用Flux.concat()不能用Flux.merge()。Flux.concat()按顺序订阅上游前一个流发送onComplete之后后一个流才开始执行正好符合“修复完成后再构建”。Flux.merge()会并发订阅数据按到达顺序交错适合并行汇聚事件源但不适合这种必须按阶段推进的链路。核心循环逻辑private FluxServerSentEventString 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 流式推给前端。注意这里所有阻塞操作都切到boundedElasticprivate FluxServerSentEventString 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) ); }); }为什么必须切线程Reactor 是“少量线程处理大量连接”的事件驱动模型核心线程负责分发和推进事件。把 build、文件读写、等待外部进程这类阻塞操作塞进去占住的不只是当前请求而是整个事件循环其他请求和 SSE 推送全受影响。Tomcat 那种“一请求一线程”模型下阻塞主要影响单个请求但 Reactor 下影响面是整个系统。构建与决策阶段private FluxServerSentEventString 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)); }afterBuild是重试决策点构建成功直接发 done 事件失败但没到上限就递归进入下一轮达到最大次数仍失败则返回最终失败结果。private FluxServerSentEventString afterBuild( FixLoopContext context, int attempt, BuildValidationResult buildResult) { FluxServerSentEventString 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)); }聊天记录的清理放在整个循环结束后用doFinally统一处理而不是每轮重试后立刻清。原因很直接每轮都清下一轮就拿不到上一轮已产生的修复信息一直不清长时间积累会增加内存压力甚至 OOM。Override public FluxServerSentEventString 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)); }); }清理逻辑本身很直接删掉对应 conversationId 的内存记录即可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); } }6. 一次失败重试的验证动作光看代码不够得实际验证重试链路。构造一个必然构建失败的场景在 SubAgent 修复前手动往源码里塞一个语法错误让第一轮 build 失败观察它是否自动进入第二轮。第一步确认当前版本目录和 metadata.json 里的 errorLog 存在cat ./versions/v3/metadata.json | jq .errorLog第二步触发修复接口用 curl 消费 SSE 流观察事件序列curl -N http://localhost:8080/app/sub-agent/fix?appId1001 \ -H Accept: text/event-stream正常应该看到这样的阶段推进event: phase data: {phase:FIXING,attempt:1} event: data data: {text:正在修复 src/pages/index.tsx 中的类型错误...} event: build data: {success:false,attempt:1,message:Build failed: Unexpected token} event: phase data: {phase:FIXING,attempt:2} event: data data: {text:根据新的错误日志修正 JSX 闭合标签...} event: build data: {success:true,attempt:2,message:Build passed, dist verified} event: done data: {success:true,attempts:2,summary:[Fix Summary] 修复了 JSX 标签闭合问题}关键验证点有三个第一轮build事件里success:false说明构建校验生效紧接着出现attempt:2的FIXING阶段说明重试被触发第二轮build成功且done事件带出[Fix Summary]说明闭环走通。第三步确认修复成功后前端把[Fix Summary]回传给主 Agent让主 Agent 感知这次修复结果而不是自己参与修复过程。这一步是主从 Agent 信息闭环的关键主 Agent 只消费摘要不消费中间的错误日志和修复尝试。如果重试 3 次仍失败done事件会返回success:false和attempts:3前端据此提示用户同时 metadata.json 和数据库状态已同步为失败不会留下脏状态。7. 本篇常见错排查报错一SSE 流卡住不推送前端一直转圈。大概率是阻塞操作没切线程。检查buildAndValidate、getErrorLogs、文件读写这些调用是否都包在Mono.fromCallable(...).subscribeOn(Schedulers.boundedElastic())里。只要有一个阻塞调用留在 Reactor 事件循环线程上整个流就会被占住。报错二第二轮修复带着第一轮的残留上下文。检查conversationId是否每次修复都重新生成。如果用了默认值或直接复用appId不同修复任务之间会上下文污染。确认generateSubAgentConversationId里带了 UUID 后缀并且通过advisorSpec.param(CONVERSATION_ID, conversationId)显式传入。报错三构建明明成功却还在重试。检查buildAndValidate的产物校验逻辑。只判断命令退出码不够还要确认dist目录下有有效产物。如果校验逻辑写反了成功会被判成失败导致无意义重试直到上限。报错四内存持续增长多轮修复后 OOM。检查cleanupConversationMemory是否真的在doFinally里被调用。如果清理逻辑抛异常被吞掉内存记录不会删除。看日志里有没有[SubAgent] Failed to clear chat memory的 warn。报错五TaoToken 请求 401 或超时。先确认TAOTOKEN_API_KEY环境变量在服务端进程里可见再确认 Base URL 是https://taotoken.net/api而不是首页地址。流式请求超时的话把timeoutSeconds调到 300 以上修复一轮本来就可能跑几分钟。排障和接入相关的配置细节对照接入文档查最快https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentKey 管理和额度查看在控制台https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content想先在对话里验证模型输出质量用模型对话页面试几轮https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content如果你要把这套闭环跑在长期编码或 Agent 任务上Coding Plan 的额度模型更合适多轮重试的消耗场景下不容易中途断档https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content整套闭环跑通后你会发现最值钱的不是“AI 能修代码”而是“修完能自动验证、失败能自动重来、结果能干净地交回主 Agent”。SubAgent 负责局部修复SSE 让过程可观察build 校验把主观判断变成客观结果自动重试让链路具备基础自愈能力。下一步可以考虑用容器隔离 AI 生成的代码让 build 在隔离环境里跑成功后再把产物复制出来展示这样连构建环境本身的风险也一起隔离掉了。