舞台工艺 × 多智能体：用两张专家团和一张状态机总表，把「审图→选题→写作」跑成自动流水线

舞台工艺 × 多智能体：用两张专家团和一张状态机总表，把「审图→选题→写作」跑成自动流水线

一、背景：两个团各自能跑，但连不上

我在本地自建了两个彼此独立的专家包：一个负责舞台工艺图纸审核，一个负责内容生产。它们的目录完全隔离，互不 import、互不 spawn，连版本号都是各走各的——审图包当前是v1.3.0，内容包是v1.1.0。

设计上这两个包是刻意"解耦"的：

• 上游（审图包）：接图纸、出审核意见、出交付件。审完一个项目，脑子里会留下大量真实素材——某个参数在几张图上不一致、某批交付件格式全线翻车、某次分级判定的边界争议。这些素材极有价值，但这个包有一条硬边界：只提炼、不写作。
• 下游（内容包）：每天要出稿，最缺的不是文笔，而是"有实证、可追溯、已脱敏"的选题。它同样有硬边界：不审图、不自创离散主题。

问题就出在交界处。最早我靠对话口头转述：审完一个项目，在会话里跟下游说"这个项目有几个点可以写"。转第一次还行，转第三次就失真——同一组数据被写成两篇互相打架的稿；或者某条结论被前一篇在结尾顺手抖完，后一篇写到一半发现没得写，只能硬凑。

根因很清楚：交接物是"对话"，而对话不是状态，不会持久化、不可查询、无法校验。一轮会话结束，状态就蒸发了。

解法也很直接：把交接物从对话搬到盘上，变成一张表。上游只往表里写行，下游只从表里取行，两边都不许绕过这张表直接交流选题。这张表的列结构就是一台状态机，每一行选题就是状态机里的一个实例。

现在的实测数据：表里共48 行选题（公众号 24 / 百家号 24），状态分布为已发布 7、待 GEO 撰写 37、已写·待主理人审阅 1、已写·待明日发 1、暂不写 1、不写 1。换句话说，这台机器已经往前走了 7 步，队列里还排着 37 步，另外 4 步是被判定"不走"或"待放行"的。

本篇讲的就是这台机器本身：两个独立专家包，怎么用一张共享的状态机咬合起来。至于每个包内部怎么搭、主理人怎么调度成员，那是另外的选题，本篇不越界。

二、环境准备

2.1 盘上三个文件

三个文件都在同一个规划目录下（本文以D:\豆包工作\GEO规划\为例）。分工必须记牢：总表是权威选题池，台账是执行排期视图，两者不是一回事，也不能互相替代。

2.2 两个专家包的物理结构

两个包都是 Team 型，结构同构，但内容零耦合：

# 确认两个包已就位（本环境 Glob 偶发假阴性，用 ls 实测为准）ls"C:/Users/admin/.workbuddy/plugins/marketplaces/my-experts/plugins/"# geo-content-team stage-design-audit-team ...ls"C:/Users/admin/.workbuddy/plugins/marketplaces/my-experts/plugins/geo-content-team/"# .codebuddy-plugin agents avatars README.md

• .codebuddy-plugin/plugin.json：包清单，含version、expertType、leadAgent、memberAgents。
• agents/：每个成员一个 md，frontmatter 里写name / description / displayName / maxTurns。
• avatars/：头像资源。
• skills/：可复用的方法学（并非每个包都有）。

两个包都已完成校验与注册、处于生效状态。本文不涉及"怎么从零搭包"，那是独立选题。

2.3 运行环境

# 本文脚本仅用标准库，3.9+ 均可；本机实测解释器：C:/Users/admin/.workbuddy/binaries/python/versions/3.13.12/python.exe-c"import sys; print(sys.version)"# 3.13.x (...)

2.4 状态机的 9 列契约

总表表头必须是这 9 列，顺序固定，一个字都不能漂：

| 项目 | 专业 | 日期 | 平台 | 主题 | 状态 | 计划写作日期 | 是否写作 | 写作具体日期 |

每一列在状态机里承担的语义：

