舞台工艺交付物的 AI 智能体分工出稿：合并稿按唯一权威源重建，以及三道质量闸门的判据与退出码

一、背景：分头交稿之后，拼接留下的不是重复，而是两套口径

某剧院舞台工艺交付物批次里，一批 Word 与 md 文档由多个 AI 智能体分头产出，最后要收敛成一份可交付的权威文档。最先暴露问题的不是内容深度，而是编号：把几份分稿首尾相接之后，编号出现断档，同一节还并排出现两种叫法，标题层级也忽上忽下。这类现象常被当成排版小毛病，实际上它说明整份文档已经没有统一的骨架——谁来定义章节顺序、谁来定义层级、谁来定义编号，没人说得清。而这一批要收敛的不止主稿，还有若干派生件——派生件，就是从主稿复制、改写后投向不同渠道的那一份副本文件；本文反复说的唯一权威源，指全批次只认一份主稿，由它定义章节顺序、层级、编号，其余分稿无权修改。

多个 AI 智能体分头写稿之后，合并不是把文本拼起来，而是从唯一权威源重建整份文档；派生件没同步、脱敏没扫净、docx 没复扫到 XML 层，就等于这份交付物还有第二个权威版本。

这句话不涉及任何审图方法、定级或参数阈值，只讲文档工程化。本篇把视角收在一条出稿流水线上：从多源分稿归集开始，走过三道质量闸门，最后落到回读比对。零散的编码与命令类坑已在别文写过，这里不重复，只谈判据怎么定、退出码怎么分、一份文档凭什么算收敛完成。也就是说，本文写的是「一条出稿流水线上的三道质量闸门」，不是一份散点式的踩坑清单。

先把四类真实事故定性，后面每一类都会变成可运行代码。

第一类是合并稿重建。多个智能体各交一份分稿，直接拼接会出现同一节两套口径、标题层级错乱、编号断档；正确做法是先定唯一权威源——全批次只认一份主稿，它的章节顺序、层级、编号是唯一标准，分稿无权修改——再把其余分稿降级为素材块。素材块，指只保留正文、不带标题层级和编号的一段可复用内容；按章节挂回主稿骨架之后，还要逐节回读比对其来源与版本。

第二类是派生件同步。源文改了标题或结论，渠道稿里有一份没跟着改，渠道元数据文件仍挂旧标题，读者就会同时看到两种表述。真实教训是某份渠道元数据文件的标题字段还留着上一版标题，源文已经改过好几轮，它却没动。

第三类是脱敏扫描。md 里清掉的姓名、单位名、地名，在 docx 里可能因为 run 拆分而残留。run 拆分，指 Word 把一个词切成好几段带格式的小片段分开存，屏幕上看着是一整句，文件里却是几段；「XX设计院」被切成「XX」和「设计院」两个片段，简单字符串匹配就扫不到。必须按段落重组全文再扫，而且 md 与 docx 两份都要扫。

第四类是 docx 全量复扫。重生成之后，除 word/document.xml 外还有页眉、页脚、脚注、尾注、批注等 XML 部件。XML 部件，指 .docx 这个压缩包里一个个独立的 .xml 文件，正文、页眉、页脚、脚注、批注各占一份；只查其中一份必然漏。异常字符包括 U+FFFD 替换字符、私用区 U+E000–U+F8FF、零宽字符 U+200B–U+200D、BOM U+FEFF，以及中文经错误编码形成的 mojibake 串。Unicode 码位，就是每个字符在全球字符表里的唯一编号，习惯写成 U+ 加十六进制数字；检测要遍历压缩包内全部 .xml 部件，按 Unicode 类别分类计数。

二、环境准备：版本、依赖、构造样本与唯一权威源的声明方式

实测环境是 Python 3.13.14，代码按 3.10 及以上写，只用标准库：re、zipfile、unicodedata、io、sys、ast。不装 python-docx，也不依赖任何外部文件——所有样本都在脚本里用字符串和内存字节流现场构造，读者把代码块整段复制到一个 .py 文件里就能跑通。这一点很重要：文档工程化的检查脚本如果自己还要先去某个目录里找素材，它本身就会变成第二个不确定因素。

