经典 SOP 流水线：一行需求跑成一个软件仓库

30 秒导读： 你敲一句"做个 2048 小游戏"，MetaGPT 会像一家软件公司一样，把它依次交给 产品经理 → 架构师 → 项目经理 → 工程师 → 测试工程师，最后吐出一个能跑的代码仓库。本章讲 清楚这条"流水线"是怎么用 set_actions + _watch 静态拼出来的——没有中央调度器，全靠 每个角色声明"我产什么、我盯谁的产物"，就自动接力成了一条 DAG（有向无环图）。

本章把前三章的抽象落地。你已经知道：

• 单个角色怎么"观察-思考-行动"地循环（见 01-role-loop.md）；
• 一个 Action / ActionNode 怎么把 LLM 输出变成结构化产物（见 02-action-actionnode.md）；
• 消息怎么在 Environment 总线上路由、Team 怎么驱动一轮轮跑（见 03-environment-message-bus.md）。

这一章只讲一件新事：把这些零件按"软件公司标准作业流程（SOP, Standard Operating Procedure）"接成一条端到端的流水线。这就是 MetaGPT 论文里 Code = SOP(Team) 那个等式的 第一代实现。

---

1. 这是什么（零基础也能懂）

一句话定义： 经典 SOP 流水线 = 把"一句话需求"变成"一个软件仓库"的固定装配线，装配线上 的每个工位是一个 AI 角色。

它解决什么问题： 一个人让 LLM"直接写个游戏"，往往得到一坨没结构、跑不起来的代码。真实 软件公司不这么干——先写需求文档、再出系统设计、再拆任务、再编码、再测试。MetaGPT 把这套 人类的分工流程照搬给一群 AI，让它们各司其职、依次接力。

用起来什么样： 就一行命令。

metagpt "Create a 2048 game"

跑完，你会在工作区里得到一个真实目录：需求文档、系统设计、任务清单、源码、单元测试，各就各位。

一句话直觉： 把它想成一条工厂流水线。原料是"一句话需求"，五个工位依次加工，每个工位 只做自己那道工序，把半成品往下一个工位传。关键是——没有工头站在中间喊"下一个"；每个工位 自己盯着上游传送带，看到属于自己的半成品就动手。这个"自己盯上游"的机制，就是本章的核心。

---

2. 顶层全景（这条线大概怎么转）

2.1 五个工位与它们的产物

经典流水线由五个角色组成，一字排开：

工位（角色）	profile	核心动作（产物）	产出物	源码
产品经理	Product Manager	WritePRD	需求文档 PRD	roles/product_manager.py
架构师	Architect	WriteDesign	系统设计（类图/时序图）	roles/architect.py
项目经理	Project Manager	WriteTasks	任务清单 + 依赖包	roles/project_manager.py
工程师	Engineer	WriteCode / WriteCodeReview / SummarizeCode	源代码	roles/engineer.py
测试工程师	Qa Engineer	WriteTest / RunCode / DebugError	单元测试 + 运行结果	roles/qa_engineer.py

2.2 一张图看懂"接力"

怎么读这张图：从左到右是数据流；每根箭头上标的是"消息的 cause_by 标签"——也就是 "这条半成品是被哪个动作产出来的"。下游角色靠订阅这个标签把半成品接住。

 一行需求
 (UserRequirement)
      │
      ▼
┌───────────────┐  cause_by=WritePRD   ┌───────────────┐
│ ProductManager│ ───────────────────▶ │  Architect    │
│  产 WritePRD  │                      │ watch WritePRD│
└───────────────┘                      │ 产 WriteDesign│
                                       └───────┬───────┘
                                               │ cause_by=WriteDesign
                                               ▼
                                       ┌───────────────┐
                                       │ ProjectManager│
                                       │watch WriteDesign
                                       │  产 WriteTasks │
                                       └───────┬───────┘
                                               │ cause_by=WriteTasks
                                               ▼
                                       ┌───────────────────────────┐
                                       │ Engineer                  │
                                       │ watch WriteTasks          │
                                       │ WriteCode→(CodeReview)→   │
                                       │ SummarizeCode             │
                                       └───────┬───────────────────┘
                                               │ cause_by=SummarizeCode
                                               │ send_to="Edward"
                                               ▼
                                       ┌───────────────────────────┐
                                       │ QaEngineer                │
                                       │ watch SummarizeCode       │
                                       │ WriteTest→RunCode→DebugError
                                       └───────────────────────────┘
                                          （若测出 bug，回传 Engineer）