1. 项目：选题来源批次（某剧院项目 / 主线规划 / 复盘批次等），也是去重的作用域。
2. 专业：舞台工艺的专业归属（灯光、机械、音响、视频、文档自动化等）。
3. 日期：上游提炼日期。
4. 平台：公众号 / 百家号，决定下游派给哪个成员。
5. 主题：一行一个主题，括号里挂区分点与锚点编号。
6. 状态：状态机的当前态，下游唯一的准入判据。
7. 计划写作日期：排期，下游取号时的排序键。
8. 是否写作：总闸，只有"否"能被消费。
9. 写作具体日期：回填字段，未写填"—"。

这 9 列里，真正决定"能不能动"的是第 6 和第 8 列。下游的权限极小——只看"是否写作＝否"的行，其余一律不碰。

三、步骤：把流水线跑通的六个动作

步骤 1：上游产出契约（每个主题必挂锚点）

审图包在每次审图收尾时，执行一条产出纪律（包内编号 R25）：从本项目的审核实证里提炼7 个公众号主题（互不重复）+ 3–5 个 百家号 主题。

关键不是"提炼几个"，而是每个主题必须挂一枚"审核实证锚点"——可追溯到具体那份文件、那个编号、那组数字。锚点是整条链路的血液。没有锚点的主题，下游写出来必然是空话，因为它没有任何可引用的真实数字。

产出落成一个独立 md，文件名有固定格式：

D:\豆包工作\GEO规划\2026-09-25_某剧院_灯光_写作主题.md

文件内含：提炼日期 / 项目简称 / 专业 / 去重范围 / 脱密要求 / 下游说明 / 排除项 / 公众号主题表（含锚点与脱密提示）/ 百家号 主题表（含锚点）/ 跨项目去重登记。

步骤 2：登记总表（上游写行）

把主题逐行追加进总表，新行的初始状态固定为三件套：

• 状态 =待 GEO 撰写
• 是否写作 =否
• 写作具体日期 =—

登记前必须查重：同主题（含跨项目）已登记则跳过。这一动作后来被我做成了脚本（见第四节），因为人肉查重在第 30 行之后基本失效。

步骤 3：下游取号（只消费"是否写作＝否"）

下游主理人每天的唯一入口是：从总表里取"是否写作＝否"的行，按计划写作日期排序，取最靠前的一条。

这一步是防跑偏的总闸。下游严禁自创离散主题——你想写什么不重要，表里排到什么才写什么。这条规则看起来很死板，但它解决了两个真实问题：一是上游辛苦提炼的实证不会被浪费；二是下游永远不会写到"没素材、只能编"的题。

状态迁移的合法路径：

步骤 4：两道防污染闸门

闸门一：排除项（脱敏闸门）。上游在提炼时就先过滤掉两类题：核心方法不公开的、需脱密的个人职业叙事不列。同时项目名、图号一律泛化（写成"某剧院 / 业主方"），不挂编制单位。

这道闸门的价值在于："能写什么"在源头就被决定了，而不是写完再删。下游拿到的每一行，天生就是脱敏过的，照着写不会踩雷。

闸门二：跨项目去重。派稿前必须做两步：通读同项目内已发布/已写的同平台稿，逐条核对钩子、场景、关键数据与核心结论；再把已消耗的素材列成"禁用清单"，随派稿一并下发。

最硬的判据是这条：若同一组数据的核心结论已被前序稿讲完（哪怕只在结尾），新题就失去独立性——换题，不要试图改写救场。这不是"撞词"级别的重叠，是"结论级撞车"，改写法救不回来。

步骤 5：回填闭环

写完、发布后，必须同时回填两处：

• 总表：状态 →已发布，是否写作 →是，写作具体日期 → 实际写作日；
• 台账：补篇序 / 平台 / 主题 / 写作日期 / 发布日期 / 链接 / 状态。

不回填就是一次性工程。流水线的价值不在跑一次，而在于状态机能持续收敛：每跑一轮，"待撰写"少一行、"已发布"多一行，37 会变成 36、35、34……如果写完不回填，下一轮取号还会取到同一行，机器原地打转。

步骤 6：一次抽取、多平台复用

这是"一张总表养两个团队"的经济学来源。同一枚审核实证锚点，可以同时喂出两种稿：