样本分三层。第一层是主稿骨架，用一个列表声明，每个元素只放章节 id、编号、层级、标题、版本五个字段；第二层是分稿素材块，带正文、来源、版本，但故意不带骨架权限；第三层是交付物实体，用 zipfile 在内存里拼出一个最小可用的 .docx 压缩包，里面塞进正文、页眉、页脚、脚注、批注几个部件。构造样本的好处是每一处异常都可控：哪个词被拆开、哪个部件藏了异常字符，都是刻意放进去的，因此跑出来的数字可复现。

素材里还会原样保留 DWG、齐套性、提资、吊杆、调光柜、DMX512、JGJ 57、WH/T 这类同行词，以及 GB/T、WH/T、JGJ 57-2016 这类规范标注。重建时按原样搬运，一个字都不改：它们是交付物的元信息，不参与任何章节结构的裁定。本文也不展开这些词背后的专业内容，只关心它们在文件里有没有被搬运丢、被改写、被重复写入。

被检查的对象固定为三类：md 源文、docx 交付件、渠道元数据文件。三者必须进同一张对照清单——这正是后面第二道闸门要守的边界。唯一权威源的声明方式同样是硬约定：骨架只有一个来源，分稿只提供素材块，任何分稿自带的标题、层级、编号一律丢弃。丢弃不是浪费，而是让「谁来定义结构」这个问题只有一个答案。

三、步骤：定权威源、按章节回填、派生件对照、闸门串联、回读比对

主流程分五步，顺序不能换。

  1. 归集多源分稿：把每个 AI 智能体交来的文件登记成「分稿 id＋来源＋版本＋章节 id」，正文先不解析。归集阶段只做清点，不做合并，避免一边读一边改。
  2. 声明唯一权威源：全批次只认一份主稿，由它给出章节顺序、层级、编号、标题与版本。骨架一旦声明，本轮重建期间不再变动。
  3. 素材块级按章节重建：把分稿拆成只含正文的素材块，按章节 id 挂到主稿骨架下；同节出现多个来源时按版本择一，并把被选中的来源与版本写进该节的留痕字段。骨架字段一律取主稿，正文一律取素材块，两者不交叉。
  4. 派生件同步矩阵：源文的标题、结论、版本三个字段，逐份与派生件做字段级对照；派生件是一等公民，渠道元数据文件同样进矩阵，不因为它「只是个配置」就跳过。
  5. 三道质量闸门串联，再回读比对：按顺序短路执行三道判据，全部放行之后逐节回读，核对标题、层级、编号以及每节素材块的来源与版本，输出一份可追溯的清单。

判据与退出码按闸门分段设计，0 只表示本轮三道判据都过，不代表文档内容正确：

  • 重建与回读：判据是每节的章节 id、层级、编号、标题、版本与权威源逐字段相等，且每节素材来源都有留痕；任一字段不等 → 退出码 2。
  • 派生件同步：判据是「派生件份数 × 字段数」的每一格都等于源文对应字段；任意一格不等 → 退出码 3。
  • 脱敏与复扫：判据是 md 与 docx 两份的脱敏命中数都为 0，且压缩包内全部 .xml 部件的异常字符计数为 0；任一不为 0 → 退出码 4。
  • 全部放行：退出码 0，流水线才允许把这一版标成可交付。

这套设计的要点是把「判据」和「结论」分开。判据是可计算的布尔值，结论是退出码；AI 智能体检查负责跑判据，人只看退出码和留痕清单。把三道质量闸门串成一条链而不是三个独立按钮，是为了让失败停在最靠前的那一环：骨架没收敛好，就没必要去扫派生件；派生件还在挂旧标题，就没必要去谈脱敏。回读比对放在最后，是因为它回答的是另一个问题——重建出来的这份，每一节到底来自谁、哪一版。

四、代码与脚本实测：五段可直接运行的 Python 检查

下面所有数字都来自可运行脚本对构造样本的实测输出，统一标注为「脚本实测」。它们不是任何批次的审核统计，也不对应任何真实项目的数量；换句话说，文中出现的个数、格数、命中数，全部是脚本跑出来的，不是估出来的。

第一段把「直接拼接」和「按章节重建」放在同一个样本上对照，并做逐节回读。

# -*- coding: utf-8 -*-
"""按章节重建 + 回读比对；对照“直接拼接”会留下的问题。"""
from __future__ import annotations
import sys