一句话主线： 需求进 → PM 写 PRD → 架构师看到 PRD 就出设计 → 项目经理看到设计就拆任务 → 工程师看到任务就写码、审码、汇总 → 测试工程师看到汇总就写测试、跑测试、修错。全程没有一个 角色"点名"下一个角色（除了最后两步用了直达地址），每个角色只是盯着自己关心的动作标签。

---

3. 核心原理：set_actions + _watch = 静态接线

这一节是全章的心脏。我们要回答一个问题：既然没有中央调度器，这条流水线的"顺序"到底刻在哪？

答案：刻在每个角色 __init__ 里的两行装配语句。

3.1 两个动词：一个声明"我产什么"，一个声明"我盯谁"

每个经典角色在初始化时都做两件事：

• set_actions([...]) —— 声明我会执行哪些动作（我的产出能力）。
• _watch({...}) —— 声明我订阅哪些动作产出的消息（我的输入来源）。

_watch 的实现极简：把动作类转成字符串标签，塞进一个集合 rc.watch （roles/role.py:288 _watch → self.rc.watch = {any_to_str(t) for t in actions}）。

到了观察阶段，角色只从消息流里挑两类消息：cause_by 命中 rc.watch 的，或者 send_to 点了自己名字的：

# roles/role.py:410 _observe —— 消息过滤的一行真相（节选）
self.rc.news = [
    n for n in news
    if (n.cause_by in self.rc.watch or self.name in n.send_to)
    and n not in old_messages
]

这就是"接力"的全部秘密。 上游角色产出消息时，会把消息的 cause_by 标成"是我哪个动作产 的"；下游角色只要在 _watch 里订阅了那个动作，_observe 就会自动把这条消息捞进来。两行装配 语句，一进一出，就在角色之间连了一条有向边。

3.2 边是怎么一条条连起来的

把五个角色的 set_actions / _watch 排在一起，DAG 就浮现了：

角色	set_actions（产出，锚点）	_watch（订阅，锚点）
ProductManager	[PrepareDocuments, WritePRD]（product_manager.py:45）	[UserRequirement, PrepareDocuments]（product_manager.py:46）
Architect	[WriteDesign]（architect.py:49）	{WritePRD}（architect.py:52）
ProjectManager	[WriteTasks]（project_manager.py:40）	[WriteDesign]（project_manager.py:41）
Engineer	[WriteCode]（engineer.py:105）	[WriteTasks, SummarizeCode, WriteCode, WriteCodeReview, FixBug, WriteCodePlanAndChange]（engineer.py:106）
QaEngineer	[WriteTest]（qa_engineer.py:58）	[SummarizeCode, WriteTest, RunCode, DebugError]（qa_engineer.py:59）

读法： 看"谁的 set_actions 里的动作，出现在谁的 _watch 里"，就是一条边。

• 架构师 _watch({WritePRD})，而 PM 的 set_actions 产 WritePRD → PM ▶ 架构师。
• 项目经理 _watch([WriteDesign])，架构师产 WriteDesign → 架构师 ▶ 项目经理。
• 工程师 _watch 里有 WriteTasks，项目经理产它 → 项目经理 ▶ 工程师。
• 测试工程师 _watch 里有 SummarizeCode，工程师产它 → 工程师 ▶ 测试工程师。

