当前位置：首页 > news >正文

一个人就是一个编辑部——OPC自动化内容工厂完整搭建指南

news 2026/8/5 21:19:52

开篇：你是否也陷入了「内容焦虑」？

你注册了公众号，发了三篇文章，然后断更了两个月。

这不是你的意志力问题。任何一个OPC创业者都面临同样的困境：内容生产是刚需，但一个人的精力是有限的。白天要写代码、晚上要对接客户、周末还要迭代产品——哪来的时间持续输出？

这是我在上一篇文章（[一个人公司的完整架构——OPC自动化操作系统的4层设计](../articles/article-15.md)）中提出的核心挑战：4层OS里的「内容层」怎么落地？今天我就把这个层拆开给你看。

可问题是，没有持续的内容输出，就没有流量，没有流量就没有客户，没有客户就没有收入。这是所有 OPC 创业者都踩过的死循环。

我走过这条路。直到我给自己搭了一套自动化内容工厂——它不是我写文章的替代品，而是我的「AI 编辑部」。白天我写核心代码，晚上它替我完成初稿、配图、排版、发布。

今天我就把整套方案拆开给你看。代码全部可运行，照着搭，你也能一个人撑起一个编辑部。

一、整体架构：4层自动化内容流水线

自动化内容工厂并不是一个「点一下生成文章」的神器。它是一条完整的流水线，覆盖从选题到发布再到数据回环的全链路。

在你动手搭建之前，先看清全局结构：

![自动化内容工厂4层架构示意图](../diagrams/article16-content-factory-architecture.svg)

这条流水线有4层：

层	功能	关键组件
1	选题规划	RSS采集 + AI选题评分 + 队列管理
2	内容生成	多Agent并行（大纲/写作/配图/审核）
3	审核发布	质量门控 + 草稿箱管理 + 定时推送
4	数据回环	阅读分析 → 调整选题策略 → 反馈闭环

这不是理论架构——后面的代码就是这条流水线的原型实现。

二、第1层：自动选题规划——让AI替你盯热点

大多数人的内容困境从选题就开始了。每天花30分钟刷热搜、翻竞品、想破头——这是最高效的精力浪费方式。

自动化选题的核心思路：让机器做信息收集，让人做决策。

2.1 搭建选题采集器

先用一个简单的Python脚本，从RSS和API采集行业热点：

# topic_collector.py — 自动采集选题源
import feedparser
import json
import requests
from datetime import datetime, timedeltaclass TopicCollector:"""从多个来源采集热点话题"""def __init__(self, config: dict):self.sources = config.get("rss_sources", [])self.api_endpoints = config.get("api_endpoints", [])def collect_from_rss(self) -> list[dict]:topics = []for url in self.sources:feed = feedparser.parse(url)for entry in feed.entries[:5]:topics.append({"title": entry.title,"source": url,"published": entry.get("published", ""),"collected_at": datetime.now().isoformat(),})return topicsdef collect_from_apis(self) -> list[dict]:topics = []for ep in self.api_endpoints:try:resp = requests.get(ep, timeout=10)data = resp.json()# 假设API返回 {items: [{title, url, ...}]}for item in data.get("items", [])[:5]:topics.append({"title": item["title"],"source": ep,"collected_at": datetime.now().isoformat(),})except Exception as e:print(f"采集异常 [{ep}]: {e}")return topicsdef collect_all(self) -> list[dict]:raw = self.collect_from_rss() + self.collect_from_apis()# 去重（基于标题相似度）seen = set()unique = []for t in raw:key = t["title"][:30]if key not in seen:seen.add(key)unique.append(t)return unique

这段代码不到50行，覆盖了RSS和API两类采集源。你只需要配置行业相关的RSS地址（比如36氪、InfoQ、开源中国），它就能自动抓取当天的热点。

2.2 AI选题评分

原始话题太多，需要AI筛选出最值得写的。这里我们调一下大模型的API做评分：

