一、为什么【入口预检】是批量最划算的一步

前两章的提交环节（2.4.1、3.4.1）也有预检，但性质不同：前两章是一条一篇，脏了就重来；本章是一批，脏了会连累整批。

一句话：入口预检是唯一一处【花一分钟能省掉后面几十分钟】的地方。

二、这一节要做的三件事

任务表（4.2）          本节                    4.3.2 的工作流
  主题列        →  ① 预检 ② 组装请求 ③ 提交等待  →  契约 JSON

三、步骤一：预检的四条规则

规则 1：去重必须先【归一化】

这是最容易漏、也最容易吃亏的一条。 归一化的意思是：在比较之前，先把两个字符串里【不影响语义的差异】抹掉——空格、全半角、首尾空白。

为什么必须做：玄关收纳三件套 与 玄关收纳 三件套，人眼看是一回事，== 判出来是两回事。而平台判同质化看的是语义，不是你的字符串。你不归一化，平台帮你判——用扣分的方式。

规则 2：主题长度要有区间

太短（例如两个字）→ Agent 拆不出角度，容易写成空话； 太长（例如一整句话）→ Agent 抓不住重点，而且【一整个句子当主题】本身就是选题没想清楚的表现。

本节示例给的是 3 到 12 字，你可以按自己的垂类调。

规则 3：单批数量要有上限

为什么不能一次塞 50 个：模型在一批里产出越多，越趋同——它的注意力是有限的。按本节的示例值，单批 8 个以内，多了就拆批。

这一条同时是风控要求：一次产出太多，后面的发布节奏就会挤（4.2 的单日上限只有 6 篇），产出和发布不匹配，稿子只能堆在草稿箱里——而在草稿箱里堆一堆同风格草稿，本身就是风险。

规则 4：高危主题拦截（本节新增，也是最重要的一条）

这一条是前两章没有的，因为头条的扣分体系把【什么话题不能碰】写得很清楚：

四、步骤二：把约束写进请求，而不是写进代码

这是本节一个很小但很值钱的设计。 对比两种做法：

所以本节的请求体长这样（下面 Demo 里有完整输出）：

{
  "batch_id": "B03",
  "task": "batch_topic_to_contract",
  "topics": ["阳台改造小成本", "书桌线缆整理", "..."],
  "constraints": {
    "title_max_chars": 30,
    "body_min_chars": 100,
    "tags_count": "3-5",
    "ai_flag": true,
    "must_differ": ["angle", "persona", "opening", "structure"],
    "forbid_words": ["震惊", "必看", "不看后悔", "第一", "国家级"]
  }
}

注意 must_differ 和 forbid_words 这两个字段——它们就是本章的差异化节点（4.3.2）和扣分预检（4.4.4）在这份请求里的落地：约束不是检查表上的口号，它是随请求一起发给模型的指令。

五、步骤三：提交与等待——超时是常态

批量提交和单篇提交最大的区别在等待：

所以本节给两条纪律：

1. 提交要带 batch_id——它是这件事的追踪号，后面所有的日志、回查、重试都靠它；
2. 不要用【等页面出现某段文字】这种判断作为唯一依据（长任务的页面会刷新、会卡），要用【轮询 + 超时 + 落盘】：每收到一篇就写进任务表一次，这样断线之后你知道停在哪。

六、动手 Demo：主题批次预检 + 组装请求

下面这段代码只用 Python 标准库。它把【第三节的四条规则】变成可执行的检查，并把通过的主题组装成一份可以直接发给智能体的请求。

# -*- coding: utf-8 -*-
"""4.4.1 Demo：主题批次预检 + 组装提交请求（纯标准库）"""
import json
import re

BATCH_MAX = 8                     # 单批最多几个主题（别一次塞太多，模型会趋同）
TOPIC_MIN, TOPIC_MAX = 3, 12      # 主题字数区间

HISTORY = ["玄关收纳三件套", "厨房台面收纳", "冰镇酸梅汤"]   # 历史批次已用过的主题

NEW = [
    "玄关收纳三件套",                     # 与历史重复
    "玄关收纳 三件套",                    # 与历史重复（只多了一个空格）
    "阳台改造小成本",
    "夏季凉拌菜的做法大全与配比说明表",     # 过长
    "推荐三只稳赚不赔的股票",              # 高危主题
    "",                                   # 空
    "书桌线缆整理",
    "衣柜换季整理",
    "旧物改造花瓶",
]