# 唯一权威源：主稿只定义章节骨架（顺序 / 层级 / 编号 / 标题）
MASTER = [
    {"sid": "S1", "num": 1, "level": 1, "title": "交付范围", "version": "v3"},
    {"sid": "S2", "num": 2, "level": 1, "title": "环境与依赖", "version": "v3"},
    {"sid": "S3", "num": 3, "level": 1, "title": "重建流程", "version": "v3"},
    {"sid": "S4", "num": 4, "level": 1, "title": "质量判据", "version": "v3"},
    {"sid": "S5", "num": 5, "level": 1, "title": "回读比对", "version": "v3"},
]

# 分稿降级为素材块：只带正文与来源版本，不带骨架权限
FRAGMENTS = [
    {"sid": "S1", "title": "交付范围", "level": 1, "num": 1,
     "body": "本批次含 Word 与 md 两类交付物。", "source": "分稿A", "version": "v1"},
    {"sid": "S2", "title": "环境与依赖", "level": 2, "num": 7,
     "body": "Python 3.10 及以上，仅用标准库。", "source": "分稿A", "version": "v1"},
    {"sid": "S3", "title": "重建流程", "level": 1, "num": 3,
     "body": "先定权威源，再按章节回填素材块。", "source": "分稿B", "version": "v2"},
    {"sid": "S3", "title": "合并流程（旧口径）", "level": 1, "num": 3,
     "body": "把各分稿首尾相接即可。", "source": "分稿C", "version": "v1"},
    {"sid": "S4", "title": "质量判据", "level": 1, "num": 4,
     "body": "三份文件都要过扫描。", "source": "分稿B", "version": "v2"},
    {"sid": "S5", "title": "回读比对", "level": 1, "num": 9,
     "body": "逐节核对标题、层级与编号。", "source": "分稿C", "version": "v1"},
    {"sid": "S6", "title": "附：旧版附录", "level": 1, "num": 6,
     "body": "此节已废弃。", "source": "分稿C", "version": "v1"},
]


def audit_concat(blocks):
    """朴素拼接后，统计口径冲突、层级错乱、编号断档、越界节。"""
    titles = {}
    conflict = 0
    for b in blocks:
        if b["sid"] in titles and titles[b["sid"]] != b["title"]:
            conflict += 1
        titles[b["sid"]] = b["title"]
    level_jump = sum(1 for b in blocks if b["level"] != 1)
    nums = [b["num"] for b in blocks]
    gaps = sum(1 for a, c in zip(nums, nums[1:]) if c != a + 1)
    known = {m["sid"] for m in MASTER}
    out_of_range = sum(1 for b in blocks if b["sid"] not in known)
    return conflict, level_jump, gaps, out_of_range


def rebuild(master, fragments):
    """骨架取权威源，正文取素材块；同节多来源按版本号择一并记录冲突。"""
    by_sid = {}
    for f in fragments:
        by_sid.setdefault(f["sid"], []).append(f)
    rebuilt, conflicts = [], 0
    for m in master:
        candidates = by_sid.get(m["sid"], [])
        chosen = max(candidates, key=lambda x: x["version"]) if candidates else None
        if len(candidates) > 1:
            conflicts += 1
        rebuilt.append({
            "sid": m["sid"], "num": m["num"], "level": m["level"],
            "title": m["title"], "version": m["version"],
            "body": chosen["body"] if chosen else "",
            "source": chosen["source"] if chosen else "主稿",
            "src_version": chosen["version"] if chosen else m["version"],
        })
    return rebuilt, conflicts


def readback(rebuilt, master):
    """回读比对：标题 / 层级 / 编号逐节对权威源，另列每个素材块来源版本。"""
    problems = []
    for got, want in zip(rebuilt, master):
        for field in ("sid", "num", "level", "title", "version"):
            if got[field] != want[field]:
                problems.append((want["sid"], field))
    return problems


def main():
    c, l, g, o = audit_concat(FRAGMENTS)
    print(f"[拼接] 同节两套口径 {c} 处；层级错乱 {l} 处；编号断档 {g} 处；越界节 {o} 个")
    rebuilt, conflicts = rebuild(MASTER, FRAGMENTS)
    problems = readback(rebuilt, MASTER)
    print(f"[重建] 权威源骨架 {len(MASTER)} 节；素材块 {len(FRAGMENTS)} 块；同节多来源 {conflicts} 节（按版本择一）")
    print(f"[回读] 逐节对照不一致 {len(problems)} 项；编号连续 1..{len(rebuilt)}")
    for s in rebuilt:
        print(f"  {s['sid']} L{s['level']} {s['title']} <- {s['source']}@{s['src_version']}")
    code = 0 if not problems else 2
    print(f"EXIT={code}")
    sys.exit(code)