注意工程师和测试工程师的 _watch 列表里，还包含了自己的动作（WriteCode、WriteTest 等）——这不是笔误，而是让角色能观察到自己发给自己的消息，从而在内部多步之间自转（3.5 详述）。

3.3 为什么这叫"静态 DAG"

关键在于：这些 set_actions / _watch 全部写死在 __init__ 里，跑之前就定了，跑的时候一个字 都不改。所以流水线的拓扑是静态的——需求还没进来，谁接谁就已经确定。

这正是它和第五章 MGX/RoleZero 的分水岭：那里 TeamLeader 在运行时动态决定下一步派谁 （见 05-mgx-rolezero.md）；而这里，顺序是编译期焊死的。

一句话记住它： 经典 SOP 流水线 = 用 _watch(cause_by) 把 set_actions 的产物静态连成 一张接力网。你增删一条边，就是改一处 _watch。

3.4 一个玩具版，把"接力"演出来

下面这段示意、非源码，用最少的代码复现 3.1 的机制，帮你建立直觉：

# 示意，非源码：接力机制的最小骨架
class Role:
    def __init__(self):
        self.produces = []   # 对应 set_actions
        self.watches = set() # 对应 _watch

    def observe(self, bus):
        # 只捞 cause_by 命中我订阅的消息（对应 _observe 的过滤）
        return [m for m in bus if m.cause_by in self.watches]

architect = Role()
architect.produces = ["WriteDesign"]
architect.watches = {"WritePRD"}       # 我盯 PM 的产物

# PM 产出一条 cause_by="WritePRD" 的消息，丢进总线
bus = [Message(content="PRD...", cause_by="WritePRD")]
print(architect.observe(bus))          # 架构师立刻接住它 → 该他干活了

重点看：架构师从没被谁通知过，它只是持续盯着总线上 cause_by="WritePRD" 的消息。产物一 出现，它就自动被激活。真实实现把"盯"落在 _observe 的列表推导上（role.py:410）。

3.5 两种接力方式：主题订阅 vs 直达地址

流水线其实混用了两种路由：

• 主题订阅（cause_by）：用于 PM→架构师→项目经理→工程师 这几步跨角色接力。发送方不关心 谁接，只标"我是哪个动作产的"；接收方靠 _watch 自取。这是松耦合的主力。
• 直达地址（send_to）：用于两个"点名"场景。工程师汇总完代码后，直接把消息寄给测试工程师 的名字 "Edward"（engineer.py:227 send_to="Edward"）；测试工程师跑完测试若发现是开发代码 的锅，就把消息寄回工程师的名字 "Alex"（qa_engineer.py:126,146）。

为什么最后两步用直达？因为工程师和测试工程师内部各有多步自循环，需要精确点名（发给自己 或发给对方），主题广播不够用。工程师内部 WriteCode → SummarizeCode 靠给自己发消息推进 （engineer.py:170 send_to=MESSAGE_ROUTE_TO_SELF）。

---

4. 端到端走一条真实路径（需求 → 归档）

现在把一句"做个贪吃蛇"从头跟到尾，看每一棒交给谁、产出什么、cause_by 标成什么。

4.1 第 0 棒：产品经理写 PRD

产品经理在经典 SOP 模式下（use_fixed_sop=True），按 BY_ORDER 依次跑两个动作：先 PrepareDocuments（初始化 git 仓库、落需求文件），再 WritePRD （product_manager.py:45-47）。

WritePRD.run 用 WRITE_PRD_NODE（一个 ActionNode，见第 2 章）把需求填成结构化 PRD，落盘， 然后返回一条**打了 cause_by=self（即 WritePRD）**的消息，instruct_content 里带上 changed_prd_filenames（write_prd.py:180-189）：

# actions/write_prd.py:180 —— PRD 完成后回传的消息（节选）
return AIMessage(
    content="PRD is completed. " + ...,
    instruct_content=AIMessage.create_instruct_value(kvs=kvs, class_name="WritePRDOutput"),
    cause_by=self,   # 标签 = WritePRD；架构师正盯着这个
)

