多智能体协作编程：从swarm-forge看AI软件开发流水线

最近在整理 AI 辅助编程相关资料时，我又翻到了unclebob/swarm-forge这个项目。单看名字，很容易联想起《代码整洁之道》作者 Robert C. Martin（Uncle Bob）：他用了大半辈子讲软件工匠精神、测试驱动开发、SOLID 原则；如今 AI 生成代码已经成为常态，他起一个叫swarm-forge的项目，显然不只是做一个“套壳 ChatBot”。

swarm是群体、蜂群，forge是锻造车间。把两个词拼在一起，可以理解成：让一组不同职责的 AI 智能体像工匠团队一样协作，把原始需求一步步“锻造”成可交付的代码工程。这种思路正是目前 AI 编程从“单点问答”走向“流水线生产”的一个典型方向。

本文会围绕unclebob/swarm-forge背后的多智能体软件开发模式展开，不逐字节介绍仓库命令——因为这类项目版本更新很快，命令和接口经常变化。重点是把核心设计思想、最小可运行示例、常见坑和工程落地建议讲清楚。适合对 AI 编程工具链感兴趣、想搭建自己的多智能体代码生成流程的开发者阅读。

1. 背景与核心概念

1.1 Uncle Bob 的理念：代码质量仍然是底线

Robert C. Martin 在国内被很多后端开发者称为“Bob 大叔”。他最有影响力的几件事是：

• 提出了 SOLID 设计原则；
• 大力推广测试驱动开发（TDD）；
• 写了《代码整洁之道》《架构整洁之道》《敏捷软件开发》等经典书籍；
• 长期在博客和演讲中强调“专业软件开发者要有工匠精神”。

很多人以为 Uncle Bob 会反对 AI 写代码，其实他更关心的是：AI 写出来的代码，是否也能满足高质量标准？如果 AI 一天生成 10 万行代码，但全是不分层、不测试、不可维护的“一次性代码”，那么软件系统的长期成本反而会上升。

所以swarm-forge这类项目，本质上不是“让 AI 多写点代码”，而是“再造一个软件开发团队”：

• 有人负责拆解需求；
• 有人负责设计模块；
• 有人负责写代码；
• 有人负责写测试；
• 有人负责代码评审；
• 有人负责修复失败用例。

1.2 为什么名字里有“swarm”

“Swarm”在分布式计算里经常指一组彼此协作的进程；在 AI 编程语境下，它通常指多个大模型智能体共同完成一项复杂任务。

举个例子：

• 一个智能体读需求，输出架构设计；
• 另一个智能体读设计，写业务代码；
• 第三个智能体读代码，写单元测试；
• 第四个智能体检查代码风格、边界条件、潜在 Bug。

这些智能体之间通过文件、文本、命令行输出交换信息，最终形成一个“流水线”。传统的单体 AI 对话是“一人分饰多角”，上下文一长就容易乱；而 swarm 模式是“专人专岗”，每个智能体只负责自己这一小段工作，上下文更聚焦，产出更稳定。

1.3 forge 的含义：把需求锻造成工程

Forge 是“锻造、铸造”。在软件开发行业，经常用“code forge”指代代码托管平台。swarm-forge把这个词带上，说明它更侧重“从 0 到 1 打磨出代码产物”，而不只是聊天。

从工程角度看，这个项目想解决的问题非常明确：

2. 多智能体协作模式拆解

2.1 单个 AI 编程助手的局限

2023 年初开始，很多开发者习惯用 AI 做结对编程。最常见的工作方式是把一大段需求粘贴给对话窗口，让模型一次生成整个项目。这种方式在几十行代码的小任务上效果不错，但一旦遇到真实业务模块，问题就开始暴露：

• 上下文窗口限制导致“看得见开头，忘了结尾”；
• 一次生成的代码没有设计文档支撑，模块边界混乱；
• 模型不会自己对代码做测试，错误只能在编译或运行时暴露；
• 反馈回路过长，改一个接口可能要重新粘贴几千字需求。

swarm-forge的核心改进，不是让模型变聪明，而是把“一整段工作”切分成多个智能体可以独立执行的小步骤，再通过流程引擎串联起来。

2.2 常见智能体编排模式

多智能体协作有几种常见模式：

1. 主从模式：一个主智能体负责任务拆分，多个子智能体并行执行；
2. 流水线模式：每个智能体按固定顺序处理任务，前一个的输出是后一个的输入；
3. 评审循环模式：代码生成后，由评审智能体给出反馈，再回到编码智能体修改，直到通过。

swarm-forge风格更接近“流水线 + 评审循环”。它既强调顺序协作，也强调失败反馈。

