开篇：一条能跑的内容流水线，比一套漂亮的架构重要得多
写这篇文章之前，我看过不下二十个团队自建的内容项目。它们的起点几乎都一样：一个 Excel 文档里列着几百个选题，一个产品同学每天手动复制粘贴到对话框里，生成出来的稿子再手动贴回文档，改一改发出去。痛点是真实的——重复、低效、容易漏。于是有人决定”系统化”。
两周以后我再去问，大部分人的回答是：”搁置了，维护成本太高。”这不是个例，是常态。所以这篇文章讲的内容流水线，不是一套架构方案，而是一段真正能跑起来、有百行量级核心代码、坏了能一眼看出哪一段出错的最小实现。它不漂亮，但它会在周一早上六点自己跑起来，把稿子放进待审目录，然后给你发一封邮件。
这条内容流水线的全部野心，就是让”选题进、初稿出、人来把关、机器定时跑”这四件事，变成一条不需要你每天盯着的传送带。
为什么是”百行”，而不是一个框架
大部分自建内容流水线，死在过度设计上
我见过的失败案例，有一个高度相似的开局：一上来就引入消息队列，理由是”生成任务耗时长，需要异步”；再引入 PostgreSQL，理由是”要记录任务状态”；然后上 Celery 或 Airflow，理由是”要定时调度”；最后写一套 React 后台，理由是”人工审核需要界面”。
这套东西在图纸上没有任何问题，问题在于它假设了一件事：这条内容流水线会稳定产生足够的价值，来摊薄它的维护成本。而现实是，内容团队通常只有一到两个人懂后端，他们同时又得写稿、对稿子、跟运营沟通。当队列里卡了一条任务，他们需要去翻 worker 日志；当数据库 schema 要加一个字段，他们需要写迁移脚本；当后台页面报错，他们需要打开浏览器开发者工具。
于是出现那个经典场景：两周后，作者自己都不想打开这个项目。它不是坏了，它只是变得比手动复制粘贴更让人不想碰。
百行的价值：一周内跑通、坏了能一眼看出、别人能读完
百行不是目标，是约束条件产生的结果。当你强行把核心代码压到一百行左右，会发生三件好事：
- 一周内能跑通。没有基础设施要搭，没有服务要部署，一个 Python 文件加一个目录结构，当天就能出第一份稿子。
- 坏了能一眼看出。一百行代码，报错的行号就是问题所在的位置。不需要跨服务查 trace id，不需要确认是哪个 worker 挂了。
- 别人能读完。这一点常被低估。内容流水线的维护者往往不是写它的人。一个新人花二十分钟读完一百行代码，就能接手；读完两千行分布式架构的代码，他更可能选择重写。
还有第四个隐性好处：一百行的内容流水线，你可以毫无心理负担地推翻它。当你意识到输入格式需要大改，或者要换模型供应商，改一百行代码的成本是半小时，改一个框架的成本是一个周末——而一个周末足以让你放弃。
什么时候”百行”这个约束会失效
必须坦诚地说，百行不是万能的。当你的日更量超过几十篇、当有多个模型并行跑、当需要跨人协作审核时，文件系统就会开始吃力。本文后面有一节专门讲升级路径，会给出明确的信号：什么情况下该加数据库，什么情况下该拆多进程。但在那之前，请先用百行的版本跑满一个月，你会对”哪里真的疼”有完全不同的判断。
整体结构：四段工序与一个目录状态机
四段的职责边界
这条内容流水线被切成四段，每一段只干一件事，且只通过文件与下一段交接：
- 选题 JSON：输入契约。定义”这一篇要写什么”，是人与机器之间的接口。
- 初稿 Markdown：生成环节。读选题、拼提示词、调模型、校验结构、落盘。
- 人工卡点：最关键的一段。稿子写完不直接发布，落在待审目录。
- 定时运行：cron 拉起第一段到第二段，日志落盘，失败告警。
请注意这个设计的核心：四段之间没有共享内存、没有共享数据库连接、没有函数调用。第二段读 JSON 文件，第三段看 Markdown 文件。这意味着任何一段都可以被手动替换——你今天想手写一篇稿子塞进待审目录，完全没问题，流水线的其余部分照常工作。
目录即状态机的整体布局
整条内容流水线路由文件系统组织，结构大致如下：
content-pipeline/
├── pipeline.py # 核心，约 100 行
├── prompts/
│ └── draft.md # 提示词模板，与代码分离
├── topics/
│ └── queue.json # 选题队列
├── pending/ # 初稿，等待人工审核
├── approved/ # 审核通过，等待发布
├── published/ # 已发布
├── done.json # 已处理 id 记录，用于幂等
└── logs/
└── run-2026-11-05.log
这个结构本身就是文档。任何人打开目录，不需要读代码，就能知道整条内容流水线现在走到了哪一步——pending 里堆了十二篇，说明人在偷懒；approved 里堆了八篇，说明发布环节还没打通；published 里上周只有三篇，说明产出不稳定。
段一：选题 JSON——把输入变成契约
字段设计：少即是多
选题 JSON 是整条内容流水线的入口。我建议的字段只有七个，全部可选的不超过三个：
{
"topics": [
{
"id": "2026-11-05-001",
"title": "百行 Python 内容流水线：从选题到定时发布",
"keywords": ["内容流水线", "Python 自动化", "定时任务"],
"category": "工程实践",
"angle": "从维护成本角度讲最小实现，而不是讲架构",
"outline": ["为什么是百行", "四段结构", "人工卡点", "升级路径"],
"sources": ["https://example.com/internal-notes"]
}
]
}
每个字段的用途要说得清楚，否则写选题的人会随手填：
id：幂等的主键。一旦用过就永远不能复用，哪怕这篇稿子被废弃。建议用日期加序号。title：最终标题。允许生成环节微调，但不要让它从零开始编标题——标题是策略，不是生成结果。keywords：必须出现的关键词，会拼进提示词作为约束。category：分类，决定栏目、模板、审稿人。这是路由信息，不要省。angle：切入角度。这是最容易被忽略、也最影响成稿质量的字段。outline：可选。有则按纲写，无则让模型自己搭。大纲越具体，成稿越可控。sources：可选。参考资料链接或原文片段，用来压事实性错误。
为什么 angle 比 title 更重要
同一句标题，可以写成一篇泛泛而谈的科普，也可以写成一篇有明确判断的实操文。区别不在标题，在角度。比如”Python 自动化写作”，角度写成”介绍常见工具”，成稿就会变成罗列；角度写成”大多数人错在过度设计”，成稿会有立场，会有取舍，读者能记住。
所以在选题阶段，我最关心的不是标题写得多好，而是 angle 那一栏有没有写出一句带判断的话。如果 angle 是空的或者写着”综合介绍”，我会直接把这条选题退回——不是因为它不能写，而是因为它写出来大概率是一篇没人读完的稿子。
outline 用还是不用
给大纲与不给大纲的差异
经验上，当选题属于你已经很熟的领域，给 outline 更好，因为它能把生成结果的方差压下来；当选题属于你不熟的领域，不给 outline 更好，因为你自己搭的纲可能是错的，反而会误导模型。
一个折中做法：先在选题阶段只写三到五个一级小标题，让模型自己补二级结构。这样既保证了主干方向，又保留了它组织细节的自由度。
为什么用 JSON 而不是 Excel
这是被问得最多的问题。Excel 直观、人人会用、能筛选能排序，凭什么不用？答案不是 Excel 不好，而是它不适合当这条内容流水线的输入契约。下面这张表把差异摊开看：
| 维度 | JSON 选题 | Excel 选题 |
|---|---|---|
| 数据结构 | 天然嵌套，keywords、outline、sources 可以是数组 | 单元格只能是字符串，多值要靠逗号分隔，容易出错 |
| 版本管理 | 纯文本，git diff 能精确看到哪条选题改了什么字段 | 二进制文件，diff 不可读，多人改容易冲突 |
| 与模型互通 | 可直接作为提示词的一部分拼入，或作为结构化输出格式 | 需要先转 CSV 再转文本，字段边界模糊 |
| 校验能力 | 可写 schema，缺字段、类型错能立刻报错 | 校验靠人眼，空单元格与空字符串难以区分 |
| 程序读取 | 一行 json.load 搞定 | 需要依赖 openpyxl 或 pandas，处理合并单元格麻烦 |
| 适合人群 | 写代码的人，或愿意用简单约定规范填写的人 | 完全不碰代码的运营同学 |
看最后一行，这是 Excel 唯一的优势，而且它是真实的优势。所以我的建议不是”禁止 Excel”，而是让 Excel 只作为收集工具，不作为存储格式。运营在表格里填，交付前由一段二十行的脚本转成 JSON，转的时候顺手做校验。这样既照顾了填写体验，又保住了这条内容流水线的输入契约。
选题队列的管理方式
一个队列文件还是多个文件
初期一个 queue.json 足够。但当选题超过两百条，单个文件会变得笨重，每次人工编辑都可能引入语法错误，一错全挂。这时候可以按周拆：topics/2026-w45.json，每周一个文件，程序读目录下所有文件合并。
选题的生命周期
选题本身也有状态：待生成、已生成、已废弃。最省事的做法是不给选题文件加状态字段，而是用 done.json 记录已处理的 id——处理过就不再做，废弃的在选题文件里删掉即可。这就是下一节要讲的幂等。
段二：初稿 Markdown——生成环节的工程细节
提示词必须外置成文件
把提示词写在代码字符串里，是这条内容流水线最常见的坏味道。原因很实际：改提示词的人往往不是改代码的人，提示词里有一个错别字需要改，就得让人打开 Py 文件、改完、重新提交、重新部署——这个流程足以让所有人放弃优化提示词。
外置成 prompts/draft.md 以后，改提示词就是改一个 Markdown 文件，改完保存，下一次运行就生效。你甚至可以把它放到共享盘里，让运营同学直接调。
提示词模板的结构
四段式骨架
# 角色
你是一位在一线做了十几年的内容编辑，写东西克制、具体、有判断。
# 任务
根据选题信息，写一篇完整的 Markdown 初稿。
# 选题信息
- 标题：{{title}}
- 分类：{{category}}
- 必须包含的关键词：{{keywords}}
- 切入角度：{{angle}}
- 建议大纲：{{outline}}
- 参考资料：{{sources}}
# 硬性要求
1. 第一行输出 `# 标题`，使用上面给定的标题。
2. 第二段输出摘要，单独一段，100 字以内。
3. 正文至少 5 个二级标题，每个下面有具体展开，不要空泛。
4. 关键词必须自然出现，不要堆砌。
5. 不要编造具体数字、人名、机构名。不确定的地方用"通常""大多"这类模糊表述。
6. 只输出 Markdown 正文，不要输出解释、不要加代码块围栏。
# 反面示例
不要写成"首先……其次……最后……"这种流水账。
不要用"在当今时代"开头。
注意”反面示例”这一段。它的作用常常被低估：模型对”不要做什么”的敏感度，往往高于对”要做什么”的。你写十条正面要求，它可能只满足六条；你写两条反面示例，它基本能避开。
结构校验：不信任任何一次返回
模型返回的内容不一定合规。它可能忘了写标题，可能把摘要塞在正文中间，可能只写了八百字就停了。这条内容流水线的原则是：不校验的生成等于没生成。因为你会在三天后发现待审目录里堆了一堆残缺稿子。
校验最少要做四项：
- 是否有标题：首行是否以
#开头。 - 是否有摘要：标题之后的前若干行里，是否有独立成段的摘要文字。
- 字数是否达标：去空白统计中文字符数，低于阈值视为失败。
- 关键词是否落地：选题里的关键词是否实际出现在正文中。
为什么不做更重的校验
有人会想加语义校验、加事实核查、加风格评分。我不建议在百行版本里做。因为这些校验每一次都是另一次模型调用，成本翻倍，且误判率高——它会拦下本来可以用的稿子，让你开始不信任这条内容流水线。结构校验是确定性的，几乎零误判，这才是百行版本该做的事。
失败重试：三次封顶，带上错误信息
重试不是简单地把同样的请求再发一遍——那样大概率得到同样的坏结果。正确的做法是把校验失败的原因带回提示词，让它知道哪里不对：
“上一次输出缺少摘要段落，请补上 100 字以内的摘要，位于标题之后。”这句话带来的修复成功率，远高于盲目重试。
重试次数建议封顶三次。三次都失败，说明要么提示词有问题，要么这条选题本身不适合生成，两种情况都不该继续烧钱。失败的 id 记录下来，人工看一眼，可能是选题字段填得有问题。
段三：人工卡点——整条流水线最不能省的一段
写完不发布，落在 pending 等过目
生成成功的稿子，一律写入 pending/ 目录，文件名用选题 id，例如 pending/2026-11-05-001.md。然后这个人做一件事：打开这个目录，逐篇看。
审完之后，两个动作：满意的，mv 到 approved/；不满意的，删除，或改完再移。只有移动动作，没有”审核通过”按钮，没有状态字段，没有数据库更新。这就是这条内容流水线最朴素也最可靠的设计。
为什么人工卡点是刚需，不是保守
防止批量错误扩散
模型的错误是有模式的。如果它今天把”内容流水线”理解成了一个具体的产品，那么它会在你接下来生成的五十篇稿子里，反复地用同一个错误的定义。没有人把关的情况下，这种错误会在半天之内扩散到整个待发队列。
单篇出错不可怕，批量出错才可怕。而批量错误恰恰是自动化的固有风险——自动化会把一个错误乘以队列长度。
防止事实性错误上线
模型会编造。它编造的往往不是那种一眼可见的离谱内容，而是那种看起来非常合理、细节到位、但根本不存在的东西：一个听起来很像真的 API 名称，一组特别整齐的数字，一段似乎确实发生过的案例。
结构校验拦不住这些，语义校验也拦不住——因为它们在形式上完全合规。只有人，一个对领域有基本判断的人，能在一分钟内看出”我不记得有这个东西”。
人工卡点的效率问题
单篇审核控制在三分钟
审稿不是改稿。三分钟内你要做的判断只有几个：事实层面有没有让你起疑的地方；角度对不对；有没有明显的空洞段落。有疑虑的，直接丢弃，不要试图修改——修改一篇怀疑稿的时间，通常超过重新生成一篇。
为什么不做审核界面
很多团队的第一反应是做后台。我的建议是：在做满一百篇之前不要做。因为审核界面解决的是”效率”问题，而这条内容流水线在初期最缺的不是效率，是判断标准。当你还没有一套稳定的判断标准时，界面只会让你审得更快、审得更糙。
一百篇以后，如果你发现自己每天都在重复同一种修改，那才是做界面的信号——而且那个界面应该做的是把这种修改变成提示词规则。
目录即状态机：用文件移动表达状态
状态机的三个状态与两条边
这条内容流水线只有三个状态：待审、已批准、已发布。两条边：pending 到 approved（人审通过），approved 到 published（发布动作完成）。初始进入点只有一个：生成成功写入 pending。
每条边都是一个 mv 命令，或者更保守一点，是 shutil.move。为什么是 move 而不是 copy？因为 move 是原子的：文件要么在 pending，要么在 approved，不会同时出现在两个地方，也不会因为写一半而出现半成品。
这个性质很重要：它让你任何时候都可以通过 ls 确定这条内容流水线的真实状态，而不需要相信任何日志或字段。
与数据库方案的对比
下面把两种方案摊开对比。结论不是”数据库不好”，而是”在这个阶段，数据库带来的收益远小于它带来的维护成本”：
| 维度 | 目录即状态机 | 数据库方案 |
|---|---|---|
| 实现成本 | 一个 mv 命令，零依赖 | 建表、写迁移、连连接池 |
| 状态可见性 | ls 一眼可见，任何文件管理器都能看 | 需要客户端，需要写查询 |
| 人工干预 | 拖拽文件即可，非技术人员也能操作 | 需要给操作入口，否则无法手工改状态 |
| 原子性 | 单机文件系统内 rename 原子 | 需要事务，但事务写错也常见 |
| 并发写入 | 弱，多进程同时写同名文件会冲突 | 强，行锁、唯一索引都能用 |
| 查询与统计 | 弱，只能靠脚本遍历 | 强，任意维度聚合 |
| 备份与迁移 | 打包目录即可 | 需要 dump 与恢复流程 |
| 跨机器协作 | 需要共享盘或 git，冲突处理粗糙 | 天然支持 |
| 适合规模 | 每天几十篇以内 | 每天上百篇，或多人多进程 |
看”并发写入”和”跨机器协作”两行，这是目录方案的真正短板，也是升级的信号。但请先跑满一个月，确认你真的遇到了这个短板，而不是因为它听起来像问题就提前规避。
目录方案的坑
文件名会被污染
有人会在文件名里加状态后缀，比如 001_approved_v2.md。不要这样做。文件名只承载 id，状态只由所在目录表达。一旦文件名里混入状态，你就有两个真相来源，而它们迟早会不一致。
移动失败要能回滚
跨文件系统移动不是原子的。如果 pending 和 approved 在两个不同的挂载点上，mv 会退化成复制加删除，中途断电可能两边都有或都没有。解决很简单：把这三个目录放在同一个文件系统里。
幂等与断点续跑：重跑不会重复花钱
幂等的基本思路
这条内容流水线最贵的一步是调用模型。所以必须保证：同一篇稿子不会被生成两次。实现方式极简——维护一个 done.json，里面是一个 id 集合：
{
"done": ["2026-11-04-001", "2026-11-04-002", "2026-11-05-001"],
"failed": ["2026-11-05-002"]
}
每次运行，先读这个文件，跳过里面已经出现的 id。生成成功后立即写回。失败的 id 单独记录，不自动重试——留给人工判断，避免一条坏选题每天烧三次钱。
断点续跑的三个场景
中途崩溃
跑了五十篇，程序在第 37 篇的时候崩了。重跑时前 36 篇被 done.json 跳过，从第 37 篇继续。这就是断点续跑，不需要任何额外机制。
手动补跑
你想重写某一篇，从 done.json 里删掉那个 id，再运行即可。这是这条内容流水线里唯一需要手工维护的数据文件，但它足够简单，一眼就能看懂。
人工删除后重跑
这篇稿子审核没过，删掉了。你可以选择直接放弃，也可以把 id 从 done.json 移除，让流水线用同样的选题再生成一版。这种”再来一次”的能力，在有数据库的系统里通常需要写一个接口，在这里只需要编辑一个 JSON。
写 done.json 的时机
必须在稿子成功写入 pending 之后再写 done.json。顺序反了的话，一旦写 pending 失败，这次生成就永久丢失了——下次跑会跳过，但稿子根本不存在。这个细节是幂等设计里最常被写错的一处。
成本控制：不让这条流水线偷偷烧钱
并发数：二到四，不要更高
模型 API 通常有速率限制，并发开大会频繁撞到限流，触发重试，反而更慢更贵。经验值是把并发控制在二到四之间，这个区间对日更几十篇的量级完全足够。
而且并发不是越高越好，还有一个隐性的考虑：并发数越高，失败时你越难判断是模型的问题还是限流的问题。低并发让故障更可诊断。
单次 token 上限与超时
必须显式设置单次生成的最大 token 数。不设的话，模型偶尔会陷入重复输出的循环，一次就能烧掉你一整天的预算。上限建议按目标字数的三倍设置：写 800 字，上限设到 2400 token 左右，留出余量但不给浪费空间。
超时同理。设置一个单次请求的超时时间，比如九十秒。超时就当失败处理，不要无限等待——一个挂住的请求会拖住整个批次。
失败不无限重试
上一节已经讲过三次封顶。这里补充一个更隐蔽的成本：结构性失败的选题会每天重试。如果一条选题因为字段填错而永远生成失败，而你的 done.json 又不记录失败，那么它每天都会烧三次额度，一年就是上千次。
所以失败必须落盘，且失败后不再自动进入下一轮。这条规则看起来只是省钱，实际上它保护的是你对这条内容流水线成本的信心。
代码骨架：真正能跑的一百行
核心函数清单
整条内容流水线的代码只有六个函数：读选题、载入已完成、拼提示词、调模型、校验、落盘。下面是完整实现，除了模型客户端，它是可以原样运行的：
import json, os, re, time, pathlib, datetime
from concurrent.futures import ThreadPoolExecutor
ROOT = pathlib.Path(__file__).parent
PENDING, PROMPTS = ROOT / "pending", ROOT / "prompts"
DONE_FILE = ROOT / "done.json"
MAX_RETRY, TIMEOUT, MIN_CHARS, WORKERS = 3, 90, 600, 3
def load_topics(path):
data = json.loads(pathlib.Path(path).read_text(encoding="utf-8"))
for t in data["topics"]:
for k in ("id", "title", "keywords", "category", "angle"):
assert t.get(k), f"选题缺少字段 {k}: {t.get('id')}"
return data["topics"]
def load_done():
if DONE_FILE.exists():
return json.loads(DONE_FILE.read_text(encoding="utf-8"))
return {"done": [], "failed": []}
def save_done(state):
DONE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
def build_prompt(topic):
tpl = (PROMPTS / "draft.md").read_text(encoding="utf-8")
for k, v in topic.items():
if isinstance(v, (list, tuple)):
v = "\n".join(f"- {x}" for x in v) if v else "（无）"
tpl = tpl.replace("{{%s}}" % k, str(v or "（无）"))
return tpl
def validate(text):
errors = []
lines = [l for l in text.strip().splitlines()]
if not lines or not lines[0].startswith("# "):
errors.append("首行缺少一级标题")
body = re.sub(r"\s+", "", text)
cn = len(re.findall(r"[\u4e00-\u9fff]", text))
if cn < MIN_CHARS:
errors.append(f"中文字数 {cn} 未达 {MIN_CHARS}")
if not re.search(r"摘要[:：]", text) and not (len(lines) > 2 and lines[2].strip()):
errors.append("缺少摘要段落")
return errors
def to_markdown(text):
text = text.strip()
fenced = re.match(r"^```(?:markdown)?\s*\n(.*)\n```$", text, re.S)
return fenced.group(1) if fenced else text
def generate(topic, state):
tid = topic["id"]
if tid in state["done"] or tid in state["failed"]:
return f"skip {tid}"
prompt = build_prompt(topic)
last_err = ""
for attempt in range(1, MAX_RETRY + 1):
try:
raw = call_model(prompt, timeout=TIMEOUT, max_tokens=2400,
fix_hint=last_err if attempt > 1 else "")
text = to_markdown(raw)
errs = validate(text)
if not errs:
PENDING.mkdir(exist_ok=True)
(PENDING / f"{tid}.md").write_text(text, encoding="utf-8")
state["done"].append(tid)
save_done(state)
return f"ok {tid} 第{attempt}次"
last_err = "；".join(errs)
except Exception as e:
last_err = f"调用异常：{e}"
time.sleep(attempt * 2)
state["failed"].append(tid)
save_done(state)
return f"fail {tid} {last_err}"
def call_model(prompt, timeout, max_tokens, fix_hint=""):
# 把 fix_hint 追加到提示词末尾，让模型知道上次哪里不合格
if fix_hint:
prompt += f"\n\n# 上次输出不合格\n{fix_hint}\n请修正后重新输出全文。"
from openai import OpenAI
client = OpenAI(timeout=timeout)
r = client.chat.completions.create(
model=os.environ.get("MODEL", "gpt-4o-mini"),
messages=[{"role": "user", "content": prompt}],
max_tokens=max_tokens, temperature=0.7)
return r.choices[0].message.content
def main():
state = load_done()
topics = load_topics(ROOT / "topics" / "queue.json")
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
for line in pool.map(lambda t: generate(t, state), topics):
print(datetime.datetime.now().strftime("%H:%M:%S"), line, flush=True)
if __name__ == "__main__":
main()
容易写错的三处
提示词占位符替换
模板里用的是 {{title}} 双花括号。为什么不用单花括号？因为 Python 的 str.format 也会吃单花括号，而提示词正文里经常出现花括号字样，容易误伤。用双花括号加简单替换，可以避免整个格式化机制带来的问题。
模型返回的代码块围栏
模型常常会把 Markdown 包在三反引号里返回，哪怕你明确要求不要。所以 to_markdown 这一步必须做剥离。这个处理看起来像补丁，实际是必要的防御——你无法靠提示词百分之百约束住这一点。
done.json 的并发写
代码里用线程池并发生成，每个线程完成后都写一次 done.json。严格来说这不是线程安全的，两个线程同时写可能丢一条。在日更几十篇的量级下，这个风险可以接受；如果要处理，最简单的办法是给写入加一把文件锁，或者把写操作收拢到主线程。
段四：定时运行与日志
cron 的最小配置
这条内容流水线的调度只需要一行 crontab。建议每天早上六点跑，因为六点到九点之间是人工审核的窗口，审完刚好能赶上午间发布：
# crontab -e 中追加
# 每天 06:00 生成初稿，输出到日志，失败时邮件告警
0 6 * * * cd /srv/content-pipeline && \
/usr/bin/python3 pipeline.py >> logs/run-$(date +\%F).log 2>&1 \
|| /usr/bin/python3 notify.py "内容流水线生成失败"
# 每天 07:30 检查 pending 是否堆积过多
30 7 * * * cd /srv/content-pipeline && \
/usr/bin/python3 notify.py "$(ls pending | wc -l) 篇待审"
# 每周一 09:00 确认还有选题可跑
0 9 * * 1 cd /srv/content-pipeline && \
/usr/bin/python3 notify.py "本周剩余选题检查"
注意三个细节：cron 里 % 必须转义成 \%，否则会被当成换行；必须 cd 到项目目录，因为 cron 的工作目录不是你的家目录；必须用绝对路径调 python，因为 cron 环境里的 PATH 和你终端里不一样。
日志落盘
日志格式
日志只记三类信息就够：时间、id、结果。结果分 ok、fail、skip 三种。好的日志长这样：
06:00:01 skip 2026-11-04-001
06:00:14 ok 2026-11-05-001 第1次
06:01:02 ok 2026-11-05-002 第2次
06:03:45 fail 2026-11-05-003 中文字数 512 未达 600
这一行 fail 就把问题说清楚了：字数不够，不是调用失败。有了这一行，你不需要打开任何调试工具。
按天切分
日志文件名带日期，比如 run-2026-11-05.log。这比配 logrotate 简单，也更容易在半年后找到某一天的记录。代价是目录会慢慢变大，定期归档即可。
失败告警
最简单的告警：退出码加邮件
cron 命令失败时 || 会触发后面的处理。用一段十几行的脚本发邮件即可，SMTP 加一个公共邮箱，成本几乎为零。关键是：告警要带上下文，不要只发”失败了”三个字，要带上日志最后二十行。
告警的克制
最容易被忽略的一点：告警必须克制。如果每次有单篇失败就发邮件，一周之后你会自动忽略所有来自这个地址的邮件——包括真正重要那封。所以告警分级应该这样：单篇失败不告警，只记日志；整批失败（退出码非零）才告警；pending 堆积超阈值告警。
大多数人做错的地方
错误一：先做审核界面，后定质量标准
界面是效率工具，效率工具只在标准清晰时才有价值。当标准还没建立，界面会让你审得更快、审得更糙，然后你会得出”自动生成质量不行”的结论。而真实情况是，你还没想清楚什么叫好稿子。
错误二：把提示词优化当成一次性工作
提示词不是写一次就固定的配置。它是一份一直在迭代的文档。我自己的习惯是每一批审稿时记下淘汰原因，一周后把这些原因合并成一条新的提示词规则。三个月后提示词会比开始时长一倍，而成稿质量会明显提升。把提示词藏进代码里，你就失去了这个迭代循环。
错误三：追求全自动，省掉人工卡点
这是最危险的错误。它会让你在某个周末发现，待发队列里有一百篇文章都用了同一个错误的事实。批量错误的成本不是线性的，而是指数级的——因为你不仅要清理，还要处理已经发出去的那些。
一个反常识的判断：人工卡点不是过渡方案，它是长期设计。哪怕这条内容流水线跑了一年，你也应该保留人工过目。区别只在于审核时长从三分钟降到三十秒，而不是从有变成没有。
错误四：追求架构完整，忽略跑通速度
先跑通，再优化。这条内容流水线的价值不在于它多优雅，而在于它今天就开始替你干活。一个跑了三个月、每天产出十篇的百行流水线，比一个写了三周、还没跑过一篇的优雅架构有用一百倍。
逐步升级路径：什么时候该动
什么时候该加数据库
出现以下任意一个信号时，才考虑数据库：
- 审核的人超过两个，且开始出现两个人同时改同一个文件。
- 需要按分类、按作者、按时间做统计，且已经开始手写脚本遍历目录。
- 日产量稳定超过三十篇，pending 目录已经需要用搜索才能找到某篇。
注意”稳定”两个字。偶发的峰值不是信号，稳定的常态才是。而且即使加了数据库，也建议保留目录结构作为另外一层可读状态——数据库负责查询，目录负责肉眼可见。
什么时候该接发布接口
当 approved/ 目录开始堆积，而你每天要花固定时间手动上传时，就该接了。典型信号是：approved 里长期有超过二十篇积压，且这些稿子的发布动作完全标准化。
接入时的一个建议：把发布动作写成单独的脚本 publish.py，读取 approved 目录，逐篇调用接口，成功后移到 published。不要把这部分逻辑塞进 pipeline.py，因为发布接口的变更频率远高于生成逻辑。
什么时候该拆成多进程
当单条生成耗时超过你的单次预算，或者并发数已经压不住总耗时，才需要拆。典型情况是单篇生成需要两分钟以上，而你每天要出三十篇——这时候线程池的时间已经不可接受。
拆法也很简单：保留 pipeline.py 的读题与校验逻辑，把”调模型”这一步改成写一个任务文件到 jobs/，由另一个进程消费。这是最原始的工作队列，但它足够用，且不引入任何新依赖。
升级的原则：每次只动一处
无论升级哪一部分，原则都一样：一次只动一个环节，动完跑满一周，确认没有退化，再动下一个。因为这条内容流水线的所有价值都建立在”能跑”之上，任何一次改动让它跑不起来，之前积累的信任就会归零。
把今天就能做的事做完
如果你读到这里，我给一个最小行动建议：今天先建四个空目录，把一条选题写成 JSON，把提示词写成 Markdown，把上面那段代码存成 py 文件，用一个假模型跑通流程。哪怕生成的是假内容，整条内容流水线的骨架就已经立起来了。
明天接真模型，后天挂 cron，周末开始审第一批稿子。一周之后你会拥有一个每天替你工作的内容流水线，而你要做的，只是每天早上花半小时看稿。
这就是百行的全部意义：不是因为它聪明，而是因为它足够简单，简单到你会一直用它。
FAQ：关于内容流水线的六个常见疑问
为什么不直接用现成的内容生成平台，非要自己写代码？
现成平台的问题是黑盒：你无法改提示词、无法校验结构、无法控制卡点和发布时机。自己搭一条百行的内容流水线，所有环节都在你手里，出了问题能定位，有想法能立刻改进。它并不比平台强大，但它比平台透明，而对内容团队来说，透明往往比强大更重要。
人工卡点会不会成为瓶颈，让内容流水线跑不起来？
会，但这是好事。卡点慢下来，说明你有稿子可审，而不是在空转。实际跑起来以后，审稿很快就会从逐字读变成扫读，单篇三分钟降到一分钟。真正需要担心的是反过来的情况：卡点太松，堆积的稿子没人审，那说明内容流水线的产出超出了团队的消化能力，该做的是降低生成频率，而不是取消卡点。
模型生成的内容每次都差不多，怎么办？
先查 angle 字段。大多数”都差不多”的情况，是因为选题里的 angle 写得太泛。同一个角度写十篇，结果自然雷同。把这十篇的 angle 分别写成不同的判断，结果会立刻分化。其次查提示词里的反面示例——一份好的反面示例能有效拉开文风差距。最后才是考虑换模型。
done.json 会越来越大，需要清理吗？
不需要。一个 id 大概二十个字节，一万篇也就两百 KB，完全不是问题。反而应该保留，因为它是这条内容流水线唯一的生成历史。真要清理时，按年拆成 done-2026.json、done-2027.json 即可，不用引入别的机制。
为什么不做语义校验，用模型检查模型？
可以做，但不该放在百行版本里。模型检查模型有两个问题：一是成本翻倍，二是误判率高。它会拦下明明可以用的稿子，让你逐渐不信任这条内容流水线，最后干脆绕过校验。结构校验虽然浅，但它是确定的、零误判的，这才是自动校验该有的样子。语义层面的判断，交给人工卡点。
这套内容流水线能承载多大的量？
在目录加单机的方案下，日更五十篇以内是舒适的。超过这个量，文件系统开始吃力，审核也会成为瓶颈。但请记住：你的瓶颈通常不是生成速度，而是审稿能力。一天写出一百篇没人审的稿子，价值是零。所以在扩量之前，先确认你的审核产能跟得上——这比任何技术升级都更值得先想清楚。
暂无评论内容