4.2 第 1 棒：架构师出系统设计

架构师 _watch({WritePRD})，于是 _observe 捞到上面那条消息，激活 WriteDesign。它读 changed_prd_filenames，用 DESIGN_API_NODE 生成"数据结构与接口""程序调用流"等设计，还会把 其中的 mermaid 存成类图/时序图 SVG，最后回传 cause_by=self（WriteDesign）、带 changed_system_design_filenames 的消息（design_api.py:166-175）。

4.3 第 2 棒：项目经理拆任务

项目经理 _watch([WriteDesign])，接住设计，跑 WriteTasks。它把设计拆成任务文件清单，同时 把每个文件需要的第三方包汇总进 requirements.txt（project_management.py:165-176 _update_requirements），回传 cause_by=self（WriteTasks）、带 changed_task_filenames 的消息 （project_management.py:123-132）。

4.4 第 3 棒：工程师写码 → 审码 → 汇总

工程师是流水线上最忙的工位，它一棒里其实跑了三个子动作，靠内部自循环串起来。

工程师 _think 看到 cause_by=WriteTasks，就调 _new_code_actions：解析任务清单里的文件列表， 为每个文件建一个 WriteCode 待办（engineer.py:280-283, 408-428）。

_act 根据当前待办类型分派（engineer.py:159-165）：

# roles/engineer.py:159 —— 工程师内部的"下一棒是谁"分派（节选）
if isinstance(self.rc.todo, WriteCode):
    self.next_todo_action = any_to_name(SummarizeCode)  # 写完码，下一步去汇总
    return await self._act_write_code()
if isinstance(self.rc.todo, SummarizeCode):
    self.next_todo_action = any_to_name(WriteCode)
    return await self._act_summarize()

三个子动作各干一件事：

子动作	干什么	源码锚点
WriteCode	按设计+任务，逐文件生成代码（一次只写一个文件）	actions/write_code.py:100 WriteCode.run
WriteCodeReview	可选：审代码，判 LGTM/LBTM，LBTM 就重写	actions/write_code_review.py:148 write_code_review_and_rewrite
SummarizeCode	通读全部文件找 bug、画调用图、列出待改清单	actions/summarize_code.py:104 SummarizeCode.run

WriteCode 的巧思：写某个文件时，会把同任务下的其它文件代码塞进 prompt 作上下文，但排除 当前文件自己（write_code.py:167 get_codes(..., exclude=...)），这样每个文件都"知道"邻居的 接口，又不会拿旧的自己去污染新生成。审码环节靠解析响应里的 LGTM/LBTM 决定是否重写 （write_code_review.py:151-153）——LGTM（Looks Good To Me）放行，LBTM 打回重写。

SummarizeCode 通过后，工程师把一条 cause_by=SummarizeCode、直达 "Edward"（测试工程师） 的消息寄出去（engineer.py:217-227），第 3 棒交接完成。

4.5 第 4 棒：测试工程师写测试 → 跑测试 → 修错

测试工程师 _watch([SummarizeCode, WriteTest, RunCode, DebugError])，接住 SummarizeCode 的消息 后，_act 根据消息的 cause_by 在三个子动作间循环（qa_engineer.py:181-195）：

 收到 SummarizeCode ─▶ _write_test  (WriteTest)  写单测，自发消息
        ▲                    │
        │                    ▼
   收到 RunCode ◀──── _run_code   (RunCode)   跑单测，看输出
   → _debug_error         │
     (DebugError)         │  测试结果里的 "Send To"
     修完自发消息 ─────────┘  = Engineer? → 回传 "Alex"
                             = QaEngineer? → 自己再修

三个子动作：

子动作	干什么	源码锚点
WriteTest	给每个源码文件生成 unittest 测试	actions/write_test.py:57 WriteTest.run
RunCode	真在子进程里 python 跑测试，抓 stdout/stderr	actions/run_code.py:92 RunCode.run_script（subprocess.Popen）
DebugError	读运行日志，判断该改开发码还是测试码并重写	actions/debug_error.py:55 DebugError.run