BANNED = [
    ("金融荐股", r"荐股|稳赚|保本|涨停|翻倍|投资指导"),
    ("医疗功效", r"根治|特效|防癌|降血糖|包治|药到病除"),
    ("绝对化",   r"最好的|第一品牌|国家级|绝无仅有|全网最低"),
    ("时政敏感", r"领导|时政|大案|军演|司法案"),
]


def norm(s):
    return re.sub(r"\s+", "", s).strip()


print("=" * 64)
print("一、主题批次预检（提交前，共 %d 条）" % len(NEW))
print("=" * 64)

ok, killed = [], []
seen_hist = {norm(t) for t in HISTORY}
seen_new = {}

for i, t in enumerate(NEW, 1):
    tag = "S%02d" % i
    k = norm(t)
    if not k:
        killed.append((tag, t, "空主题"))
        continue
    n = len(k)
    if n < TOPIC_MIN:
        killed.append((tag, t, "主题仅 %d 字，太短（%d~%d 字）" % (n, TOPIC_MIN, TOPIC_MAX)))
        continue
    if n > TOPIC_MAX:
        killed.append((tag, t, "主题 %d 字，超出 %d 字上限" % (n, TOPIC_MAX)))
        continue
    hit = None
    for label, pat in BANNED:
        if re.search(pat, t):
            hit = label
            break
    if hit:
        killed.append((tag, t, "命中高危主题：%s（平台按最高档扣分处置）" % hit))
        continue
    if k in seen_hist:
        killed.append((tag, t, "与历史批次主题重复（归一化后相同）"))
        continue
    if k in seen_new:
        killed.append((tag, t, "与本批 %s 主题重复（归一化后相同）" % seen_new[k]))
        continue
    seen_new[k] = tag
    ok.append((tag, t))

for tag, t, why in killed:
    print("  [拦截] %-4s %-22s %s" % (tag, t[:22], why))
for tag, t in ok:
    print("  [通过] %-4s %s" % (tag, t))
print("  通过 %d 条 / 拦截 %d 条" % (len(ok), len(killed)))

if len(ok) > BATCH_MAX:
    print("  [!] 通过 %d 条 > 单批上限 %d 条，需要拆批" % (len(ok), BATCH_MAX))

print()
print("=" * 64)
print("二、组装提交给智能体的请求（干跑，不发出去）")
print("=" * 64)
req = {
    "batch_id": "B03",
    "task": "batch_topic_to_contract",
    "topics": [t for _, t in ok],
    "constraints": {
        "title_max_chars": 30,
        "title_target_chars": "18-28",
        "body_min_chars": 100,
        "body_good_chars": 800,
        "tags_count": "3-5",
        "ai_flag": True,
        "must_differ": ["angle", "persona", "opening", "structure"],
        "forbid_words": ["震惊", "必看", "不看后悔", "第一", "国家级"],
    },
}
print(json.dumps(req, ensure_ascii=False, indent=2))

实际运行输出（真跑，不是示意）：

================================================================
一、主题批次预检（提交前，共 9 条）
================================================================
  [拦截] S01  玄关收纳三件套                与历史批次主题重复（归一化后相同）
  [拦截] S02  玄关收纳 三件套               与历史批次主题重复（归一化后相同）
  [拦截] S04  夏季凉拌菜的做法大全与配比说明表       主题 16 字，超出 12 字上限
  [拦截] S05  推荐三只稳赚不赔的股票            命中高危主题：金融荐股（平台按最高档扣分处置）
  [拦截] S06                         空主题
  [通过] S03  阳台改造小成本
  [通过] S07  书桌线缆整理
  [通过] S08  衣柜换季整理
  [通过] S09  旧物改造花瓶
  通过 4 条 / 拦截 5 条

================================================================
二、组装提交给智能体的请求（干跑，不发出去）
================================================================
{
  "batch_id": "B03",
  "task": "batch_topic_to_contract",
  "topics": [
    "阳台改造小成本",
    "书桌线缆整理",
    "衣柜换季整理",
    "旧物改造花瓶"
  ],
  "constraints": {
    "title_max_chars": 30,
    "title_target_chars": "18-28",
    "body_min_chars": 100,
    "body_good_chars": 800,
    "tags_count": "3-5",
    "ai_flag": true,
    "must_differ": [
      "angle",
      "persona",
      "opening",
      "structure"
    ],
    "forbid_words": [
      "震惊",
      "必看",
      "不看后悔",
      "第一",
      "国家级"
    ]
  }
}

读结果，四个值得注意的地方：

