AI 自动化工作流实战：一条命令从数据到成片（选题、生成、组装、发布）

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

AI 自动化工作流实战：一条命令从数据到成片（选题、生成、组装、发布）

LLM 系列又一篇。批量处理篇讲了几千条数据怎么并发跑，这篇讲另一面：每天重复的内容生产体力活——刷数据、写稿、组装、发布，怎么压成一条命令。核心思想一句话：把创意留给自己，把体力全交给流水线。

一、先盘点：哪些环节值得自动化

拿内容生产举例，一天的体力活拆开看：

判断标准就一条：规则重复、不需要品味的环节，全部自动化；需要判断的环节，只留一个。整条链路里真正需要大模型的往往只有一步——从原始数据里挑出有价值的、并按模板生成。其他环节是普通代码。

二、架构：四个环节一条命令

数据抓取 → LLM 生成（筛选+写稿） → 模板组装 → 发布
↑ |
└────────── 明天再来一遍 ←───────────────┘

四个环节四个脚本，一条命令串起来：

python fetch_data.py && python generate.py && python assemble.py && python publish.py
# 或者做成 Makefile / npm script：make daily

每个环节独立脚本、中间产物落盘（data.json → draft.json → output/）——好处是失败可以单独重跑，不用整条链重来（批量篇的幂等老话）。

三、核心实现：一次调用顶一个环节

1. 最容易犯的错：每个条目调一次模型

25 条数据逐条生成，25 次调用、25 次 token 重复——把全部数据一次性喂给模型，让它自己筛选+生成，一次调用完成。模型能看到全局，筛选判断更准，token 省五倍。

2. 结构化输出 + 防编造约束

一次调用的 prompt 有三个刻意设计（结构化输出篇 + 垂类篇的老话合体）：

SYSTEM = """你是资深内容主编，从候选数据中挑出最有价值的 5 条并按模板生成。
严格要求：
1. 只输出一个 JSON 对象：
{"items": [{"headline": "20字以内", "body": "60-90字",
"tags": ["2-3个"], "source": "从候选中原样复制"}]}
2. body 口语化，讲清楚是什么、为什么重要；
3. source 必须从候选列表原样复制，禁止编造。"""

• 角色锚定：语气稳定，不会一会儿学术一会儿营销
• 每个字段给字数上限：防止某条写超了破坏整体节奏
• 引用字段强制「原样复制」：杜绝模型编造链接——AI 生成内容最容易翻车的点（垂类篇的 P0 拦截老话）

3. JSON 自愈重试

def chat_json(system: str, user: str, model: str, retries: int = 1) -> dict:
for i in range(retries + 1):
extra = "" if i == 0 else "\\n【重要】上一次输出不是合法 JSON，只输出 JSON 对象。"
text = call_llm(model, system + extra, user,
response_format={"type": "json_object"})
try:
cleaned = re.sub(r"```(json)?", "", text).strip()
start, end = cleaned.find("{"), cleaned.rfind("}")
return json.loads(cleaned[start:end + 1]) # 剥代码块 + 截首尾大括号
except (ValueError, json.JSONDecodeError):
if i == retries:
raise

开了 json_object 也不是银弹：模型偶尔还在 JSON 外包一层代码块或解释文字。解析层做「剥代码块 + 截取首个 { 到末个 }」的容错，再加一次带格式约束的重试——实测重试触发率不到 5%（结构化输出篇的容错层思路）。

四、组装：模板驱动，数据变成品

生成结果落盘后，组装环节是纯代码——数据驱动渲染：

TEMPLATE = """## {headline}\\n\\n{body}\\n\\n标签: {tags}\\n来源: {source}\\n"""

def assemble(draft: dict, out_dir: str):
for i, item in enumerate(draft["items"]):
text = TEMPLATE.format(**item)
pathlib.Path(out_dir, f"item-{i}.md").write_text(text, encoding="utf-8")

组装环节的原则：模板里没有逻辑，只有占位符——逻辑全在生成环节的结构化 JSON 里，模板只管排版。要改样式改模板，要改内容改 prompt，两边不纠缠。

五、发布 + 失败重跑

发布环节接各平台 API（CSDN、公众号都有接口）。两个工程细节：

def publish_all(items: list, published: set) -> list:
failed = []
for it in items:
if it["id"] in published: # 幂等：已发跳过
continue
try:
publish(it)
published.add(it["id"])
except Exception as e:
failed.append({"id": it["id"], "error": str(e)})
return failed # 补跑只跑 failed

六、成本账（接成本篇）

这种工作流的成本结构很友好：每天一次调用、一次几千输入 + 几百输出：

cost = task_cost(rounds=30, ctx_tokens=4600, out_tokens=700, # 一个月
price_in=2, price_out=8)
print(cost) # 一个月模型开销几块钱

对比每天两小时的人力，ROI 碾压级。省 token 的大头是「一次调用 > 多次调用」——25 条一次喂比逐条调省五倍，还让模型看到全局。

七、踩坑提醒

总结

结合系列：批量篇的并发、结构化输出篇的容错、成本篇的账本、可观测性篇的失败分类——自动化工作流就是那几张网的日常应用。每天重复的体力活，值得一条命令。觉得有用点个关注。