下面是一个简化后的流程：

需求文档 ↓ 架构智能体 → 设计文档 ↓ 编码智能体 → 代码文件 ↓ 测试智能体 → 测试文件 ↓ 执行 pytest ↓ 失败 → 把错误信息回传给编码智能体，重新生成代码 ↓ 通过 评审智能体 → 评审报告

这种模式的优点：

• 每个步骤都有明确输入输出；
• 失败时可以定点修复，而不是从头再来；
• 中间产物全部保留，方便审计；
• 便于接入 CI/CD 流程。

2.3 TDD 思想贯穿始终

Bob 大叔几十年一直在强调 TDD。在 AI 编程场景下，TDD 变成了一种非常实用的“Agent 控制手段”：

• 先让测试智能体根据需求写出测试；
• 再让编码智能体实现代码让测试通过；
• 测试通过后再让评审智能体做最终检查。

这样做的好处是，AI 生成的代码不是凭感觉拍板，而是必须通过可执行测试。只要测试写得好，代码质量就有了一个硬性下限。这也是我在实际使用所谓“AI 原生开发工具链”时，认为最值得借鉴的一点。

3. 环境准备与项目结构

3.1 环境要求

在搭建一个swarm-forge风格的最小工作流之前，我们需要准备以下环境：

• 操作系统：Linux、macOS 均可，Windows 建议使用 WSL；
• Python：3.10 或更高版本；
• 大模型 API：OpenAI 兼容接口即可，也可以用本地模型服务；
• 包管理工具：pip 或 pipenv；
• 版本控制：Git，用于保存每次智能体生成的中间产物；
• IDE：VS Code 或任意支持 Python 的编辑器。

如果你的环境版本与本文不一致，没关系。下面示例的重点是流程设计思路，版本差异只需在安装依赖时留意。

需要说明的是，本文示例默认使用一个可运行的 mock 模式。没有 API Key 也能看到完整流程；接入真实模型时，再设置对应的环境变量即可。

3.2 项目目录设计

建议按下面结构组织代码：

swarm-forge-demo/ ├── config.yaml # 智能体角色配置 ├── requirements.txt # Python 依赖 ├── requirements.md # 示例需求文档 ├── llm.py # 模型调用客户端 ├── forge.py # 工作流主程序 └── output/ # 中间产物输出目录

这个目录结构与实际的多智能体项目相比已经足够精简。真实项目中可能还会增加logs/、cache/、tests/等目录，便于追踪每次生成的历史记录。

3.3 配置文件：角色职责清晰化

# 文件路径：swarm-forge-demo/config.yaml swarm: max_retry: 3 output_dir: output roles: architect: "你是资深软件架构师，负责把需求拆成模块，并给出文件结构建议。" coder: "你是高级 Python 工程师，负责编写简洁、可运行的业务代码。" test_writer: "你是测试工程师，负责编写 pytest 单元测试。" reviewer: "你是代码评审专家，负责检查逻辑漏洞、边界条件和代码规范。"

这里有一个很关键的设计：每个角色只在自己的职责范围内工作。不要让一个智能体既写代码又写测试，也不要让它在回答问题时带出大量无关内容。

4. 完整实战：搭建一个 swarm-forge 风格的多智能体工作流

下面我们写一个最小可运行的多智能体工作流。它不是完整的unclebob/swarm-forge仓库，而是帮助你理解这类项目核心原理的演示代码。

4.1 依赖清单

# 文件路径：swarm-forge-demo/requirements.txt pyyaml requests pytest

安装命令：

pip install -r requirements.txt

如果网络环境受限，也可以只安装pyyaml，requests和pytest保持系统自带版本即可。

4.2 封装模型调用客户端

调用大模型 API 时，我建议统一封装一层客户端，避免业务代码里到处写 HTTP 请求。下面这个客户端支持 OpenAI 兼容的 Chat Completions 接口，也支持 mock 模式。

# 文件路径：swarm-forge-demo/llm.py import os import requests class LLMClient: def __init__(self, mock: bool = False): self.mock = mock def chat(self, prompt: str) -> str: if self.mock: print(f"[mock] 收到 prompt，长度 {len(prompt)} 字符") return ( "这是 mock 模式返回的内容。\n" "真实接入时，请设置 LLM_API_KEY，并确保模型接口可用。" ) api_key = os.environ["LLM_API_KEY"] endpoint = os.environ.get("LLM_ENDPOINT", "https://api.openai.com/v1/chat/completions") model = os.environ.get("LLM_MODEL", "gpt-4o-mini") payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.2, } response = requests.post( endpoint, headers={"Authorization": f"Bearer {api_key}"}, json=payload, timeout=120, ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"]