1. S01 与 S02 被同一条规则抓下——这正是【归一化】的价值所在。 如果你用 == 判重，S02 会通过预检，然后它和 S01 一起进入生成环节，产出两篇几乎一样的稿子，到时候你要在更贵的地方（4.3.2 的同质化检测）才发现；
2. S05 是本节最值钱的一条：【推荐三只稳赚不赔的股票】这种主题，在预检这一关就该死掉。 注意拦截理由是【命中高危主题】而不是【长度不对】——它排在长度检查之后、提交之前，位置是对的：既然它不可能被写出来，就不要浪费一次生成；
3. 9 条进、4 条出——通过率不到一半。 这不是演示效果夸张，这是批量生产的真实情况：主题池子里永远混着重复的、太长的、碰红线的。预检的价值就是【在这把它们筛掉】，而不是【让它们一路走到发布】；
4. 请求体里的 constraints 是本节唯一真正【发出去】的东西。它把 4.3.2 的差异化要求（must_differ）和 4.4.4 的违禁词要求（forbid_words）提前写进了给模型的指令——约束是【发出去的】，不是【收回来才检查的】。

七、⚠️ 2026 现状：本节设计的三处依据

1. 提交方式正在从【点界面】转向【调接口】：文心智能体这类平台，除了网页对话，也在提供开发者接口。判断标准仍然是一条：能调接口就不点界面（第一章 1.x 的准则）。但注意：本节不依赖任何一种——预检与组装这两步，无论走界面还是走接口，代码都是一样的，这就是【先定字段再定步骤】的好处；
2. 高危主题的清单来自官方扣分细则：金融荐股、医疗功效、绝对化表述都对应头条官方的扣分条目，且 2026-03-24 起【违规推广】从 30 分调整为 20 分、【违反法律法规和相关政策】从 50 分调整为 70 分。关键词表要按最新公告维护，别照抄任何一篇教程；
3. 长任务的等待时间比旧教程描述的更长：旧教程常假设【几十秒出结果】。批量、长文、多节点工作流，实际等待往往是几分钟。 所以【轮询 + 超时 + 落盘】不是过度设计，是批量的必要配置。

30 秒实验

把 Demo 里 BATCH_MAX 从 8 改成 2，再跑一次。

会发生什么：

• 预检本身不会报错——因为它只在超过时提示一句（你这次通过 4 条、上限 2 条，会打印【需要拆批】）；
• 你会看到一个本该【拆开跑两批】的批次被完整地送出去了——上限只提示、不强制。

结论：上限是一个【批处理策略】，不是一条【对错规则】——它取决于你的模型、你的主题复杂度、你想要的差异化程度。所以它只提示，不阻断。

对比一下：高危主题是【对错规则】（必须阻断），批次上限是【策略选择】（可以提示）。 分清这两种，你的检查器就不会既烦人又漏事。

常见误区

1. 【主题重复用 == 判就够了】——错。 加一个空格、换一个字，== 就判不出来了。必须先归一化（去空格、统一全半角）再比较，因为平台判的是语义，不是你的字符串。
2. 【先让 Agent 写出来，写出来再检查有没有违规】——错，而且是本末倒置。 像【推荐三只稳赚不赔的股票】这种主题，从选题那一刻就已经违规了，写得多好都没用。拦主题，等于拦住了整篇的所有后续成本。
3. 【约束写死在脚本里，改起来快】——错。 写进请求体的 constraints，改一次只是改一行数据，而且它跟着批次被记录下来，事后能审计。能被数据表达的要求，就不要用代码表达。
4. 【预检会砍掉一半主题，说明规则太严了】——错。 那是批量生产的正常损耗。规划主题池就按产出目标的 1.5 到 2 倍准备——这是排期时要做的事，不是提交时该发现的事。
5. 【等结果就是等页面出现某段文字】——错。 长任务页面会刷新、会卡、会静默失败。正确做法是轮询 + 超时 + 每收到一篇就落盘一次，这样断线才知道从哪继续。

• ☐ 下一节（4.4.2 RPA获取创作结果）

相关笔记

• RPA提交创作信息 —— 单篇版的提交与前检
• RPA提交创作信息 —— 上一章的字段组装，本节多了一层【批次预检】
• 今日头条批量发布助手Agent部分的工作流设计 —— 本节请求里的 constraints 就是它的落地
• 今日头条批量发布助手工作流程设计 —— 主题池与排期、单日上限的关系
• 🤖 Agent+RPA实战 · 索引 —— 全课路线图