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

舞台工艺 × 多智能体用两张专家团和一张状态机总表把「审图→选题→写作」跑成自动流水线一、背景两个团各自能跑但连不上我在本地自建了两个彼此独立的专家包一个负责舞台工艺图纸审核一个负责内容生产。它们的目录完全隔离互不 import、互不 spawn连版本号都是各走各的——审图包当前是v1.3.0内容包是v1.1.0。设计上这两个包是刻意解耦的上游审图包接图纸、出审核意见、出交付件。审完一个项目脑子里会留下大量真实素材——某个参数在几张图上不一致、某批交付件格式全线翻车、某次分级判定的边界争议。这些素材极有价值但这个包有一条硬边界只提炼、不写作。下游内容包每天要出稿最缺的不是文笔而是有实证、可追溯、已脱敏的选题。它同样有硬边界不审图、不自创离散主题。问题就出在交界处。最早我靠对话口头转述审完一个项目在会话里跟下游说这个项目有几个点可以写。转第一次还行转第三次就失真——同一组数据被写成两篇互相打架的稿或者某条结论被前一篇在结尾顺手抖完后一篇写到一半发现没得写只能硬凑。根因很清楚交接物是对话而对话不是状态不会持久化、不可查询、无法校验。一轮会话结束状态就蒸发了。解法也很直接把交接物从对话搬到盘上变成一张表。上游只往表里写行下游只从表里取行两边都不许绕过这张表直接交流选题。这张表的列结构就是一台状态机每一行选题就是状态机里的一个实例。现在的实测数据表里共48 行选题公众号 24 / 百家号 24状态分布为已发布 7、待 GEO 撰写 37、已写·待主理人审阅 1、已写·待明日发 1、暂不写 1、不写 1。换句话说这台机器已经往前走了 7 步队列里还排着 37 步另外 4 步是被判定不走或待放行的。本篇讲的就是这台机器本身两个独立专家包怎么用一张共享的状态机咬合起来。至于每个包内部怎么搭、主理人怎么调度成员那是另外的选题本篇不越界。二、环境准备2.1 盘上三个文件文件角色谁写 / 谁读_写作主题总表.md状态机本体上游写行 · 下游读行发布追踪台账.md执行侧六段下游登记 · 回填校验日期_项目_专业_写作主题.md单项目交接物上游产出 · 下游取锚点三个文件都在同一个规划目录下本文以D:\豆包工作\GEO规划\为例。分工必须记牢总表是权威选题池台账是执行排期视图两者不是一回事也不能互相替代。2.2 两个专家包的物理结构两个包都是 Team 型结构同构但内容零耦合# 确认两个包已就位本环境 Glob 偶发假阴性用 ls 实测为准lsC:/Users/admin/.workbuddy/plugins/marketplaces/my-experts/plugins/# geo-content-team stage-design-audit-team ...lsC:/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/每个成员一个 mdfrontmatter 里写name / description / displayName / maxTurns。avatars/头像资源。skills/可复用的方法学并非每个包都有。两个包都已完成校验与注册、处于生效状态。本文不涉及怎么从零搭包那是独立选题。2.3 运行环境# 本文脚本仅用标准库3.9 均可本机实测解释器C:/Users/admin/.workbuddy/binaries/python/versions/3.13.12/python.exe-cimport sys; print(sys.version)# 3.13.x (...)2.4 状态机的 9 列契约总表表头必须是这 9 列顺序固定一个字都不能漂| 项目 | 专业 | 日期 | 平台 | 主题 | 状态 | 计划写作日期 | 是否写作 | 写作具体日期 |每一列在状态机里承担的语义项目选题来源批次某剧院项目 / 主线规划 / 复盘批次等也是去重的作用域。专业舞台工艺的专业归属灯光、机械、音响、视频、文档自动化等。日期上游提炼日期。平台公众号 / 百家号决定下游派给哪个成员。主题一行一个主题括号里挂区分点与锚点编号。状态状态机的当前态下游唯一的准入判据。计划写作日期排期下游取号时的排序键。是否写作总闸只有否能被消费。写作具体日期回填字段未写填—。这 9 列里真正决定能不能动的是第 6 和第 8 列。下游的权限极小——只看是否写作否的行其余一律不碰。三、步骤把流水线跑通的六个动作步骤 1上游产出契约每个主题必挂锚点审图包在每次审图收尾时执行一条产出纪律包内编号 R25从本项目的审核实证里提炼7 个公众号主题互不重复 3–5 个 百家号 主题。关键不是提炼几个而是每个主题必须挂一枚审核实证锚点——可追溯到具体那份文件、那个编号、那组数字。锚点是整条链路的血液。没有锚点的主题下游写出来必然是空话因为它没有任何可引用的真实数字。产出落成一个独立 md文件名有固定格式D:\豆包工作\GEO规划\2026-09-25_某剧院_灯光_写作主题.md文件内含提炼日期 / 项目简称 / 专业 / 去重范围 / 脱密要求 / 下游说明 / 排除项 / 公众号主题表含锚点与脱密提示/ 百家号 主题表含锚点/ 跨项目去重登记。步骤 2登记总表上游写行把主题逐行追加进总表新行的初始状态固定为三件套状态 待 GEO 撰写是否写作 否写作具体日期 —登记前必须查重同主题含跨项目已登记则跳过。这一动作后来被我做成了脚本见第四节因为人肉查重在第 30 行之后基本失效。步骤 3下游取号只消费是否写作否下游主理人每天的唯一入口是从总表里取是否写作否的行按计划写作日期排序取最靠前的一条。这一步是防跑偏的总闸。下游严禁自创离散主题——你想写什么不重要表里排到什么才写什么。这条规则看起来很死板但它解决了两个真实问题一是上游辛苦提炼的实证不会被浪费二是下游永远不会写到没素材、只能编的题。状态迁移的合法路径状态取值含义是否可消费待 GEO 撰写排队中是唯一入口已写·待主理人审阅已产出待放行否已写·待明日发已放行待发否已发布已回填链接否暂不写 / 不写已否决否步骤 4两道防污染闸门闸门一排除项脱敏闸门。上游在提炼时就先过滤掉两类题核心方法不公开的、需脱密的个人职业叙事不列。同时项目名、图号一律泛化写成某剧院 / 业主方不挂编制单位。这道闸门的价值在于能写什么在源头就被决定了而不是写完再删。下游拿到的每一行天生就是脱敏过的照着写不会踩雷。闸门二跨项目去重。派稿前必须做两步通读同项目内已发布/已写的同平台稿逐条核对钩子、场景、关键数据与核心结论再把已消耗的素材列成禁用清单随派稿一并下发。最硬的判据是这条若同一组数据的核心结论已被前序稿讲完哪怕只在结尾新题就失去独立性——换题不要试图改写救场。这不是撞词级别的重叠是结论级撞车改写法救不回来。步骤 5回填闭环写完、发布后必须同时回填两处总表状态 →已发布是否写作 →是写作具体日期 → 实际写作日台账补篇序 / 平台 / 主题 / 写作日期 / 发布日期 / 链接 / 状态。不回填就是一次性工程。流水线的价值不在跑一次而在于状态机能持续收敛每跑一轮待撰写少一行、已发布多一行37 会变成 36、35、34……如果写完不回填下一轮取号还会取到同一行机器原地打转。步骤 6一次抽取、多平台复用这是一张总表养两个团队的经济学来源。同一枚审核实证锚点可以同时喂出两种稿技术稿讲根因、复现、脚本与判据面向开发者叙事稿讲现场、决策、冲突与代价面向行业读者。真实事例一次交付件格式异常一批交付文档里绝大多数表线渲染成蓝色这一枚锚点同时产出了技术稿与叙事稿一次分级判定几十条意见的等级分布结论同样一分为二。注意复用不等于复读。同一锚点派生到不同平台时必须按各平台调性重构——技术账号之间尤其不能字节级一致那是另一条独立的硬闸。四、代码把状态机做成可执行脚本下面这个脚本是整套机制的收口解析总表、schema 断言、取号、去重、回填、双处一致性校验。全部只用标准库。4.1 解析与 schema 断言#!/usr/bin/env python# -*- coding: utf-8 -*-topic_pipeline.py —— 总表即状态机取号 / 去重 / 回填 / 一致性校验importargparseimportreimportsysfromcollectionsimportCounterfrompathlibimportPath COLS[项目,专业,日期,平台,主题,状态,计划写作日期,是否写作,写作具体日期]NCOLlen(COLS)defsplit_row(line):把一行 Markdown 表格拆成单元格列表非表格行返回 Nonelineline.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(encodingutf-8).splitlines(),1):cellssplit_row(raw)ifcellsisNone:continueifis_separator(cells):seen_sepTruecontinueifnotseen_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(未解析到任何数据行检查表头或分隔行是否被破坏)returnrows4.2 取号与去重STOP{的,了,和,与,是,在,把,用,从,到,一个,怎么,为什么,哪些,如何}deftokens(text):中文粗分词去括号注释 → 只留中英文数字 → 去停用词textre.sub(r[(].*?[)],,text)textre.sub(r[^\u4e00-\u9fa5A-Za-z0-9], ,text)return{tfortintext.split()iflen(t)1andtnotinSTOP}defconsumed(rows):已被消耗的行已写过的 已明确否决的return[rforrinrowsifr[是否写作]是orr[状态]in(不写,暂不写)]defcollide(rows,topic,ratio0.6):结论级撞车检测只跟已消耗行比避免与未来选题互撞atokens(topic)hits[]forrinconsumed(rows):btokens(r[主题])ifnotaornotb:continueovlen(ab)/min(len(a),len(b))ifovratio:hits.append((round(ov,2),r[平台],r[状态],r[主题]))returnsorted(hits,reverseTrue)defnext_topic(rows,platformNone):下游唯一入口只消费 是否写作否 且 状态待 GEO 撰写 的行cand[rforrinrowsifr[是否写作]否andr[状态]待 GEO 撰写]ifplatform:cand[rforrincandifr[平台]platform]cand.sort(keylambdar:(r[计划写作日期]—,r[计划写作日期],r[日期]))returncand[0]ifcandelseNone4.3 回填与双处一致性校验defbackfill(md,row,statusNone,dateNone):回填状态 / 是否写作 / 写作具体日期 一起改禁止只改一处linesmd.read_text(encodingutf-8).splitlines(keependsTrue)idxrow[lineno]-1cellssplit_row(lines[idx])iflen(cells)!NCOL:raiseValueError(目标行列数异常拒绝写入)ifstatus:cells[COLS.index(状态)]status cells[COLS.index(是否写作)]是ifdate:cells[COLS.index(写作具体日期)]date lines[idx]| | .join(cells) |\nmd.write_text(.join(lines),encodingutf-8)print(f[回填] 第{row[lineno]}行 -{cells[COLS.index(主题)]}|{status}|{date})defcross_check(rows,ledger):总表已发布条数 必须等于 台账已登记条数publishedsum(1forrinrowsifr[状态]已发布)ledger_nsum(1forlninledger.read_text(encodingutf-8).splitlines()ifln.strip().endswith(| 已发布 |))okpublishedledger_nprint(f[校验] 总表已发布{published}台账已登记{ledger_n}f{一致ifokelse★不一致闭环断了})return0ifokelse14.4 命令行入口defmain():apargparse.ArgumentParser()ap.add_argument(--table,requiredTrue)ap.add_argument(--ledger)subap.add_subparsers(destcmd,requiredTrue)sub.add_parser(stats)psub.add_parser(next);p.add_argument(--platform)psub.add_parser(dedupe);p.add_argument(--topic,requiredTrue)psub.add_parser(backfill);p.add_argument(--row,typeint,requiredTrue)p.add_argument(--status);p.add_argument(--date)sub.add_parser(check)aap.parse_args()mdPath(a.table)rowsparse_table(md)ifa.cmdstats:fork,vinCounter(r[状态]forrinrows).most_common():print(f{k:16}{v})print(平台分布:,dict(Counter(r[平台]forrinrows)))return0ifa.cmdnext:tnext_topic(rows,a.platform)ifnott:print(无可消费选题池已空)return1print(f[取号] 第{t[lineno]}行{t[平台]}|{t[主题]})print(f 计划写作日期{t[计划写作日期]}f锚点来源{t[项目]}/{t[专业]})return0ifa.cmddedupe:hitscollide(rows,a.topic)ifnothits:print([去重] 未撞车可登记)return0forov,plat,st,tpinhits:print(f[撞车{ov}]{plat}/{st}/{tp})print([判定] 结论级撞车 → 换题不要改写救场)return2ifa.cmdbackfill:targetnext((rforrinrowsifr[lineno]a.row),None)ifnottarget:print(行号不存在)return1backfill(md,target,a.status,a.date)return0ifa.cmdcheck: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) 派稿前去重结论级撞车一票否决退出码 2python 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 列的表。这大概也是跨团编排最朴素的一层道理系统之间不需要深度融合只需要一个双方都认、且能被执行校验的接口。