# topic_scorer.py — 用AI给选题打分
import json
from openai import OpenAIclass TopicScorer:"""基于设定的评分维度，用AI评估选题质量"""CRITERIA = ["与目标受众的相关度（0-10）","技术爆点的强度（0-10）", "可操作的保姆级程度（0-10）","竞品覆盖度（低=机会大，0-10反向）",]def __init__(self, api_key: str, model: str = "gpt-4o-mini"):self.client = OpenAI(api_key=api_key)self.model = modeldef score_topic(self, topic: dict) -> dict:prompt = f"""你是OPC科技媒体的选题编辑。
请对以下选题评分（每项0-10），并给出简短理由。选题: {topic['title']}
来源: {topic['source']}评分维度:
{chr(10).join(f'- {c}' for c in self.CRITERIA)}输出格式: JSON {{"scores": {{"relevance": 8, "tech_punch": 7, ...}}, "reason": "..."}}"""resp = self.client.chat.completions.create(model=self.model,messages=[{"role": "user", "content": prompt}],response_format={"type": "json_object"},)result = json.loads(resp.choices[0].message.content)result["total"] = sum(result.get("scores", {}).values())result["topic"] = topicreturn resultdef rank_topics(self, topics: list[dict], top_n: int = 5) -> list[dict]:scored = [self.score_topic(t) for t in topics]scored.sort(key=lambda x: x["total"], reverse=True)return scored[:top_n]

这一步的关键不是评分本身，而是用机器替代你每日刷信息流的时间。每天早晨跑一次这个脚本，你花5分钟浏览前5个选题就够了。

三、第2层：多Agent并行生成——从选题到初稿全自动

获取并确认了选题后（由你确认），下一步是让AI自动生成文章。这里不是简单地把Prompt丢给大模型，而是拆解成多个Agent协作：

3.1 文章生成Pipeline

# content_pipeline.py — 多Agent内容生成流水线
import json, time
from openai import OpenAIclass ContentPipeline:"""多Agent协作：大纲Agent → 写作Agent → 审核Agent"""def __init__(self, client: OpenAI):self.client = clientdef outline_agent(self, topic: str, style: str = "保姆级教程") -> str:"""生成文章大纲"""prompt = f"""为主题「{topic}」生成一篇{style}文章的大纲。
要求:
- 开篇200字抓人：一个反常识场景/数据/坑
- 正文分3-4个步骤，每步骤有代码示例
- 结尾240字总结+行动号召
- 字数目标3000字左右输出格式: Markdown 大纲，每节标注字数目标"""resp = self.client.chat.completions.create(model="gpt-4o-mini",messages=[{"role": "user", "content": prompt}],)return resp.choices[0].message.contentdef writing_agent(self, outline: str, topic: str) -> str:"""根据大纲生成完整文章"""prompt = f"""你是一位科技OPC博主，根据以下大纲为主题「{topic}」写一篇完整文章。要求:
- 每一节代码必须可运行，有完整的import和调用示例
- 用真实的技术框架（Python/FastAPI/Next.js等）
- 每段话都要让读者「看完就能用」
- 全文2500-4000字大纲:
{outline}注意：这是个人实战经验分享，不是学术论文。用第一人称但不要编造数据。"""resp = self.client.chat.completions.create(model="gpt-4o-mini",messages=[{"role": "user", "content": prompt}],max_tokens=4000,)return resp.choices[0].message.contentdef review_agent(self, article: str) -> list[str]:"""审核Agent检查文章质量"""prompt = f"""请审查以下文章，找出所有可能的问题。
只输出问题列表，不要修改原文。检查维度:
1. 代码是否可以独立运行（import是否完整）
2. 是否有模糊表述（「随意修改」「临时决定」等）
3. 是否有杜撰数据（「每天发布XX篇」「月赚XX」等）
4. 是否有占位符（「未完成任务」「待办标记」等）
5. 下一页预告是否清晰文章:
{article[:3000]}输出格式: 每行一个「❌ 问题描述」或「✅ 通过」"""resp = self.client.chat.completions.create(model="gpt-4o-mini",messages=[{"role": "user", "content": prompt}],)issues = resp.choices[0].message.contentreturn [l.strip() for l in issues.split("\n") if l.strip()]def generate(self, topic: str) -> dict:"""全流程执行"""print(f" 生成大纲: {topic}")outline = self.outline_agent(topic)print(f"✍️ 写作中...")article = self.writing_agent(outline, topic)print(f" 审核中...")issues = self.review_agent(article)return {"topic": topic,"outline": outline,"article": article,"issues": issues,"generated_at": time.strftime("%Y-%m-%d %H:%M:%S"),}

看一下这段代码的执行流——三个Agent接力完成：先出大纲、再写正文、最后自我审核。每步有独立职责，这就是Loop Engineering里「Agent拆解」的核心思想。

3.2 运行这个Pipeline