• 技术稿：讲根因、复现、脚本与判据，面向开发者；
• 叙事稿：讲现场、决策、冲突与代价，面向行业读者。

真实事例：一次交付件格式异常（一批交付文档里绝大多数表线渲染成蓝色）这一枚锚点，同时产出了技术稿与叙事稿；一次分级判定（几十条意见的等级分布结论）同样一分为二。

注意"复用"不等于"复读"。同一锚点派生到不同平台时，必须按各平台调性重构——技术账号之间尤其不能字节级一致，那是另一条独立的硬闸。

四、代码：把状态机做成可执行脚本

下面这个脚本是整套机制的收口：解析总表、schema 断言、取号、去重、回填、双处一致性校验。全部只用标准库。

4.1 解析与 schema 断言

#!/usr/bin/env python# -*- coding: utf-8 -*-"""topic_pipeline.py —— 总表即状态机：取号 / 去重 / 回填 / 一致性校验"""importargparseimportreimportsysfromcollectionsimportCounterfrompathlibimportPath COLS=["项目","专业","日期","平台","主题","状态","计划写作日期","是否写作","写作具体日期"]NCOL=len(COLS)defsplit_row(line):"""把一行 Markdown 表格拆成单元格列表；非表格行返回 None"""line=line.strip()ifnotline.startswith("|"):returnNonereturn[c.strip()forcinline.strip("|").split("|")]defis_separator(cells):"""分隔行形如 | --- | --- |，只含 - / : / 空格"""returnset("".join(cells))<=set("- :")defparse_table(md):rows,seen_sep=[],Falseforlineno,rawinenumerate(md.read_text(encoding="utf-8").splitlines(),1):cells=split_row(raw)ifcellsisNone:continueifis_separator(cells):seen_sep=Truecontinueifnotseen_sep:ifcells[:NCOL]!=COLS:raiseValueError(f"表头漂移（第{lineno}行）：{cells}")continueiflen(cells)!=NCOL:raiseValueError(f"第{lineno}行列数={len(cells)}，应为{NCOL}")rows.append({"lineno":lineno,**dict(zip(COLS,cells))})ifnotrows:raiseValueError("未解析到任何数据行：检查表头或分隔行是否被破坏")returnrows

4.2 取号与去重

STOP={"的","了","和","与","是","在","把","用","从","到","一个","怎么","为什么","哪些","如何"}deftokens(text):"""中文粗分词：去括号注释 → 只留中英文数字 → 去停用词"""text=re.sub(r"[（(].*?[)）]","",text)text=re.sub(r"[^\u4e00-\u9fa5A-Za-z0-9]+"," ",text)return{tfortintext.split()iflen(t)>1andtnotinSTOP}defconsumed(rows):"""已被消耗的行：已写过的 + 已明确否决的"""return[rforrinrowsifr["是否写作"]=="是"orr["状态"]in("不写","暂不写")]defcollide(rows,topic,ratio=0.6):"""结论级撞车检测：只跟已消耗行比，避免与未来选题互撞"""a=tokens(topic)hits=[]forrinconsumed(rows):b=tokens(r["主题"])ifnotaornotb:continueov=len(a&b)/min(len(a),len(b))ifov>=ratio:hits.append((round(ov,2),r["平台"],r["状态"],r["主题"]))returnsorted(hits,reverse=True)defnext_topic(rows,platform=None):"""下游唯一入口：只消费 是否写作=否 且 状态=待 GEO 撰写 的行"""cand=[rforrinrowsifr["是否写作"]=="否"andr["状态"]=="待 GEO 撰写"]ifplatform:cand=[rforrincandifr["平台"]==platform]cand.sort(key=lambdar:(r["计划写作日期"]=="—",r["计划写作日期"],r["日期"]))returncand[0]ifcandelseNone

4.3 回填与双处一致性校验