if __name__ == "__main__":
    main()

脚本实测（python rebuild_check.py，构造样本：一份主稿骨架＋若干分稿素材块）：

  • [拼接] 同节两套口径 1 处；层级错乱 1 处；编号断档 5 处；越界节 1 个
  • [重建] 权威源骨架 5 节；素材块 7 块；同节多来源 1 节（按版本择一）
  • [回读] 逐节对照不一致 0 项；编号连续 1..5
  • S1 L1 交付范围 <- 分稿A@v1，S3 L1 重建流程 <- 分稿B@v2，S5 L1 回读比对 <- 分稿C@v1
  • EXIT=0

拼接侧四处异常，重建侧逐节不一致 0 项，同节多来源按版本择一后留痕完整。注意 rebuild() 里骨架字段全部取自 master，分稿的 title、level、num 从头到尾没被读进结果——这正是「重建」与「拼接」在实现上的分水岭。

第二段做派生件同步矩阵，把源文与每份派生件拆到字段级对照。

# -*- coding: utf-8 -*-
"""派生件同步矩阵：源文与每份派生件做字段级对照。"""
from __future__ import annotations
import sys

SOURCE = {
    "title": "合并稿重建：从唯一权威源重建整份文档",
    "conclusion": "重建而非拼接",
    "version": "v3",
    "channel": "主稿",
}
DERIVED = [
    {"channel": "渠道甲", "title": SOURCE["title"], "conclusion": SOURCE["conclusion"], "version": SOURCE["version"]},
    {"channel": "渠道乙", "title": SOURCE["title"], "conclusion": SOURCE["conclusion"], "version": SOURCE["version"]},
    {"channel": "渠道丙", "title": "合并稿重建（旧标题）", "conclusion": SOURCE["conclusion"], "version": SOURCE["version"]},
    {"channel": "渠道丁", "title": SOURCE["title"], "conclusion": SOURCE["conclusion"], "version": "v2"},
    {"channel": "渠道戊", "title": SOURCE["title"], "conclusion": SOURCE["conclusion"], "version": SOURCE["version"]},
]
FIELDS = ("title", "conclusion", "version")


def check(source, derived, fields=FIELDS):
    total, bad = 0, []
    for d in derived:
        for f in fields:
            total += 1
            if d[f] != source[f]:
                bad.append((d["channel"], f, d[f]))
    return total, bad


def main():
    total, bad = check(SOURCE, DERIVED)
    print(f"[同步矩阵] 派生件 {len(DERIVED)} 份 × 字段 {len(FIELDS)} 个 = 对照 {total} 格")
    for ch, f, val in bad:
        print(f"  不一致: {ch} / {f} = {val}")
    print(f"[同步矩阵] 不一致 {len(bad)} 格；源文标题 = {SOURCE['title']}")
    code = 0 if not bad else 3
    print(f"EXIT={code}")
    sys.exit(code)


if __name__ == "__main__":
    main()

脚本实测（python sync_matrix.py，构造样本：一份源文＋若干渠道派生件）：

  • [同步矩阵] 派生件 5 份 × 字段 3 个 = 对照 15 格
  • 不一致: 渠道丙 / title = 合并稿重建（旧标题）
  • 不一致: 渠道丁 / version = v2
  • [同步矩阵] 不一致 2 格；源文标题 = 合并稿重建：从唯一权威源重建整份文档
  • EXIT=3

矩阵的价值在于把「好像都改了」变成可数的格数：15 格里 2 格不等，一处是旧标题，一处是落后版本。两者都不在正文里，光读正文永远发现不了。

第三段处理脱敏扫描，md 与 docx 两份都扫，docx 按段落重组后再匹配。

# -*- coding: utf-8 -*-
"""脱敏扫描：md 与 docx 两份都扫；docx 必须按段落重组 run 再匹配。"""
from __future__ import annotations
import io
import re
import sys
import zipfile

SENSITIVE = ("XX设计院", "张某", "某地")
MD_TEXT = "本批次交付物由多个 AI 智能体分头产出，合并后统一归档。"
SPLIT_RUNS = [
    "<w:p><w:r><w:t>编制单位：</w:t></w:r>"
    "<w:r><w:t>XX</w:t></w:r>"
    "<w:r><w:t>设计院</w:t></w:r></w:p>",
    "<w:p><w:r><w:t>校核人：张某；地点：某地</w:t></w:r></w:p>",
]
DOCUMENT_XML = (
    '<?xml version="1.0" encoding="UTF-8"?>'
    '<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main">'
    "<w:body>" + "".join(SPLIT_RUNS) + "</w:body></w:document>"
)


