AI博主站上风口之后,真正值得关注的不是“一天能生成多少字”,而是内容生产能不能变成一条稳定、可控、可迭代的流水线。很多人把重点放在大模型本身的生成能力上,实际做过 AI 内容产品的人会告诉你,纯粹调用模型输出文本并不难,难的是把选题、初稿、评审、配图、视频素材、发布和数据反馈组织成一个闭环。这篇文章从工程实践角度,以 Python 为例搭建一条 AI 博主内容生产流水线,跑通“选题到成文”的最小闭环,再延伸到多模态素材、Agent 编排、云服务器部署和常见问题排查。适合正在学习大模型应用开发、想进入 AI 内容领域,或者打算用 AI 工具提升内容生产效率的开发者参考。
1. 先拆解AI博主的完整生产链路,再决定哪些环节自动化
1.1 一条内容从选题到发布,到底经过哪些环节
所谓 AI 博主,并不是“让模型写一篇文章,然后发出去”这么简单。完整的内容生产链路可以拆成六个环节:
- 选题策划:确定目标读者、热点方向、差异化角度。
- 素材收集:查阅资料、整理数据、确认事实来源。
- 内容写作:标题、提纲、初稿、修改、排版。
- 可视化:配图、代码块截图、封面图、视频素材、字幕。
- 多平台发布:适配不同平台的排版习惯、标题风格和封面尺寸。
- 数据回流:阅读量、评论、收藏和转化数据,反向修正下一个选题。
把这六步列成表格,就能直观看出哪里适合自动化,哪里必须保留人工。实际项目里最不应该做的,就是让模型一口气完成从选题到发布的所有步骤。
| 生产环节 | 人工成本 | 自动化难度 | 自动化建议 |
|---|---|---|---|
| 选题策划 | 中 | 中 | 让模型生成候选选题,人工筛选 |
| 素材收集 | 高 | 高 | 接检索、知识库或网页读取工具 |
| 内容写作 | 高 | 中 | 先用模板生成初稿,再人工修订 |
| 可视化 | 中 | 中 | AI 配图、TTS 配音、剪裁脚本半自动 |
| 发布 | 中 | 低 | 调用平台开放 API 或半自动发布 |
| 数据反馈 | 中 | 中 | 定时拉取数据,生成趋势报告 |
表格里的关键判断是:发布环节自动化难度低,但风险高,因为不同平台对 AI 内容、外链和格式的规则并不一致,发布前仍然需要人工确认。
1.2 自动化优先级:先跑通文本,再扩展多模态
初期的自动化优先级应该非常明确:先解决文本生成 Pipeline,再考虑图片和视频。理由是文本链路最容易验证,一次调用就能看到结果,也最容易定位是提示词问题、模型问题还是解析问题。
推荐顺序是:
- 第一优先:文本生产 Pipeline,包括选题生成、初稿生成、自动评审。
- 第二优先:质量检查,把低分内容过滤掉,只让合格内容进入人工审核。
- 第三优先:多模态素材生成,例如配图、语音、视频分镜。
- 第四优先:发布和数据回流。
这个顺序的好处是每一步都有独立的验证标准。文本 Pipeline 跑不通,不要急着接视频接口,否则出错时很难判断是哪个环节出了问题。
1.3 一套可复现的最小闭环
本文要搭建的最小闭环是这样的:本地 Python 脚本读取一个领域关键词,生成 5 到 10 个候选选题;对每个选题生成初稿;再用一个评审提示词对初稿打分;最后把合格的文章保存到本地目录。
全流程不追求 100% 无人化。建议把人工介入率控制在 30% 左右,也就是模型负责初稿和初筛,人负责选题确认、事实核查和最终修改。这样既保留效率,又避免生成内容出现事实性错误却无人发现的问题。
2. 搭建AI内容生产工作流前,先对齐环境与依赖
2.1 Python 环境准备
下面的示例基于 Python 3.10 及以上版本。先在本地创建项目目录,再建立虚拟环境,避免依赖冲突。
mkdir ai-blogger-pipeline cd ai-blogger-pipeline python -m venv .venv source .venv/bin/activate如果在 Windows 上操作,激活虚拟环境的命令是.venv\Scripts\activate。激活后执行pip install --upgrade pip,把包管理工具升级到较新版本。
2.2 依赖清单
用一个requirements.txt管理依赖。示例项目只需要几个核心库:
openai>=1.30.0 pydantic>=2.0.0 python-dotenv>=1.0.0 jinja2>=3.1.0 apscheduler>=3.10.0说明一下每个库的用途:
| 依赖 | 用途 | 说明 |
|---|---|---|
| openai | 调用大模型接口 | 新版 SDK 支持任意 OpenAI 兼容接口,只需改 base_url 和 model |
| pydantic | 结构化输出解析 | 后续做评审结果、配置解析时更稳妥 |
| python-dotenv | 加载环境变量 | 避免把 API Key 写进代码 |
| jinja2 | 渲染提示词模板 | 把提示词从代码里拆出来,便于维护 |
| apscheduler | 定时任务 | 让 Pipeline 可以每天定时跑 |
如果你的模型服务不是这个 SDK,也可以换成对应的官方 SDK,或者直接用 requests 调用 HTTP 接口。工程上的核心是接口层统一,这样后续切换模型服务时不用改业务代码。
2.3 项目目录结构
推荐按“代码、配置、提示词、产物”四类文件分开。目录结构如下:
ai-blogger-pipeline/ ├── .env ├── requirements.txt ├── content_pipeline.py ├── run_pipeline.py ├── check_connection.py ├── prompts/ │ ├── topic.jinja2 │ ├── draft.jinja2 │ └── review.jinja2 ├── outputs/ └── logs/prompts/放提示词模板,content_pipeline.py放核心 Pipeline 类,run_pipeline.py是入口脚本,outputs/按日期保存生成结果。这样做的原因是:提示词改动的频率远高于代码,拆开以后可以直接调整模板,不需要重新修改 Python 逻辑。
2.4 环境变量与最小连通性验证
在.env文件中保存模型接口配置:
LLM_API_BASE=https://your-llm-service.example.com/v1 LLM_API_KEY=sk-your-key-here LLM_MODEL=your-model-name OUTPUT_DIR=./outputs这里的接口地址只是一个占位写法。实际项目中,这个地址要替换成你真实使用的模型服务地址,并且要确认服务提供 OpenAI 兼容接口。以下是启动前验证连通性的最小脚本check_connection.py:
import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client = OpenAI( base_url=os.getenv("LLM_API_BASE"), api_key=os.getenv("LLM_API_KEY"), ) resp = client.chat.completions.create( model=os.getenv("LLM_MODEL"), messages=[{"role": "user", "content": "只回复两个字:正常"}], temperature=0, ) print(resp.choices[0].message.content)运行命令:
python check_connection.py预期输出是“正常”。这一步能暴露绝大部分配置问题。这里最常见的坑有三个:
- API Key 复制时带有空格或换行,导致鉴权失败。
LLM_API_BASE末尾多写了斜杠,例如写成https://xxx/v1/,部分服务会返回 404。- 模型名称填写错误。模型名不是系统名,而是服务商提供的具体部署名,填错会直接报“model not found”。
学习环境里,可以用本地模型或云端免费额度快速验证;生产环境则要固定模型版本,不能今天一个版本明天一个版本,否则内容风格和质量很难稳定。
3. 核心实现:把“选题到成文”写成可复用 Pipeline
3.1 为什么提示词要模板化,而不是手写在代码里
把提示词直接写在 Python 字符串里不是不行,但后续维护成本很高。提示词稍有变化,就要重新发布代码。更好的做法是把提示词放进独立模板文件,用 Jinja2 渲染变量。
这样做有三个实际收益:
- 改动文案不需要改代码,直接编辑
.jinja2文件。 - 模板可以进入 Git 版本管理,方便对比不同提示词版本的效果。
- 可以针对不同内容方向准备多套模板,例如“技术教程风格”和“产品评测风格”。
3.2 选题生成模块
新建prompts/topic.jinja2:
你是{{ niche }}领域的选题策划编辑,请基于当前内容趋势,给出{{ count }}个有明确读者场景的内容选题。 要求: 1. 每个选题包含目标读者、核心痛点、预期解决的问题。 2. 用 Markdown 无序列表输出,每行一个选题。 3. 不要输出选题之外的说明文字,不要写开头语和结尾语。这里让模型输出 Markdown 列表,是为了方便代码解析。如果输出长段文字,解析逻辑会很脆弱。把temperature调高到 0.9,可以让选题更多样,减少重复。
3.3 初稿生成模块
新建prompts/draft.jinja2:
你是一位技术写作编辑。请根据下面的选题信息,产出一篇适合技术博客发布的正文草稿。 选题:{{ topic }} 目标读者:{{ audience }} 写作要求: - 开头直接说明问题和适用人群,不写客套话。 - 像有经验的技术开发者在讲解,解释每一步背后的原因。 - 必须包含代码示例或配置片段,并解释关键参数。 - 如果信息不足,不要编造数据,明确标注“需要人工补充确认”。 - 输出 Markdown 格式,不要输出与正文无关的内容。这个模板里最关键的是“如果信息不足,不要编造数据”。大模型很擅长生成看起来合理的错误数据,如果不加约束,生成的教程可能会包含根本不存在的依赖版本或过时的 API 用法。
3.4 质量评审模块
新建prompts/review.jinja2:
请检查下面的文章草稿,输出 JSON 格式的评审结果。 输出格式: {"score": 1-10, "issues": [], "suggestions": []} 检查维度: 1. 技术准确度。 2. 结构完整性。 3. 是否有空话、套话。 4. 是否缺少运行和验证步骤。 5. 是否包含未经验证的结论。 文章草稿: {{ draft }}这里要求输出 JSON,是为了让代码可以直接解析分数。低于某个阈值的初稿可以自动打回重写,也可以进入人工修改队列。这一步等于给 Pipeline 加了一道自动质检关卡。
3.5 Pipeline 主类与运行脚本
核心文件content_pipeline.py实现一个ContentPipeline类,把三个模块串起来:
import json import os import re from pathlib import Path from typing import Any from dotenv import load_dotenv from jinja2 import Environment, FileSystemLoader from openai import OpenAI load_dotenv() def parse_topics(text: str) -> list[str]: lines = [line.strip() for line in text.splitlines() if line.strip().startswith("- ")] return [line[2:].strip() for line in lines] def clean_json(content: str) -> str: content = content.strip() if content.startswith("```"): content = re.sub(r"^```(?:json)?\s*|\s*```$", "", content) return content class ContentPipeline: def __init__(self, model: str, prompt_dir: Path): self.client = OpenAI( base_url=os.getenv("LLM_API_BASE"), api_key=os.getenv("LLM_API_KEY"), ) self.model = model self.jinja_env = Environment(loader=FileSystemLoader(prompt_dir)) def chat(self, system: str, user: str, temperature: float = 0.7) -> str: resp = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=temperature, ) return resp.choices[0].message.content or "" def generate_topics(self, niche: str, count: int = 10) -> list[str]: template = self.jinja_env.get_template("topic.jinja2") prompt = template.render(niche=niche, count=count) return parse_topics(self.chat("你是一个严谨的内容策划助手。", prompt, temperature=0.9)) def generate_draft(self, topic: str, audience: str) -> str: template = self.jinja_env.get_template("draft.jinja2") prompt = template.render(topic=topic, audience=audience) return self.chat("你是一位有十年经验的技术写作者。", prompt, temperature=0.7) def review_draft(self, draft: str) -> dict[str, Any]: template = self.jinja_env.get_template("review.jinja2") prompt = template.render(draft=draft) content = self.chat("你是一个严谨的技术编辑,只输出 JSON。", prompt, temperature=0.2) try: return json.loads(clean_json(content)) except json.JSONDecodeError as exc: return {"score": 0, "issues": [f"评审结果解析失败: {exc}"], "suggestions": [content]} def save_draft(self, topic: str, draft: str, output_dir: Path) -> Path: output_dir.mkdir(parents=True, exist_ok=True) safe_topic = re.sub(r"[^\w\u4e00-\u9fff-]+", "_", topic)[:30] path = output_dir / f"{safe_topic}.md" path.write_text(draft, encoding="utf-8") return path入口脚本run_pipeline.py负责读取配置、生成选题、写初稿和打印评审结果:
import os from pathlib import Path from content_pipeline import ContentPipeline if __name__ == "__main__": model = os.getenv("LLM_MODEL", "") if not model: raise SystemExit("请先在 .env 中配置 LLM_MODEL") pipeline = ContentPipeline(model=model, prompt_dir=Path("prompts")) niche = os.getenv("CONTENT_NICHE", "AI 工程实践") topics = pipeline.generate_topics(niche=niche, count=5) for topic in topics: print(f"选题: {topic}") draft = pipeline.generate_draft(topic, "有 Python 基础的后端开发者") review = pipeline.review_draft(draft) print(f"评审: {review}") if review.get("score", 0) >= 6: output_path = pipeline.save_draft( topic=topic, draft=draft, output_dir=Path("outputs"), ) print(f"已保存: {output_path}")运行方式:
python run_pipeline.py正常运行时会依次打印选题、评审结果和保存路径。如果评审分数低于 6,当前文章不会保存,这样可以避免低质量内容进入输出目录。
这里要强调的是:代码里的score >= 6是一个示例阈值,实际项目中要根据内容类型调整。技术教程类内容通常要求更高,可以设置为 8 分以上才放行;资讯摘要类内容可以适当放宽。
4. 从文本到多模态:AI配图、音频与视频的扩展路径
4.1 AI 配图的工程化接入
文本 Pipeline 跑通之后,下一步通常是给文章配图。很多模型服务提供图像生成接口,调用方式和文本接口类似。示例代码如下:
import os from dotenv import load_dotenv from openai import OpenAI load_dotenv() client = OpenAI( base_url=os.getenv("LLM_API_BASE"), api_key=os.getenv("LLM_API_KEY"), ) resp = client.images.generate( model="your-image-model", prompt="技术博客封面图,程序员在调试大模型,简洁扁平风格,无文字", size="1024x1024", ) print(resp.data[0].url)需要注意,不同模型服务的图像接口返回格式不一样,有的返回 URL,有的返回 base64。实际接入时要先看接口文档,再决定是直接下载图片还是解码保存。
AI 配图目前最常见的问题是画面一致性和版权。同一系列文章最好固定画风和关键词,并且确认模型服务商的图片使用授权范围,避免商用后产生版权纠纷。
4.2 “AI视频一键成片”,工程上到底在做什么
“AI 视频一键成片”这个词很容易让人误以为只需输入一个主题就能得到完整视频。实际工程上至少包含四个环节:
- 脚本生成:把文章或大纲拆成多条分镜脚本。
- 语音合成:为每条脚本生成配音。
- 画面生成:用视频生成或图片生成工具产出画面素材。
- 字幕与合成:把画面、音频、字幕对齐,再导出成视频。
一个简化的结构化视频脚本 JSON 如下:
{ "title": "如何搭建AI博主内容流水线", "shots": [ { "duration_seconds": 5, "voiceover": "先用选题模块生成10个候选题材", "visual_hint": "电脑屏幕上有多个选题卡片", "subtitle": "选题自动化" }, { "duration_seconds": 5, "voiceover": "再让大模型生成初稿", "visual_hint": "编辑器里自动出现正文", "subtitle": "初稿自动化" } ] }有了脚本 JSON,后续的 TTS 配音、画面生成和字幕对齐都可以按字段消费。视频合成的最后一步也可以用 ffmpeg 完成:
ffmpeg -i shot1.mp4 -i shot2.mp4 -filter_complex \ "[0:v][0:a][1:v][1:a]concat=n=2:v=1:a=1[v][a]" \ -map "[v]" -map "[a]" -c:v libx264 -c:a aac final.mp4这条命令把两个视频片段按顺序拼接,保留各自音轨并合并。AI 短剧、AI 漫剧等内容形态,本质上都是这条管线在分镜数量上的扩展。
4.3 文本 Pipeline 与多模态 Pipeline 如何衔接
文本 Pipeline 的结果应该作为多模态 Pipeline 的输入,而不是把图片和视频逻辑都写进同一个脚本。
推荐的做法是:
- 文本 Pipeline 生成文章后,先落盘为 Markdown 文件。
- 另一个脚本读取 Markdown,抽取出适合做成视频的段落。
- 多模态管线根据段落生成配音、画面和字幕。
- 完成后的视频文件也按日期和标题保存到独立目录。
这样拆分的好处是,某一个环节失败时不会影响其他环节。例如画面生成接口超时,已经生成的文章不会丢失,重新执行视频任务即可。
5. 从单次运行到持续运行:定时任务与 Agent 化
5.1 用 APScheduler 或 cron 定时执行
内容生产 Pipeline 的价值在于持续运行,而不是每次手动执行。最简单的方案是用 cron,或者使用 Python 的 APScheduler。
下面是用 APScheduler 实现每天 8 点运行任务的例子:
from pathlib import Path from apscheduler.schedulers.blocking import BlockingScheduler from content_pipeline import ContentPipeline model = "your-model-name" pipeline = ContentPipeline(model=model, prompt_dir=Path("prompts")) def daily_job(): topics = pipeline.generate_topics("AI 工程实践", count=5) for topic in topics: draft = pipeline.generate_draft(topic, "后端开发者") review = pipeline.review_draft(draft) if review.get("score", 0) >= 6: pipeline.save_draft(topic, draft, Path("outputs")) scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job(daily_job, "cron", hour=8, minute=0, id="daily_content") scheduler.start()也可以直接用 crontab:
0 8 * * * cd /opt/ai-blogger-pipeline && /opt/ai-blogger-pipeline/.venv/bin/python run_daily.py >> logs/cron.log 2>&1使用 cron 时要注意,环境变量不会自动读取.env,所以脚本里要保证load_dotenv()会被执行。同时需要显式使用绝对路径,避免 cron 里的工作目录和预期不一致。
5.2 Agent 化:动态流程更灵活,也更难控制
固定 Pipeline 适合“流程明确、步骤稳定”的内容生产。如果希望系统能根据内容主题动态决定“先检索资料再写稿”,或者“先对比多个模型输出再选稿”,就需要引入 Agent。
Agent 的核心是让模型决定调用哪些工具。简单示例如下:
def search_reference(topic: str) -> list[str]: # 调用搜索接口或本地知识库,返回标题和链接 return [] def call_llm(prompt: str) -> str: # 调用大模型 return "result"实际接入 Agent 之前要明确两点:
- Agent 必须设置最大迭代步数和超时时间,否则模型可能陷入循环。
- 所有工具调用都要记录日志,方便追踪模型做了什么决策。
在内容生产场景中,固定 Pipeline 仍应作为默认方案。只有在需要动态检索、多轮改写和复杂决策时,才值得引入 Agent。
5.3 用 SQLite 记录每次运行历史
流水线一旦定时运行,就需要记录每一次的选题、评分、状态和输出路径。推荐使用 SQLite,轻量且不需要额外服务。
表结构可以这样设计:
CREATE TABLE content_run ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, topic TEXT NOT NULL, review_score INTEGER, status TEXT NOT NULL, output_path TEXT );运行完成后插入一条记录,方便后续统计选题命中率、评分变化趋势,以及失败任务的重新执行。
6. 部署到云服务器:从本机跑通到生产可运行
6.1 为什么需要一台云服务器
内容 Pipeline 如果要每天定时运行,依赖本地电脑并不现实。建议把任务部署到一台云服务器(VPS)上,操作系统选择带 systemd 的 Linux 发行版,例如 Ubuntu 22.04 L