两个值得记的细节：

• RunCode 真的执行代码。 它 subprocess.Popen 起进程、装依赖、带 10 秒超时 （run_code.py:106-118）。跑完让 LLM 总结结果，并在总结里写一个 Send To 字段决定甩锅给谁 （开发码错→Engineer，测试码错→QaEngineer）。_run_code 解析这个字段，把消息寄回对应的人 （qa_engineer.py:125-147）。
• DebugError 会先检查"是不是已经过了"。 它用正则 Ran (\d+) tests in ... OK 匹配 stderr，命中 就直接返回、不折腾（debug_error.py:60-63）。这是廉价的提前退出。

整个测试回合最多转 test_round_allowed = 5 轮（qa_engineer.py:47, 164），超了就停，避免无限修。

至此，一句"做个贪吃蛇"已经变成了磁盘上的：docs/prd/*、docs/system_design/*、docs/tasks/*、 源码目录、tests/*、requirements.txt——一个完整的软件仓库。

---

5. 巧妙之处（可借鉴的技术）

① 用"标签订阅"取代"函数调用"来编排流水线。 传统写法是 pm.run(); arch.run(pm.output); ... ——调用方必须知道全流程。MetaGPT 反过来：每个角色只声明产出（set_actions）和订阅 （_watch），编排逻辑不集中在任何地方，而是分布在各角色的两行装配里。加一个工位＝改一处 _watch，不用动主流程。锚点：role.py:410 的 _observe 过滤。

② 消息传"文件名"而非"文件内容"。 角色间的消息 instruct_content 里放的是 changed_prd_filenames / changed_system_design_filenames 这类路径，不是几十 KB 的文档正文 （贯穿 write_prd.py:174 / design_api.py:162 / project_management.py:119）。下游要用时按名去 仓库读。好处：消息轻、总线不膨胀、天然支持增量（只重算变过的文件）。

③ 写单个文件时喂"邻居"、藏"自己"。 WriteCode.get_codes(exclude=当前文件) （write_code.py:167）让每个文件的生成都带着同项目其它文件的接口上下文，却不会拿旧版自己干扰 新生成——这是让多文件项目彼此接口对得上的关键小技巧。

④ 审码用一个词做门闸。 代码评审不返回复杂结构，只让 LLM 写 LGTM/LBTM 一个词 （write_code_review.py:62, 151），LBTM 就带着评审意见重写。极简、可靠、易解析。

---

6. 边界与局限（诚实说）

这条经典流水线在当前 commit 里，大部分是"关着的"。 这一点务必讲清楚，否则读代码会困惑。

看默认入口 software_company.py 的组队逻辑（software_company.py:45-61）：

# metagpt/software_company.py:45 —— 默认团队（节选）
company.hire([
    TeamLeader(),
    ProductManager(),
    Architect(),
    Engineer2(),
    # ProjectManager(),
    DataAnalyst(),
])
# if implement or code_review:
#     company.hire([Engineer(n_borg=5, use_code_review=code_review)])
# if run_tests:
#     company.hire([QaEngineer()])

由此能读出几件事：

• 经典 Engineer / QaEngineer 的装配被整段注释掉了（software_company.py:56-61）， ProjectManager() 也被注释（software_company.py:51）。默认团队走的是新一代 Engineer2 + DataAnalyst + TeamLeader，即第五章的 MGX/RoleZero 路线（见 05-mgx-rolezero.md）。
• ProductManager / Architect 虽还在默认团队里，但已被重写成 RoleZero 子类。 它们本章描述 的经典 SOP 接线（set_actions([PrepareDocuments, WritePRD]) + _watch），只有在 use_fixed_sop=True 时才生效（product_manager.py:43 if self.use_fixed_sop:）；架构师的经典 _watch({WritePRD}) 也在注释里标注"仅当 use_fixed_sop 改为 True 才有效" （architect.py:46-52）。

所以本章讲的是"第一代实现"的骨架——它在源码里完整保留、可被 use_fixed_sop 或旧版路径 激活，是理解 Code=SOP(Team) 思想的最佳标本，但不是当前默认跑的那条路。

其它已知局限：

• 拓扑写死。 想插一个"安全审计"工位、或让某步并行，都得改角色的 __init__ 装配，不能在运行时 动态编排——这正是 MGX 要解决的痛点。
• RunCode 的沙箱很薄。 它直接在本机 subprocess 跑生成的代码、还会 pip install （run_code.py:103, 148-173），只有 10 秒超时兜底，没有真正的隔离沙箱。
• 一次只写一个文件。 WriteCode 刻意约束"THIS ONLY ONE FILE"（write_code.py:77），跨文件的 大改动依赖任务拆分的粒度，拆不好就写不长。

---

7. 横向对比（同 shelf 内）

维度	经典 SOP 流水线（本章）	MGX / RoleZero（第五章）
编排时机	编译期静态（set_actions+_watch 写死）	运行期动态（TeamLeader 现场派活）
谁决定下一步	没人——靠 cause_by 标签自动接力	TeamLeader 看局面调度
角色间耦合	松（主题订阅为主，末端才点名）	命令式（TeamLeader 直接指派）
适合场景	流程固定的标准软件开发	开放、多变、需要临场决策的任务
代表角色	ProductManager/Architect/…/QaEngineer	TeamLeader/Engineer2/DataAnalyst

想先理解单个角色的"观察-思考-行动"循环，回 01-role-loop.md；想理解每个 WriteXxx 动作内部怎么把 LLM 输出变结构化，回 02-action-actionnode.md； 想理解消息在总线上具体怎么发/怎么收，回 03-environment-message-bus.md。

---

8. 代码地图（导航索引）

用符号名 grep 比行号更抗漂移；下表按"你想看什么"组织。

主题	文件	关键符号
接力过滤的一行真相	metagpt/roles/role.py	Role._observe（cause_by in rc.watch or name in send_to）
订阅集合怎么建	metagpt/roles/role.py	Role._watch → self.rc.watch
默认组队（经典装配被注释）	metagpt/software_company.py	generate_repo / company.hire([...])
PM 经典接线（仅 fixed_sop）	metagpt/roles/product_manager.py	ProductManager.__init__（set_actions+_watch）
架构师订阅 WritePRD	metagpt/roles/architect.py	Architect.__init__（_watch({WritePRD})）
项目经理订阅 WriteDesign	metagpt/roles/project_manager.py	ProjectManager.__init__（set_actions([WriteTasks])）
工程师内部三步自循环	metagpt/roles/engineer.py	Engineer._act / _new_code_actions / _act_summarize
测试工程师内部三步自循环	metagpt/roles/qa_engineer.py	QaEngineer._act / _write_test / _run_code / _debug_error
写 PRD 并回传 cause_by	metagpt/actions/write_prd.py	WritePRD.run
出系统设计	metagpt/actions/design_api.py	WriteDesign.run
拆任务 + 汇总依赖包	metagpt/actions/project_management.py	WriteTasks.run / _update_requirements
逐文件写码、喂邻居藏自己	metagpt/actions/write_code.py	WriteCode.run / get_codes(exclude=...)
审码 LGTM/LBTM 门闸	metagpt/actions/write_code_review.py	write_code_review_and_rewrite
通读找 bug、画调用图	metagpt/actions/summarize_code.py	SummarizeCode.run
写单测	metagpt/actions/write_test.py	WriteTest.run
子进程真跑测试	metagpt/actions/run_code.py	RunCode.run_script（subprocess.Popen）
判过/修错	metagpt/actions/debug_error.py	DebugError.run（Ran N tests ... OK 正则）