def make_docx():
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as z:
        z.writestr("word/document.xml", DOCUMENT_XML.encode("utf-8"))
    return buf.getvalue()


def paragraph_text(xml_text):
    """把一个 <w:p> 内所有 <w:t> 的文本接起来，抵消 run 拆分。"""
    out = []
    for p in re.findall(r"<w:p[ >].*?</w:p>", xml_text, re.S):
        out.append("".join(re.findall(r"<w:t[^>]*>(.*?)</w:t>", p, re.S)))
    return "\n".join(out)


def hits(text):
    return [(w, text.count(w)) for w in SENSITIVE if w in text]


def main():
    raw = make_docx()
    with zipfile.ZipFile(io.BytesIO(raw)) as z:
        xml_text = z.read("word/document.xml").decode("utf-8")
    naive = hits(xml_text)
    rebuilt = hits(paragraph_text(xml_text))
    md = hits(MD_TEXT)
    print(f"[脱敏] md 命中 {len(md)} 处；docx 朴素串匹配命中 {len(naive)} 处：{naive}")
    print(f"[脱敏] docx 段落重组后命中 {len(rebuilt)} 处：{rebuilt}")
    print("[脱敏] 结论：只扫 md 会放行，docx 必须重组 run 后再扫")
    code = 4 if rebuilt else 0
    print(f"EXIT={code}")
    sys.exit(code)


if __name__ == "__main__":
    main()

脚本实测（python desens_scan.py，构造样本：一份已清理的 md＋一份内存构造的 docx）：

  • [脱敏] md 命中 0 处；docx 朴素串匹配命中 2 处：[('张某', 1), ('某地', 1)]
  • [脱敏] docx 段落重组后命中 3 处：[('XX设计院', 1), ('张某', 1), ('某地', 1)]
  • [脱敏] 结论：只扫 md 会放行，docx 必须重组 run 后再扫
  • EXIT=4

朴素串匹配漏掉的正是被拆开的那一个：XX 与 设计院 分属两个 <w:r>，中间的标签把词切断了。paragraph_text() 先把一个 <w:p> 内所有 <w:t> 的文本接起来，再交给匹配函数，这一段是整个扫描里最不能省的代码。

第四段做全量 XML 复扫，遍历压缩包内全部 .xml 部件，按 Unicode 类别分类计数。

# -*- coding: utf-8 -*-
"""全量 XML 复扫：遍历 zip 内所有 .xml 部件，按 Unicode 类别分类计数。"""
from __future__ import annotations
import io
import re
import sys
import unicodedata
import zipfile

W_NS = 'xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"'


def part(tag, text):
    return ('<?xml version="1.0" encoding="UTF-8"?>'
            f'<w:{tag} {W_NS}>{text}</w:{tag}>')


def make_docx():
    mojibake = "中文".encode("utf-8").decode("cp1252")  # 构造乱码串
    bodies = {
        "word/document.xml": ("document", "正文：舞台工艺交付物 \ufffd 复核完成。"),
        "word/header1.xml": ("hdr", "页眉：\ue000 内部编号。"),
        "word/footer1.xml": ("ftr", "页脚：\u200b\u200c\u200d 零宽。"),
        "word/footnotes.xml": ("footnotes", f"脚注：{mojibake} 误编码。"),
        "word/comments.xml": ("comments", "批注：\ufeff 待确认。"),
    }
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as z:
        for name, (tag, text) in bodies.items():
            z.writestr(name, part(tag, text).encode("utf-8"))
        z.writestr("word/settings.xml", "<w:settings/>".encode("utf-8"))
    return buf.getvalue()


MOJIBAKE = re.compile(r"[\u0080-\u00ff\u0100-\u017f\u2000-\u206f]{4,}")


def mojibake_fix(s):
    for codec in ("cp1252", "latin-1"):
        try:
            fixed = s.encode(codec).decode("utf-8")
        except (UnicodeEncodeError, UnicodeDecodeError):
            continue
        if any("\u4e00" <= c <= "\u9fff" for c in fixed):
            return fixed
    return None