这里有几个细节：

• temperature设置为0.2，减少随机性，避免同样的需求每次生成结果差异过大；
• timeout设置为 120 秒，防止大模型推理时间过长导致连接挂起；
• 使用LLM_ENDPOINT环境变量，方便切换本地模型或第三方兼容接口。

4.3 编写需求文档

为了演示，我们准备一个简单的需求：实现一个计算器模块，支持加法和除法，并且除法需要考虑除数为 0 的场景。

# 文件路径：swarm-forge-demo/requirements.md # 需求：简单计算器模块 实现一个 Python 计算器模块 sample.py，要求： 1. 提供 add(a, b) 函数，返回两个数相加的结果； 2. 提供 divide(a, b) 函数，返回两个数相除的结果； 3. divide 遇到 a 或 b 不是数字时应抛出 TypeError； 4. divide 遇到除数为 0 时应抛出 ValueError； 5. 模块应能被 pytest 直接导入测试。

4.4 编写工作流主程序

工作流主程序负责：

• 读取配置；
• 判断是否使用 mock 模式；
• 依次调用架构、编码、测试、评审智能体；
• 执行 pytest，如果失败则回传给编码智能体修复；
• 保留每个阶段的输出文件。

# 文件路径：swarm-forge-demo/forge.py import os import re import subprocess import sys import pathlib import yaml from llm import LLMClient def load_config(path="config.yaml"): with open(path, "r", encoding="utf-8") as file: return yaml.safe_load(file) def write_output(output_dir, filename, content): output_path = pathlib.Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) (output_path / filename).write_text(content, encoding="utf-8") def extract_code(text: str) -> str: """从模型返回文本中提取 python 代码块，若没有代码块则返回原文。""" match = re.search(r"```python\s*(.*?)```", text, re.S) return match.group(1).strip() if match else text.strip() def build_prompts(config, requirements, design=None, code=None, feedback=None): roles = config["swarm"]["roles"] architect_prompt = ( f"角色：{roles['architect']}\n" f"请基于下面需求输出模块设计、文件结构和接口定义：\n{requirements}" ) coder_prompt = ( f"角色：{roles['coder']}\n" f"需求：\n{requirements}\n" f"架构设计：\n{design or '暂无'}\n" "请只输出 Python 代码，文件名为 sample.py，不要输出 Markdown 代码块以外的说明。" ) if feedback: coder_prompt += f"\n\n上轮测试失败信息：\n{feedback}\n请修复后重新输出完整 sample.py。" test_prompt = ( f"角色：{roles['test_writer']}\n" f"下面是 sample.py 的代码：\n{code or '暂无'}\n" "请输出 pytest 测试代码，文件名为 test_sample.py，只输出 Python 代码。" ) review_prompt = ( f"角色：{roles['reviewer']}\n" f"请审阅 sample.py 和 test_sample.py：\n{code or '暂无'}\n" "输出评审报告，包括逻辑问题、边界条件和改进建议。" ) return { "architect": architect_prompt, "coder": coder_prompt, "test_writer": test_prompt, "reviewer": review_prompt, } def run_pytest(output_dir): result = subprocess.run( [sys.executable, "-m", "pytest", output_dir, "-q"], capture_output=True, text=True, timeout=60, ) return result.returncode, result.stdout + result.stderr def main(): config = load_config() output_dir = config["swarm"]["output_dir"] # 未设置 API Key 时，自动进入 mock 模式 mock = os.environ.get("LLM_MOCK", "1") == "1" or "LLM_API_KEY" not in os.environ client = LLMClient(mock=mock) requirements = pathlib.Path("requirements.md").read_text(encoding="utf-8") # 第一步：架构智能体 prompts = build_prompts(config, requirements) print(">>> 架构智能体生成设计中...") design = client.chat(prompts["architect"]) write_output(output_dir, "design.md", design) # 第二步：编码智能体 print(">>> 编码智能体生成代码中...") code = client.chat(prompts["coder"]) code = extract_code(code) write_output(output_dir, "sample.py", code) # 第三步：测试智能体 print(">>> 测试智能体生成测试中...") tests = client.chat(prompts["test_writer"]) tests = extract_code(tests) write_output(output_dir, "test_sample.py", tests) # 第四步：执行测试，失败则回传修复 if mock: print(">>> mock 模式，跳过 pytest 执行") else: for attempt in range(1, config["swarm"]["max_retry"] + 1): print(f">>> 第 {attempt} 次 pytest 执行...") return_code, test_output = run_pytest(output_dir) if return_code == 0: print(">>> 测试通过") break print(f">>> 测试失败，原因：{test_output[-500:]}") prompts = build_prompts(config, requirements, design, code, test_output) new_code = client.chat(prompts["coder"]) new_code = extract_code(new_code) write_output(output_dir, "sample.py", new_code) code = new_code else: print(">>> 重试次数用完，需要人工介入") # 第五步：评审智能体 print(">>> 评审智能体输出评审报告...") review = client.chat(prompts["reviewer"]) write_output(output_dir, "review.md", review) print(">>> 工作流结束，中间产物目录：", output_dir) if __name__ == "__main__": main()