defbackfill(md,row,status=None,date=None):"""回填：状态 / 是否写作 / 写作具体日期 一起改，禁止只改一处"""lines=md.read_text(encoding="utf-8").splitlines(keepends=True)idx=row["lineno"]-1cells=split_row(lines[idx])iflen(cells)!=NCOL:raiseValueError("目标行列数异常，拒绝写入")ifstatus:cells[COLS.index("状态")]=status cells[COLS.index("是否写作")]="是"ifdate:cells[COLS.index("写作具体日期")]=date lines[idx]="| "+" | ".join(cells)+" |\n"md.write_text("".join(lines),encoding="utf-8")print(f"[回填] 第{row['lineno']}行 ->{cells[COLS.index('主题')]}|{status}|{date}")defcross_check(rows,ledger):"""总表'已发布'条数 必须等于 台账已登记条数"""published=sum(1forrinrowsifr["状态"]=="已发布")ledger_n=sum(1forlninledger.read_text(encoding="utf-8").splitlines()ifln.strip().endswith("| 已发布 |"))ok=published==ledger_nprint(f"[校验] 总表已发布={published}台账已登记={ledger_n}"f"{'一致'ifokelse'★不一致，闭环断了'}")return0ifokelse1

4.4 命令行入口

defmain():ap=argparse.ArgumentParser()ap.add_argument("--table",required=True)ap.add_argument("--ledger")sub=ap.add_subparsers(dest="cmd",required=True)sub.add_parser("stats")p=sub.add_parser("next");p.add_argument("--platform")p=sub.add_parser("dedupe");p.add_argument("--topic",required=True)p=sub.add_parser("backfill");p.add_argument("--row",type=int,required=True)p.add_argument("--status");p.add_argument("--date")sub.add_parser("check")a=ap.parse_args()md=Path(a.table)rows=parse_table(md)ifa.cmd=="stats":fork,vinCounter(r["状态"]forrinrows).most_common():print(f"{k:<16}{v}")print("平台分布:",dict(Counter(r["平台"]forrinrows)))return0ifa.cmd=="next":t=next_topic(rows,a.platform)ifnott:print("无可消费选题（池已空）")return1print(f"[取号] 第{t['lineno']}行{t['平台']}|{t['主题']}")print(f" 计划写作日期={t['计划写作日期']}"f"锚点来源={t['项目']}/{t['专业']}")return0ifa.cmd=="dedupe":hits=collide(rows,a.topic)ifnothits:print("[去重] 未撞车，可登记")return0forov,plat,st,tpinhits:print(f"[撞车{ov}]{plat}/{st}/{tp}")print("[判定] 结论级撞车 → 换题，不要改写救场")return2ifa.cmd=="backfill":target=next((rforrinrowsifr["lineno"]==a.row),None)ifnottarget:print("行号不存在")return1backfill(md,target,a.status,a.date)return0ifa.cmd=="check":ifnota.ledger:print("需要 --ledger")return1returncross_check(rows,Path(a.ledger))if__name__=="__main__":sys.exit(main())

4.5 日常跑法

cd/d/豆包工作/GEO规划# 1) 看状态机当前分布python topic_pipeline.py--table_写作主题总表.md stats# 2) 下游取号（以 百家号 线为例）python topic_pipeline.py--table_写作主题总表.md next--platform百家号# 3) 派稿前去重（结论级撞车一票否决，退出码 2）python topic_pipeline.py--table_写作主题总表.md dedupe--topic"你的新主题"# 4) 写完回填三列python topic_pipeline.py--table_写作主题总表.md backfill\--row19--status已发布--date2026-09-30# 5) 两处一致性校验：退出码即结论python topic_pipeline.py--table_写作主题总表.md--ledger发布追踪台账.md checkecho$?

要点是第 5 步：脚本的退出码就是结论，不接受"口头声称已回填"。这一点是从另一条交付纪律里借来的思路——凡是可以机器判定的，就不要交给人去记。

4.6 脱敏扫描（闸门一的机械化）

# 登记前扫一遍主题列，命中即拒绝入库grep-nE"项目真名|设计院|有限公司|_图号"_写作主题总表.md

把真实项目名、编制单位、可反推身份的地名写进 grep 模式，返回为空才允许入库。上游的排除项清单同样可以用一个常量数组固化在脚本里，让"不列什么"变成可执行约束，而不是靠人记住。

五、踩坑实录