def classify(text):
    counts = {"U+FFFD 替换字符": 0, "私用区 Co": 0, "零宽/格式 Cf": 0, "mojibake 串": 0}
    for ch in text:
        if ord(ch) == 0xFFFD:
            counts["U+FFFD 替换字符"] += 1
        elif unicodedata.category(ch) == "Co":
            counts["私用区 Co"] += 1
        elif unicodedata.category(ch) == "Cf":
            counts["零宽/格式 Cf"] += 1
    for m in MOJIBAKE.findall(text):
        if mojibake_fix(m):
            counts["mojibake 串"] += 1
    return counts


def main():
    raw = make_docx()
    total, per_part = {}, {}
    with zipfile.ZipFile(io.BytesIO(raw)) as z:
        names = [n for n in z.namelist() if n.endswith(".xml")]
        for n in names:
            c = classify(z.read(n).decode("utf-8"))
            per_part[n] = sum(c.values())
            for k, v in c.items():
                total[k] = total.get(k, 0) + v
    only_doc = per_part.get("word/document.xml", 0)
    all_hits = sum(per_part.values())
    print(f"[复扫] zip 内 .xml 部件 {len(names)} 个：{', '.join(names)}")
    for k, v in total.items():
        print(f"  {k}: {v}")
    print(f"[复扫] 只查 document.xml 命中 {only_doc} 个；全量命中 {all_hits} 个；漏检 {all_hits - only_doc} 个")
    code = 4 if all_hits else 0
    print(f"EXIT={code}")
    sys.exit(code)


if __name__ == "__main__":
    main()

脚本实测（python xml_unicode_scan.py，构造样本：内存构造的六部件 docx）：

  • [复扫] zip 内 .xml 部件 6 个：word/document.xml, word/header1.xml, word/footer1.xml, word/footnotes.xml, word/comments.xml, word/settings.xml
  • U+FFFD 替换字符: 1；私用区 Co: 1；零宽/格式 Cf: 5；mojibake 串: 1
  • [复扫] 只查 document.xml 命中 1 个；全量命中 8 个；漏检 7 个
  • EXIT=4

这里有两处值得说明。一是 mojibake_fix() 用「按候选编码重新编码再按 UTF-8 解码」的往返法判断乱码串，能解回汉字就记为命中，比硬列一张乱码字符表稳。二是按 Unicode 类别计数允许同一字符在不同维度各记一次，构造的乱码串里恰好带一个软连字符 U+00AD，它同时归入 Cf，所以 Cf 计到 5；8 是「命中次数」，不是「互异字符个数」，两者不要混用。

第五段把三道判据串成一条链，绑定退出码并短路。

# -*- coding: utf-8 -*-
"""三道质量闸门串联：判据、退出码与短路顺序。"""
from __future__ import annotations
import sys


def gate_rebuild():
    return True, "5/5 节标题·层级·编号一致，素材来源逐节留痕"


def gate_sync():
    return False, "15 格对照中 2 格不一致：旧标题 1、版本 v2 1"


def gate_scan():
    return True, "md 0 命中；docx 重组后 3 命中；全量 XML 8 命中"


# 每道闸门绑定独立退出码；0 只表示三道全部放行
GATES = (
    (2, "重建与回读", gate_rebuild),
    (3, "派生件同步", gate_sync),
    (4, "脱敏与复扫", gate_scan),
)


def run_pipeline(gates):
    lines = []
    for code, name, fn in gates:
        ok, detail = fn()
        lines.append(f"[{name}] {'放行' if ok else '拦截'}，{detail}")
        if not ok:
            return lines, code
    return lines, 0


def main():
    lines, code = run_pipeline(GATES)
    for line in lines:
        print(line)
    print(f"EXIT={code}")
    sys.exit(code)


if __name__ == "__main__":
    main()

脚本实测（python gates_pipeline.py，把前四段脚本的实测结论作为三道质量闸门输入）：

  • [重建与回读] 放行，5/5 节标题·层级·编号一致，素材来源逐节留痕
  • [派生件同步] 拦截，15 格对照中 2 格不一致：旧标题 1、版本 v2 1
  • EXIT=3

第二道拦截后，第三道没有执行——这就是短路的意义：不把一份还在挂旧标题的稿子送去做下游扫描。run_pipeline() 里 fn() 是惰性调用的，前面的闸门拦下，后面的检查连跑都不会跑。

五、踩坑：现象、成因、解法