这段代码比较长，但逻辑很清晰：

• build_prompts负责按角色拼接 prompt，避免 prompt 散落在各个函数里；
• extract_code解决模型输出 Markdown 代码块的问题；
• write_output统一管理文件写入，保证每次生成结果都有记录；
• mock 模式不执行 pytest，因为 mock 返回的内容不是真实 Python 代码。

4.5 运行与验证

先跑一次 mock 模式：

cd swarm-forge-demo python forge.py

预期输出类似：

>>> 架构智能体生成设计中... [mock] 收到 prompt，长度 XXX 字符 >>> 编码智能体生成代码中... [mock] 收到 prompt，长度 XXX 字符 >>> 测试智能体生成测试中... [mock] 收到 prompt，长度 XXX 字符 >>> mock 模式，跳过 pytest 执行 >>> 评审智能体输出评审报告... >>> 工作流结束，中间产物目录： output

运行结束后，output/目录下会出现design.md、sample.py、test_sample.py、review.md四个文件。

如果要接入真实模型，需要先设置环境变量再运行：

export LLM_API_KEY="你的 API Key" export LLM_ENDPOINT="https://api.openai.com/v1/chat/completions" export LLM_MODEL="gpt-4o-mini" export LLM_MOCK="0" python forge.py

真实模型模式下，工作流会真正执行pytest，并根据测试结果决定是否把错误信息回传给编码智能体。

4.6 预期结果说明

在真实模型且需求清晰的前提下，你应当得到一个能通过pytest的sample.py和test_sample.py。例如sample.py可能长这样：

# 文件路径：swarm-forge-demo/output/sample.py def add(a, b): return a + b def divide(a, b): if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): raise TypeError("a and b must be numeric") if b == 0: raise ValueError("cannot divide by zero") return a / b

test_sample.py对应的测试可以覆盖：

• 两个正常数字相加；
• 两个数字相除；
• 除以 0 抛出ValueError；
• 非数字类型抛出TypeError。

5. 关键设计点分析

5.1 中间产物为什么重要

在多智能体工作流中，中间产物是“智能体之间的通信协议”。

如果架构智能体只输出一段话，然后编码智能体只记住第一句，后面就可能跑偏。所以每个角色必须把结果写成文件，下一个角色只需要读取文件，而不需要回忆之前的完整对话。

这也是swarm-forge这类项目和“在一个聊天窗口里反复追问”的本质区别：

• 聊天窗口是隐式状态；
• 文件系统是显式状态。

显式状态的好处是可以审计、可以回滚、可以重放。

5.2 失败回传机制的边界

我在代码里设置了max_retry: 3。这是很必要的：

• 如果模型第一次生成的代码有语法错误，回传错误信息通常能修复；
• 但如果需求本身矛盾，或者测试用例写得过于严格，重试再多轮也无法通过。

工程上的做法是：设置最大重试次数，超过后保留日志，交给人工介入，而不是让智能体无限循环。无限循环在 demo 里看起来很“智能”，在生产环境里只会消耗 API 费用。

5.3 角色隔离与 Prompt 规范

每个角色的 prompt 开头都声明了角色定位。这是一种“角色隔离”：

• 架构智能体不看测试代码；
• 测试智能体不看架构设计；
• 评审智能体则综合看到代码和测试。

角色隔离能让每个智能体的上下文保持精简，减少模型出错的可能性。但评审环节需要看完整代码，因为评审智能体就是要“找茬”的。

6. 常见问题与排查清单

在实践过程中，你可能会遇到下面这些问题。我把它们整理成一张表，方便快速定位。

下面展开讲几个高频问题。

6.1 模型输出格式不稳定

这是最常遇到的问题。模型有时候会输出：