坑 1：状态停在对话里，下游凭记忆取题。
现象：同一枚锚点被写成两篇，口径还不一致。成因：交接物是对话，会话结束即蒸发，无任何可查询的状态。解法：状态必须显式落表，“写完"这件事的唯一证据是表里那一行的三列被改掉，不是谁说了句"写完了”。

坑 2：表头/列数漂移。
现象：手工追加行时少写了一列，9 列变成 8 列，下游按列索引取到的字段整体错位。成因：Markdown 表格没有任何 schema 约束，多一个少一个管道符照样能渲染。解法：解析时做两道断言——表头必须逐字匹配，数据行列数必须等于 9；写行一律走脚本，不手敲。

坑 3：无锚点的主题混进池子。
现象：下游拿到一行主题，写得洋洋洒洒却一个真实数字都没有，读者一眼看出是编的。成因：上游提炼时偷懒，锚点列留空。解法：登记前校验锚点非空；锚点为空的主题不予入库，退回上游补。

坑 4：结论级撞车，改写救场越改越糟。
现象：新题写到一半发现核心结论前一篇已经讲过，试图换个角度硬写，结果两篇互相削弱。成因：去重只比了标题关键词，没比"这组数据的结论有没有被消耗"。解法：硬判据——同一组数据的核心结论被前序稿消耗完，就换题；同时在派稿时下发"禁用素材清单"，命中即打回。

坑 5：脱敏在下游补，越补越漏。
现象：主题里带了真实项目名与图号，下游照抄进稿，发出去才发现。成因：把脱敏当成发布前的检查项，而不是上游的过滤项。解法：闸门一前移到上游提炼阶段——项目名/图号泛化、排除项不列，让下游拿到的每一行天生干净。

坑 6：回填只改一处。
现象：总表改成"已发布"，台账没登记；过几周一查，7 对 6，不知道漏的是哪篇。成因：两处文件都靠手改，人的注意力只够照顾一处。解法：一条命令同时改总表三列，并用check子命令拿退出码卡住——总表"已发布"条数必须等于台账已登记条数。

坑 7：一锚双投变成复读机。
现象：同一锚点派生到两个技术账号，两稿除标题外逐字相同，被判同质。成因：图省事直接复制。解法：把"两稿不得字节级一致、H1 必须不同"做成交付前的自审硬闸；哪怕有人口头说"照抄就行"，也要先确认再放行。

坑 8：表格宽了，发布端直接崩。
现象：9 列的总表结构被原样搬进文章，发布端渲染错乱，手机端必崩。成因：发布端编辑器对表格的容错远弱于本地渲染器。解法：文章里超过 3 列、或单元格超过 20 字的一律改写成列表形态；宽结构只在代码块或编号列表里出现。

坑 9：路径查不到，以为文件没了。
现象：脚本报"文件不存在"，但实际文件好端端在盘上。成因：本环境的 Glob 对既有文件偶发返回空（假阴性），工具类搜索结果不可全信。解法：路径落点先查记忆，再用 Read 或 Python 实际读一次验证，不拿搜索结果当结论。

六、总结

这套东西拆开看，没有任何一项是新发明：专家包是现成的多角色封装，总表是普通的 Markdown 表格，脚本不过两百行。真正让它变成"流水线"的，是三个约束：

第一，交接物必须是盘上的状态，不能是对话里的记忆。上游写行、下游取行，两边都不许绕过这张表。

第二，下游的权限必须被压缩到最小。只消费"是否写作＝否"的行，不自创主题。这条看起来最不自由，却是整条链路不跑偏的总闸。

第三，闭环必须机械化。回填状态、写作日期、链接，一个命令改三列，退出码即结论；总表与台账条数必须对齐。跑一次不叫流水线，状态机能持续收敛才叫流水线——队列从 37 走到 36、35、34，每一步都在表里留下可验证的痕迹。

做完这些，两个专家包依然是两个独立的包，各有各的边界，谁也没有变成谁的插件。它们之间唯一的连接，就是那张 9 列的表。这大概也是跨团编排最朴素的一层道理：系统之间不需要深度融合，只需要一个双方都认、且能被执行校验的接口。