之前在一个中大型项目里尝试用 AI 编程助手做整模块开发发现一个很典型的问题任务一旦拆细Agent 的执行链路就会变得很长每个步骤都要等待上下文重新加载前面的设计决策到最后往往被“遗忘”或者走样。后来接触到 Agentic Coding 中的Wb-Flow思想核心就两个词Planned计划驱动和Parallel Waves并行波次。整套思路对提升 AI 编程的吞吐量和结果稳定性帮助很大。本文将围绕这个概念从原理拆解到可落地的执行流程给出完整示例和避坑方案。1. Agentic Coding 背景下为什么需要 Wb-Flow1.1 从“对话式编程”到“Agentic Coding”早期使用 AI 编程工具时我们处于“对话式辅助”阶段人给出一个需求AI 生成一段代码人再把代码粘回项目里遇到报错再继续对话修复。这种方式在单文件、小函数场景下很高效但面对“跨模块、多文件、多服务”的工程任务时痛点非常明显上下文窗口有限聊到后面前面的关键约束被截断或冲淡。代码生成是“单点式”的缺少全局设计约束。人需要不断把编译错误、运行日志、代码片段手动贴回对话框。Agentic Coding智能体编程则是把 AI 从一个“对话工具”升级为一个“可以自主执行任务的智能体”。它通常具备以下能力多步骤执行能自己拆解任务、调用工具、修改文件、运行命令。反馈闭环能读取编译结果、测试报告、运行时日志并根据反馈修正代码。记忆与状态在任务过程中维护一个状态列表知道自己现在做到哪一步、还需要做什么。文件级操作能直接读写项目文件而不是只生成代码片段。当你把任务交给一个 Agent 时理想状态下它会自动完成需求理解 → 方案设计 → 代码实现 → 测试验证 → 问题修复 → 最终交付。1.2 单 Agent 顺序执行的瓶颈最开始大家普遍采用“单 Agent 顺序执行”的方式也就是把任务拆成一个长长的待办列表Agent 从头到尾一项项执行任务 A1 → A2 → A3 → A4 → A5这种方式在任务量小的时候没什么问题但一旦遇到以下情况效率就会急剧下降任务链太长上下文窗口被逐步占满早期的关键约束容易被“遗忘”。后续任务依赖前序任务的输出任何一个环节卡住整个链路就阻塞。Agent 反复读取代码、分析上下文产生大量无效 Token 消耗。换句话说单 Agent 顺序执行的“单位并行度”是 1无论任务容量多大执行管道都只有一条吞吐量上不去。1.3 Wb-Flow 要解决的核心问题Wb-FlowWork-Breakdown Flow即工作分解流是一种面向 Agentic Coding 的执行编排模式。它不像传统方式那样让 Agent“线性地走完所有步骤”而是先把一个大任务拆解成一系列互相独立的子任务再根据依赖关系把它们划分到不同的“波次Waves”中同一个波次内的任务可以并行执行不同波次之间按计划串行推进。这种模式的关键在于两个词Planned先规划后执行。所有任务拆分、依赖分析、波次划分都在动手写代码之前完成。Parallel Waves同一波次内多个 Agent 并行工作波次之间靠依赖关系衔接从而大幅提升任务吞吐量。用一个简单的图来表示Wave 1: Task A1 Task B1 Task C1 \ | / v v v Wave 2: Task A2 Task B2 \ / v v Wave 3: Task Final同一个 Wave 内的任务互不依赖所以可以并行Wave 之间则保持严格的先后顺序保证下游任务拿到的是上游已经完成的成果。2. Wb-Flow 的核心设计思路2.1 工作分解结构WBSWb-Flow 的第一步和项目管理里的 WBSWork Breakdown Structure工作分解结构非常相似把一个宏大目标逐层拆解成具体、可执行、可验证的任务单元。例如你的目标是“开发一个用户中心模块”可以拆成层级任务项一级用户中心模块二级数据层、服务层、接口层、前端页面三级用户表的建表 SQL、MyBatis Mapper、用户 Service、Controller、登录页、注册页拆解的原则是任务粒度适中太粗的任务无法并行太细的任务会导致拆分和维护成本过高。任务尽量独立并行任务之间最好不要有太多共享可变状态。任务可验证每个子任务都要有明确的验收标准例如“接口能返回 200”“数据库迁移能成功执行”。2.2 依赖分析与波次划分任务拆解完成后需要分析任务之间的依赖关系。依赖关系一般分为三种无依赖Task A3 不依赖任何任务可以和任何任务并行。前驱依赖Task B2 必须等 Task B1 完成后才能开始。汇聚依赖Task Final 依赖多个前驱任务都完成。根据依赖关系可以把任务放进不同的“波次”Wave 1无依赖任务: - 建表 SQL - 创建项目骨架 - 定义 API 接口文档 Wave 2依赖 Wave 1 的任务: - 实现 Mapper 层依赖建表 SQL - 实现 Controller 骨架依赖接口文档 - 实现前端静态页面依赖项目骨架 Wave 3依赖 Wave 2 的任务: - 实现 Service 层 - 联调测试 Wave 4最终汇总: - 集成验证、修复问题一个波次内的任务全部完成后才统一进入下一个波次。这样做的好处是每一波都有明确的时间窗口Agent 不需要关心“我和其他任务之间的实时状态”只需要聚焦自己当前任务即可。2.3 规划阶段生成任务清单在 Wb-Flow 中“计划”不是一句空话而是要生成一份结构化的任务清单通常是一个 JSON 或 YAML 文件里面记录了任务 ID任务名称任务描述输入文件/依赖任务输出文件/产物验收标准所属波次Wave下面是一个示例{ project: user-center, waves: [ { wave: 1, tasks: [ { id: T1, name: create-user-table-sql, description: 创建用户表 t_user 的建表 SQL包含 id、username、password_hash、email、created_at、updated_at 字段, output: sql/v1_create_user_table.sql, acceptance: SQL 可以在 MySQL 8.0 中执行成功 }, { id: T2, name: create-project-skeleton, description: 创建 Spring Boot 项目骨架包含 pom.xml、主启动类、application.yml, output: pom.xml, src/main/java/..., src/main/resources/application.yml, acceptance: 项目可以通过 mvn compile 编译成功 }, { id: T3, name: define-api-contract, description: 定义用户模块的 REST API 接口文档包含注册、登录、获取用户信息三个接口, output: docs/api.md, acceptance: 接口路径、请求参数、响应格式定义完整 } ] }, { wave: 2, tasks: [ { id: T4, name: implement-mapper-layer, description: 基于 T1 的建表 SQL实现 MyBatis 的 Entity、Mapper 接口和 XML, dependencies: [T1], output: src/main/java/com/example/user/entity/, src/main/java/com/example/user/mapper/, acceptance: Mapper 接口方法返回值与表结构一致 } ] } ] }2.4 执行阶段并行波次任务清单生成后进入执行阶段。执行阶段通常由一个“调度器”或“编排引擎”来控制从任务清单中取出当前 Wave 的全部任务。为每个任务分配一个 Agent 执行实例。Agent 执行完任务后汇报结果包括状态成功/失败、产物文件、日志摘要。所有任务都完成后调度器汇总结果若涉及合并则执行合并然后进入下一个 Wave。如果某个任务失败调度器标记该任务为失败可单独重试或回滚到前一个稳定状态。3. 从零搭建一个 Wb-Flow 风格的 Agentic Coding 流程下面用一个实际案例来演示 Wb-Flow 的落地过程。我们的目标不是开发一个完整的框架而是演示“规划 并行波次”的执行思想如何应用到一个真实的开发流程中。3.1 环境准备与工具说明工具说明Python 3.10用于编写调度脚本Git用于代码版本管理任意支持 Agent 模式的 AI 编程工具例如 Cursor、Claude Code、Codex CLI、Aider 等用于执行代码生成任务一个示例项目本文用一个简单的 Python 计算器项目演示版本方面需要根据你实际使用的工具调整本文的核心是流程和思路不依赖某个特定工具的版本。3.2 定义任务清单我们先规划一个“Python 计算器 CLI 工具”的 Wb-Flow 任务清单。功能需求是支持 add、sub、mul、div 四个运算。支持从命令行读取两个数字和操作符。输出结果到终端。拆解后的任务清单如下Wave任务 ID任务内容依赖1T1创建项目结构和 README无1T2编写核心计算函数 calculator/operations.py无1T3编写命令行入口 calculator/cli.py无2T4编写单元测试 tests/test_operations.pyT22T5编写 README 使用说明T13T6集成测试与联调修复T4, T5这里有一个值得注意的设计T2 和 T3 都放在 Wave 1因为 CLI 入口虽然在逻辑上“使用”了核心函数但只要约定好接口两者就可以并行开发由 T4 和 T6 来校验合并结果。3.3 编写波次调度脚本我们可以用 Python 写一个轻量级的调度脚本读取任务清单 JSON然后并行执行每个任务对应的 Agent 命令。任务清单文件tasks.json{ project: calculator, waves: [ { wave: 1, tasks: [ { id: T1, name: init-project, description: 创建项目目录结构包含 calculator/ 包和 README.md, command: python agent_runner.py --task init-project }, { id: T2, name: write-operations, description: 在 calculator/operations.py 中实现 add/sub/mul/div 四个函数, command: python agent_runner.py --task write-operations }, { id: T3, name: write-cli, description: 在 calculator/cli.py 中实现命令行入口调用 operations 中的函数, command: python agent_runner.py --task write-cli } ] }, { wave: 2, tasks: [ { id: T4, name: write-tests, description: 编写 tests/test_operations.py覆盖四个运算函数, dependencies: [T2], command: python agent_runner.py --task write-tests }, { id: T5, name: write-docs, description: 完善 README.md补充安装与使用说明, dependencies: [T1], command: python agent_runner.py --task write-docs } ] }, { wave: 3, tasks: [ { id: T6, name: integration-test, description: 运行 pytest 并修复所有失败用例, dependencies: [T4, T5], command: python agent_runner.py --task integration-test } ] } ] }3.4 调度脚本核心实现下面是一个简化版调度脚本重点演示波次调度逻辑不依赖任何第三方库使用 Python 内置的concurrent.futures来实现并行执行。# 文件路径scheduler.py import json import subprocess from concurrent.futures import ThreadPoolExecutor, as_completed def load_tasks(json_path: str) - dict: 从 JSON 文件中加载任务清单 with open(json_path, r, encodingutf-8) as f: return json.load(f) def execute_task(task: dict) - dict: 执行单个任务命令返回任务执行结果 print(f[执行] {task[id]} - {task[name]}) try: result subprocess.run( task[command], shellTrue, capture_outputTrue, textTrue, timeout120, ) success result.returncode 0 return { id: task[id], name: task[name], success: success, stdout: result.stdout[-2000:], stderr: result.stderr[-2000:], } except subprocess.TimeoutExpired: return { id: task[id], name: task[name], success: False, stdout: , stderr: 任务执行超时, } def run_wave(wave_tasks: list) - bool: 并行执行一个波次内的所有任务 print(f\n 并行执行 {len(wave_tasks)} 个任务 ) results [] # 使用线程池并行执行max_workers 可以按需调整 with ThreadPoolExecutor(max_workerslen(wave_tasks)) as executor: future_to_task { executor.submit(execute_task, task): task for task in wave_tasks } for future in as_completed(future_to_task): task future_to_task[future] result future.result() results.append(result) status 成功 if result[success] else 失败 print(f[完成] {task[id]} - {task[name]} - {status}) if not result[success]: print(f错误信息: {result[stderr]}) return all(r[success] for r in results) def run_plan(plan: dict) - None: 按波次顺序调度执行整个计划 waves plan[waves] print(f项目: {plan[project]}) print(f总波次数: {len(waves)}) for wave in waves: wave_num wave[wave] tasks wave[tasks] success run_wave(tasks) if not success: print(f\nWave {wave_num} 存在失败任务停止后续执行。) return print(f\nWave {wave_num} 全部通过进入下一波。) print(\n所有波次执行完成。) if __name__ __main__: plan load_tasks(tasks.json) run_plan(plan)这个脚本的核心思想就是加载任务清单。按 Wave 分组。同一 Wave 内的任务通过线程池并行执行。每个任务通过subprocess调用真实的 Agent 命令。当前 Wave 全部成功后再进入下一波。3.5 Agent 任务执行脚本示例上面调度脚本里调用了agent_runner.py这是实际去调用 AI 编程工具的地方。这里以一个通用思路演示实际使用时需要根据你选择的工具做适配。# 文件路径agent_runner.py import argparse import subprocess import sys TASK_PROMPTS { init-project: 请创建项目目录结构包含 calculator/ 包和 README.md。, write-operations: ( 请在 calculator/operations.py 中实现以下四个函数\n def add(a, b): ...\n def sub(a, b): ...\n def mul(a, b): ...\n def div(a, b): ...\n 要求类型注解完整div 函数在除数为 0 时抛出 ValueError。 ), write-cli: ( 请在 calculator/cli.py 中实现命令行入口使用 argparse 接收两个数字和一个操作符 调用 calculator/operations.py 中对应函数并打印结果。 ), write-tests: ( 请编写 tests/test_operations.py使用 pytest 覆盖 add/sub/mul/div 四个函数的正常场景和异常场景。 ), write-docs: 请完善 README.md补充安装与使用说明。, integration-test: 请运行 pytest并修复所有失败的测试代码。, } def main(): parser argparse.ArgumentParser() parser.add_argument(--task, requiredTrue) args parser.parse_args() prompt TASK_PROMPTS.get(args.task, 请根据项目现状自行决策需要完成的开发任务) # 这里替换为你的 Agent 调用命令例如 # - Claude Code 的非交互模式 # - Cursor 的命令行接口 # - 自定义提示词管道 cmd [your-agent-cli, run, --prompt, prompt] print(fAgent 执行任务: {args.task}) # 实际开发中用下面的代码执行 Agent 命令 # result subprocess.run(cmd, capture_outputTrue, textTrue) # print(result.stdout) # sys.exit(result.returncode) # 示例中打印提示词便于观察任务内容 print(f提示词{prompt}) sys.exit(0) if __name__ __main__: main()这里需要特别说明不同 AI 编程工具的命令行接口差异很大而且版本更新很快所以上面的cmd只是一个思路示意。落地时要换成你实际环境的命令一般包含两个能力传入足够明确的提示词。让 Agent 在指定工作目录下执行文件读写操作。3.6 运行与预期结果运行调度脚本python scheduler.py预期输出大致如下项目: calculator 总波次数: 3 并行执行 3 个任务 [执行] T1 - init-project [执行] T2 - write-operations [执行] T3 - write-cli [完成] T1 - init-project - 成功 [完成] T2 - write-operations - 成功 [完成] T3 - write-cli - 成功 Wave 1 全部通过进入下一波。 并行执行 2 个任务 [执行] T4 - write-tests [执行] T5 - write-docs [完成] T4 - write-tests - 成功 [完成] T5 - write-docs - 成功 Wave 2 全部通过进入下一波。 并行执行 1 个任务 [执行] T6 - integration-test [完成] T6 - integration-test - 成功 所有波次执行完成。注意一个重点Wave 1 里的 T2 和 T3 虽然是并行执行的但两者通过“约定接口”来避免冲突——T3 知道要调用calculator/operations.py的函数但不需要等 T2 真正写完。真正的接口一致性校验发生在 Wave 3 的集成测试阶段。4. Wb-Flow 的关键细节与参数设计4.1 任务粒度的控制任务粒度过大并行度下降粒度过小任务间通信和合并成本反而超过并行收益。根据实践经验判断粒度是否合适的标准可以看两点每个任务是否能在 1 到 2 轮 Agent 交互内完成。任务产物是否是一个可验证的单元一个文件、一组接口、一个测试。例如把“实现整个登录模块”拆成“建表 SQL”“实体类”“Mapper 接口”“Service 逻辑”“Controller 接口”“单元测试”六个任务就比“Agent 你把这个模块写了”更适合并行执行。4.2 波次并行度的选择并行度不是越大越好。需要考虑几个因素上下文窗口限制同时运行太多 Agent可能每个人都加载了部分项目上下文内存消耗巨大。代码冲突风险如果多个 Agent 同时修改同一个文件合并时会非常痛苦。工具限流很多 AI 编程工具有分钟级或小时级调用限制。所以波次内的并行任务数量建议从 2 到 3 个起步运行稳定后再逐步提高。4.3 依赖与合并策略不同任务产生的结果最终要合并到一个项目中。合并策略一般有三种合并策略适用场景说明文件级合并各任务操作不同文件冲突最少最推荐模块级合并各任务操作同一模块不同子目录需要提前约定包结构代码级合并各任务操作同一文件冲突风险最高尽量避免实际执行时调度器会在每个 Wave 结束后执行一次“汇总检查”例如def merge_check(plan: dict, completed_tasks: list) - bool: 检查所有预期产物是否都存在 import os for task in completed_tasks: output task.get(output, ) if output: for path in output.split(,): path path.strip() if path and not os.path.exists(path): print(f[汇总检查] 缺少产物: {path}) return False return True4.4 失败重试与恢复并行环境中单个任务失败是常态关键是设计好失败恢复策略单个任务重试Agent 的执行结果不稳定失败后可以先重试一次仍然失败再人工介入。波次内隔离一个任务失败不影响同波次其他任务的执行。调度器应该等当前波次全部完成后再决定是否重试或中止。回滚策略Wave 2 失败时建议回到 Wave 1 的成果而不是直接对 Wave 2 进行部分修补。5. 实战中的常见问题与排查思路5.1 并行任务修改了同一个文件问题现象常见原因解决思路多个 Agent 同时修改utils.py代码互相覆盖任务边界划分不清晰在任务拆分时明确“文件所有权”每个文件只允许一个 Agent 修改合并时出现大量 Git 冲突并行任务之间共享了公共模块把公共模块提取到独立任务中优先完成后再进入并行波次核心原则并行任务之间要尽量实现“文件级隔离”。如果多个任务必须修改同一个文件就应该把它们放在同一个波次里串行执行或者拆成不同波次。5.2 Agent 上下文丢失任务执行走样问题现象常见原因解决思路Agent 在 Wave 2 里忘记了 Wave 1 约定的接口上下文太长被截断把关键接口约定写入项目内的AGENTS.md或CLAUDE.md文件同一任务重复执行时行为不一致提示词不够明确在任务描述里写清楚输入文件、输出文件、验收标准一个非常实用的做法是在每个 Wave 开始前把当前项目的“约定文件”比如接口签名、目录结构作为公共上下文注入到 Agent 的提示词中。这样可以大幅减少上下文丢失问题。5.3 并行执行反而比串行更慢问题现象常见原因解决思路4 个 Agent 并行写代码但总耗时比单个 Agent 顺序执行还长任务拆分过细调度与上下文加载开销大减少任务数量增大任务粒度多 Agent 同时启动工具调用被限流工具 API 限流降低max_workers或为不同 Agent 错开启动时间并行不是目的提高有效吞吐量才是目的。如果任务本身已经很小强行并行反而会产生负收益。5.4 失败的 Agent 任务反复重试不成功问题现象常见原因解决思路重试 3 次仍然失败提示词存在歧义或代码库本身有编译错误先检查代码库基线看看是不是基座就有问题单独重试成功但放在 Wave 里就失败并行任务之间产生了隐性依赖检查任务依赖关系把隐性依赖显式化在 Wb-Flow 中失败重试要有终止条件。一般建议最多重试 2 次仍然失败就停止该任务等人工介入调整任务描述之后再继续。6. Wb-Flow 落地最佳实践6.1 任务清单需写成代码仓库的一部分不要把任务清单放在临时目录里建议将tasks.json、调度脚本、提示词模板统一放到项目仓库的.wbflow/目录下.wbflow/ tasks.json prompts/ agent_system.md agent_user_template.md scripts/ scheduler.py agent_runner.py这样做的好处是任务清单可以走 Git 评审别人可以 review 任务拆得是否合理。提示词模板可以版本化管理方便回溯“哪次生成效果变了是因为哪次提示词改动”。新成员接手项目时可以快速了解项目的任务结构。6.2 约定项目级公共上下文文件在项目根目录维护一个公共上下文文件例如AGENTS.md里面记录所有 Agent 必须遵守的约束项目的技术栈与版本。目录结构规范。命名规范。关键接口签名。测试执行命令。每个 Agent 在执行任务前都需要先读取这个文件。这样即使多个 Agent 并行运行它们对项目的基本认知也是一致的。6.3 任务验收标准要可自动化“任务完成”不能靠感觉要尽可能写成可自动化的验收命令任务类型自动验收方式编写 SQL在测试库执行该 SQL检查是否成功编写 Java 类执行mvn compile确认编译通过编写接口启动服务后调用接口检查状态码编写算法函数执行单元测试检查断言通过调度器在 Agent 完成任务后应该自动运行对应的验收命令而不是直接信任 Agent 的“完成”汇报。6.4 安全与权限边界在执行 Wb-Flow 时Agent 拥有文件读写、命令执行等能力所以安全边界需要提前设好工作目录隔离Agent 只能操作指定项目目录不允许访问系统其他目录。敏感信息保护数据库密码、API Key 等不要写入任务提示词中优先使用环境变量注入。命令白名单可以在调度器中对 Agent 执行的命令做白名单限制避免执行危险命令。变更可回滚整个项目应处于 Git 版本管理之下每个 Wave 执行前打一个 tag便于回滚。6.5 从单 Agent 到多 Agent 的渐进式切换如果你的团队目前还在用单 Agent 开发不建议直接切换到大规模并行。可以按以下步骤渐进式推进第一阶段——串行 任务清单先把大任务拆成有序任务清单Agent 按顺序逐个完成培养任务拆分习惯。第二阶段——局部并行选择两个完全独立的任务比如“写前端页面”和“写后端接口”尝试放在同一波次并行执行。第三阶段——完整 Wb-Flow跑通完整的波次调度、依赖管理、失败重试、结果合并流程。这样每个阶段都能积累经验同时把风险控制在可接受的范围内。7. 一个更贴近真实业务的拆解示例为了帮助读者进一步理解这里再给出一个更贴近后端业务开发的 Wb-Flow 示例主题是“用户订单查询接口”。需求给前端提供一个订单分页查询接口。初步拆解Wave任务 ID任务内容依赖1A1编写订单表 order 的建表 SQL无1A2定义订单查询接口的 OpenAPI 文档无1A3创建订单查询 DTO 类无2A4编写 Mapper 层分页查询订单A12A5编写 Service 层处理分页参数和异常A2, A42A6编写 Controller 层接入接口A2, A33A7联调测试并修复问题A5, A6这里的设计就很有意思A3DTO 类虽然从逻辑上讲是 A6 的依赖但只要你提前把 DTO 的字段定义约定好A3 完全可以放在 Wave 1 并行完成A6 在 Wave 2 直接使用即可。实际运行时调度器在 Wave 1 完成后的“合并检查”里会重点检查SQL 文件是否存在。API 文档是否完整。DTO 类是否定义好了字段。只要这三个产物存在Wave 2 的三条流水线就可以并行开工。整个过程就像流水线车间一样不同工位互不打扰最后一个 Wave 再做总装和质检。8. 总结与实践建议Wb-Flow 用一句话概括就是先花时间把任务拆清楚、把依赖理清楚、把波次排清楚然后让多个 Agent 在同一个波次里并行干活。这套方法比较适合以下场景项目功能边界清晰可以拆成多个相对独立的模块。任务量较大单 Agent 顺序执行时间过长。团队已经具备一定提示词工程和 Agent 编排能力。不太适合的场景项目代码耦合极重任务之间大量共享可变状态。任务本身非常小几分钟就能完成。Agent 工具稳定性不足频繁执行错误。落地时最值得优先关注的三个风险点任务边界划分要保证文件级隔离、每个任务都要有自动验收方式、失败重试要有次数上限。把这三件事做好Wb-Flow 的稳定性就能得到基本保障。建议从一个小型项目开始按上文“渐进式切换”的步骤先跑通串行任务清单再尝试局部并行最后再上完整的波次调度。每个阶段都记录下任务拆解的时间和 Agent 执行的时间用数据判断并行是否真的带来了收益。