好的，下面是代码： ```python def add(a, b): return a + b

如果直接把这段内容写入 `.py` 文件，pytest 根本跑不起来。解决方案就是 `extract_code`：优先用正则匹配 ```` ```python ```` 代码块；如果模型没有输出代码块，则把整段文本去空格后写入。 更保险的做法是同时在 prompt 中声明： ```text 请只输出 Python 代码，不要输出任何解释性文字，不要使用 Markdown 代码块。

6.2 测试智能体生成的测试不符合需求

测试智能体可能写出“无论如何都通过”的假测试，也可能写出过拟合需求的脆弱测试。

建议在测试智能体的 prompt 中加入验收条件清单，例如：

• 必须覆盖正常输入；
• 必须覆盖异常输入；
• 必须断言异常类型；
• 不允许只写assert True。

6.3 失败信息过长导致上下文溢出

回传测试失败信息时，不要直接把整个 pytest 输出全部粘贴给编码智能体。pytest 异常堆栈可能几百行，会浪费大量上下文。

实践中可以只截取最后 300 到 500 个字符，也可以自己解析 pytest 的失败摘要，结构化提取失败用例名和异常信息。

7. 工程化落地建议

搭建一个 demo 很容易，但要把它用到真实项目中，还需要考虑很多工程问题。

7.1 密钥与权限管理

永远不要把 API Key 写进代码仓库。推荐使用环境变量或专用的密钥管理服务（比如 Vault、云平台的 Secret Manager）。

如果你是在 CI 中运行swarm-forge流程，也要把密钥配置在 CI 的 Secret 中，而不是写在.gitlab-ci.yml或.github/workflows/*.yml里。

7.2 全量保留生成记录

每次运行工作流时，最好按时间戳建目录：

output/ └── 20250615-103000/ ├── requirements.md ├── design.md ├── sample.py ├── test_sample.py └── review.md

这样每个阶段的产物都是独立的，出了问题可以回溯到具体版本。

7.3 增加静态检查与安全扫描

pytest 只能保证“测试通过”，不能保证“代码安全”。建议在测试通过后继续增加：

• ruff或flake8做代码风格检查；
• bandit做 Python 安全扫描；
• mypy做类型检查（如果项目启用了类型标注）。

这些检查可以再次作为“评审智能体”的输入，让模型根据报告做修复。

7.4 控制 API 成本

多智能体流程的 API 费用比单次对话高，因为每个角色都会调用一次模型。控制成本可以从三方面入手：

1. 合理使用小模型：代码格式检查、简单测试生成用轻量模型；
2. 增加缓存：相同需求、相同 prompt 对应的结果可以缓存，避免重复调用；
3. 限制重试次数：不要无限“自我修复”。

7.5 安全边界

如果工作流生成的代码会被真正执行，必须在隔离环境中运行，尤其是 pytest 阶段。

千万不要在本地机器或生产服务器上直接运行一个由大模型生成的、内容未知的代码文件。建议使用容器或沙箱环境：

docker run --rm -v $(pwd)/output:/workspace python:3.11 bash -c "cd /workspace && pip install pytest && pytest -q"

7.6 人工评审不能缺位

无论多智能体流程设计得多完善，最终代码合入主干前，都应该有人工评审环节。AI 生成代码可以提速，但责任仍然在开发者身上。

8. 总结与下一步

通过unclebob/swarm-forge这个项目名称，我们可以看到一个清晰的趋势：AI 编程正在从“单模型聊天”走向“多智能体协作流水线”。

本文分享了一套最小工作流的设计与实现，核心要点包括：

• 把软件开发流程拆分为架构、编码、测试、评审等角色；
• 每个角色由一个独立智能体承担；
• 用文件系统作为智能体之间的通信接口；
• 用 pytest 作为代码正确性的硬性校验；
• 用失败回传机制实现自动修复循环；
• 最后用评审智能体做一次质量复盘。

如果你对这类项目感兴趣，下一步可以从这几个方向继续深入：

1. 给工作流增加更强健的 Prompt 模板，让每个角色的输出格式更稳定；
2. 接入本地模型，查看不同模型对最终代码质量的影响；
3. 把工作流接入 Git 提交钩子，实现提交前自动生成测试；
4. 研究 Agent 编排框架的并发设计，让多个业务模块并行生成代码；
5. 尝试结合 TDD，先让测试智能体写测试，再让编码智能体实现代码。

无论未来的开源项目叫什么名字，多智能体协作的思想已经确定会渗透到软件开发的日常流程中。对普通开发者来说，尽早理解这种工作方式，并动手搭建一个属于自己的“智能体小队”，会让后续学习成本低很多。你可以把本文的代码复制到本地，改一改配置，接上自己的模型服务，很快就能看到一套全自动“需求到代码”流水线真实跑起来的效果。