这一节不是另列一份踩坑清单，而是把三道质量闸门各自的边界条件写清楚：每个现象都对应到具体判据，成因指向判据漏了什么，解法就是把那条判据补回去。

重建后编号看着连续，某一节却仍是旧口径

  • 现象：回读比对全部通过，但读者在某一节看到两套说法。
  • 成因：重建时把分稿自带的标题也当成骨架用了，骨架不再只有一个来源。
  • 解法：骨架字段只从权威源取，分稿只贡献正文素材块；回读时对章节 id、层级、编号、标题、版本逐字段比对，而不是只比编号。

md 脱敏零命中，docx 却被退回

  • 现象：源文扫下来干净，交付件检查不过。
  • 成因：run 拆分让敏感词在文件里不再连续，朴素串匹配只能扫到没被拆开的那几个。
  • 解法：先按段落重组全文再匹配，md 与 docx 两份都扫，两份都零命中才算过。

只查正文部件一切正常，页眉却冒出私用区字符

  • 现象：脚本报绿，打开交付件在页眉看到异常方块。
  • 成因：.docx 是压缩包，页眉、页脚、脚注、尾注、批注都是独立 XML 部件，只遍历正文那一份必然漏。
  • 解法：用 zipfile.namelist() 取全部 .xml 部件逐个读，按 Unicode 类别分类计数；异常字符范围要覆盖 U+FFFD、私用区、零宽字符与 BOM，乱码串用往返解码判定。

派生件同步矩阵少了一列

  • 现象：矩阵全绿，渠道元数据文件里却还挂着上一版标题。
  • 成因：对照清单只收了正文文件，把渠道元数据文件当成「配置」排除在外。
  • 解法：派生件一律作为一等公民进矩阵，标题、结论、版本三个字段全覆盖，格数按「份数 × 字段数」算清楚，不许凭印象说「应该都改了」。

脚本报了错，流水线却继续往下走

  • 现象：前一道检查已经失败，后一道仍然执行完并输出了结果。
  • 成因：调用方没有接收子进程退出码，或者把不同失败原因压成同一个返回值。
  • 解法：每道闸门绑定独立退出码，非零即短路；0 只表示三道全部放行，不复用、不覆盖、不吞掉。

六、总结：重建而非拼接，是一份可验收的工程契约

把四个事故和五段脚本放在一起看，结论其实很朴素：多个 AI 智能体分头写稿带来的不是「文本变多了」，而是「结构来源变多了」。AI 协作把产出速度提上去的同时，也把结构一致性变成了一笔必须显式支付的成本。结构来源一多，就必须有人显式声明哪一份是唯一权威源，其余全部降级为素材块；否则拼接出来的东西看着像一份文档，实际是好几份文档叠在一起。

重建之所以值得单独写成一条主流程，是因为它把三件原本靠默契的事变成了可验收的契约。骨架只认一份主稿，章节顺序、层级、编号、标题、版本都由它定义；正文按章节 id 回填，每个素材块的来源与版本逐节留痕；派生件不再是「顺手改一下」的附属物，而是与源文做字段级对照的一等公民。这三件事都能落到布尔判据上，也就都能落到退出码上。

三道质量闸门串起来之后，脱敏扫描与全量 XML 复扫不再是可有可无的收尾动作，而是流水线上的正式判据。md 扫完还要扫 docx，扫到字符串还不够，要按段落重组；docx 查完正文部件还不够，要遍历压缩包内全部 .xml 部件并按 Unicode 类别分类计数。任何一道给出非零，这一版就不标成可交付。

对舞台工艺这类交付物而言，DWG、齐套性、提资、吊杆、调光柜、DMX512、JGJ 57、WH/T 这些词本来就该原样留在文件里，GB/T、WH/T、JGJ 57-2016 这类规范标注也一个字都不能动。文档工程化要守的不是这些专业内容，而是「同一份内容只存在一个权威版本」这件事本身。回读比对输出的那张清单——每节来自谁、哪一版、编号是否连续——就是这份契约的验收单。合并稿的终点不是把文本拼完，而是能拿得出这张单子。

# python
Logo
北京朝阳AI社区

更多推荐

cover

桌面双屏翻译机 + 教育垂直大模型：涉外教育咨询语言沟通全流程落地实践

avatar 北京朝阳AI社区
cover

81:靠人不顾面子纠错？这才是最贵的系统设计

avatar 北京朝阳AI社区
cover

Gemma铁路调度智能体如何防幻觉

avatar 北京朝阳AI社区