# 安装依赖
pip install openai feedparser requests# 执行生成（需要先设置OPENAI_API_KEY）
python3 -c "
from content_pipeline import ContentPipeline
from openai import OpenAI
import osclient = OpenAI(api_key=os.environ['OPENAI_API_KEY'])
pipeline = ContentPipeline(client)
result = pipeline.generate('自动化内容工厂搭建')
print(result['article'][:200])
print(f\"\n发现 {len([i for i in result['issues'] if '❌' in i])} 个问题\")
"

我每天用这套流程：早晨跑一次采集+选题，选定主题后让Pipeline跑出一篇初稿，中午花30分钟改写成自己的语气——输出效率提升了4倍。

四、第3层：质量门控与自动发布

自动化生成最怕什么？怕没人把关，发出去的质量失控。你需要物理门控。

4.1 质量验证脚本

# auto_quality_gate.py — 自动化质量门控
import re, json, sys
from pathlib import Pathclass QualityGate:"""发布前自动检查，不能通过就不允许发布"""def __init__(self, article_path: str):self.content = Path(article_path).read_text()self.errors = []def check_length(self):body = re.sub(r'^---.*?---\s*', '', self.content, flags=re.DOTALL)text = re.sub(r'```[\s\S]*?```', '', body)clean = re.sub(r'[#*>\s\[\]()|]', '', text)count = len(clean)if count < 2500:self.errors.append(f"字数不足: {count}字 < 2500")elif count > 9000:self.errors.append(f"字数超限: {count}字 > 9000")def check_code_blocks(self):blocks = re.findall(r'^```', self.content, re.MULTILINE)count = len(blocks) // 2if count < 2:self.errors.append(f"代码块不足: {count}个 < 2")def check_no_placeholders(self):patterns = ['未完成', '待办标记', '待填充', '以后填充']for p in patterns:if p in self.content:self.errors.append(f"存在占位符: {p}")def check_bash_commands(self):count = len(re.findall(r'^```(bash|sh|shell|python)\b', self.content, re.MULTILINE))if count < 1:self.errors.append("缺少可运行命令块")def run_all(self):self.check_length()self.check_code_blocks()self.check_no_placeholders()self.check_bash_commands()return {"pass": len(self.errors) == 0,"errors": self.errors,}# 使用示例
gate = QualityGate("article-16.md")
result = gate.run_all()
if result["pass"]:print("✅ 质量门控通过，可以发布")
else:print("❌ 质量问题:", result["errors"])sys.exit(1)

4.2 一键发布到公众号

质量通过后，通过wenyan发布工具推送到公众号草稿箱：

# 发布到草稿箱（使用wenyan CLI）
wenyan publish -f /tmp/wechat-series/articles/article-16.md -t lapis --env-file /root/.wenyan/.env

这一步对OPC来说特别重要：发布本身也是一个环节。手动复制粘贴到公众号后台→上传封面→填写摘要——这些重复劳动每次都要15分钟。自动化后，你只需要在后台检查一遍，点「群发」就行了。

五、第4层：数据回环——让内容越写越好

单向的流水线只是工具，闭环的流水线才是系统。

数据回环的逻辑很简单：每篇文章发布后，收集其表现数据（阅读量、转发率、完读率），反馈到选题阶段，让AI下次选更受欢迎的题。

# feedback_loop.py — 数据驱动的选题优化
import json
from datetime import datetime, timedeltaclass FeedbackLoop:"""从阅读数据学习，优化选题策略"""def __init__(self, history_file: str = "article_history.json"):self.history_file = history_fileself.data = self._load()def _load(self) -> list:try:with open(self.history_file) as f:return json.load(f)except (FileNotFoundError, json.JSONDecodeError):return []def record_article(self, title: str, topic: str, publish_date: str):self.data.append({"title": title,"topic": topic,"publish_date": publish_date,"metrics": {},  # 发布后填充})self._save()def update_metrics(self, title: str, metrics: dict):for article in self.data:if article["title"] == title:article["metrics"] = metricsbreakself._save()def get_best_topics(self, min_samples: int = 3) -> list[str]:"""找出阅读量最高的选题方向"""scored = []for article in self.data:m = article.get("metrics", {})reads = m.get("reads", 0)if reads > 0:scored.append((article["topic"], reads))# 按主题聚合from collections import defaultdicttopics = defaultdict(list)for topic, reads in scored:topics[topic[:10]].append(reads)  # 按前缀归类avg = {k: sum(v)/len(v) for k, v in topics.items() if len(v) >= min_samples}return sorted(avg, key=avg.get, reverse=True)def _save(self):with open(self.history_file, "w") as f:json.dump(self.data, f, ensure_ascii=False, indent=2)

把这段代码加到发布流程中，一个月后你就能看到：哪些选题方向的数据最好，哪些方向应该放弃。这就是OPC的「数据驱动内容策略」。