微信公众号运营的核心痛点是持续产出高质量内容。大模型的出现让"AI辅助写作"成为可能，但从"辅助"到"全自动"，中间还有大量的工程问题需要解决。
这篇文章拆解一个完整的技术方案：从微信发一条指令，AI自动生成内容并发布到公众号，支持多账号矩阵和定时发布。
一、整体架构
┌──────────────────────────────────────────────────┐
│ 控制层 │
│ 微信消息 → 指令解析 → 任务队列 │
└────────────┬─────────────────────────────────────┘
│
┌────────────▼─────────────────────────────────────┐
│ 生成层 │
│ 大模型API → 内容生成 → 格式化 → 质量检查 │
└────────────┬─────────────────────────────────────┘
│
┌────────────▼─────────────────────────────────────┐
│ 发布层 │
│ 素材上传 → 图文创建 → 定时/即时发布 │
│ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │账号1│ │账号2│ │账号3│ ... 多账号矩阵 │
│ └─────┘ └─────┘ └─────┘ │
└──────────────────────────────────────────────────┘二、微信公众号API核心接口
微信公众号的自动化发布主要依赖以下几个API：
1. 获取Access Token
所有API调用的前提。Access Token有效期2小时，需要做缓存和自动刷新。
import requests
import time
import json
class WechatAPI:
def __init__(self, appid, appsecret):
self.appid = appid
self.appsecret = appsecret
self.token_cache = {"token": None, "expires": 0}
def get_access_token(self):
"""获取access_token，带缓存"""
now = time.time()
if self.token_cache["token"] and now < self.token_cache["expires"]:
return self.token_cache["token"]
url = "https://api.weixin.qq.com/cgi-bin/token"
params = {
"grant_type": "client_credential",
"appid": self.appid,
"secret": self.appsecret
}
resp = requests.get(url, params=params).json()
if "access_token" in resp:
self.token_cache = {
"token": resp["access_token"],
"expires": now + resp["expires_in"] - 300 # 提前5分钟刷新
}
return resp["access_token"]
raise Exception(f"获取token失败: {resp}")2. 上传素材
def upload_image(self, image_path):
"""上传图片素材，返回media_id"""
token = self.get_access_token()
url = f"https://api.weixin.qq.com/cgi-bin/media/uploadimg?access_token={token}"
with open(image_path, "rb") as f:
files = {"media": f}
resp = requests.post(url, files=files).json()
return resp.get("url")
def upload_thumb(self, image_path):
"""上传封面图（永久素材），返回thumb_media_id"""
token = self.get_access_token()
url = f"https://api.weixin.qq.com/cgi-bin/material/add_material?access_token={token}&type=thumb"
with open(image_path, "rb") as f:
files = {"media": f}
resp = requests.post(url, files=files).json()
return resp.get("media_id")3. 创建图文草稿并发布
def create_draft(self, title, content, thumb_media_id, digest=""):
"""创建图文草稿"""
token = self.get_access_token()
url = f"https://api.weixin.qq.com/cgi-bin/draft/add?access_token={token}"
data = {
"articles": [{
"title": title,
"author": "南有归心",
"digest": digest,
"content": content,
"thumb_media_id": thumb_media_id,
"content_source_url": "",
"need_open_comment": 1
}]
}
resp = requests.post(url, json=data).json()
return resp.get("media_id") # 草稿的media_id
def publish(self, draft_media_id):
"""发布草稿（群发）"""
token = self.get_access_token()
url = f"https://api.weixin.qq.com/cgi-bin/freepublish/submit?access_token={token}"
data = {"media_id": draft_media_id}
resp = requests.post(url, json=data).json()
return resp微信公众号的群发接口有频率限制：认证服务号每月4次群发，订阅号每天1次。超过限制调用会失败。系统需要做好频率控制和日志记录。
三、AI内容生成
Prompt设计
好的Prompt是内容质量的关键。需要控制风格、长度、结构：
def generate_article(topic, style="专业", word_count=1500):
"""调用大模型生成公众号文章"""
prompt = f"""请为微信公众号撰写一篇文章。
## 要求
- 主题：{topic}
- 风格：{style}（可选：专业/轻松活泼/深度分析/科普）
- 字数：约{word_count}字
- 结构：标题 + 引言 + 3-4个小节 + 总结
- 每个小节有小标题
- 语言自然流畅，避免AI味重的套话
- 适当使用数据和案例增加说服力
## 格式要求
- 标题单独一行，不加任何标记
- 小标题用【】包裹
- 段落之间空一行
- 不要使用Markdown语法（公众号不支持）
## 禁止
- 不要出现"作为AI"、"我无法"等自我暴露的表述
- 不要使用"首先、其次、最后"等程式化结构词
- 不要在结尾写"如果觉得有用请点赞关注"等套路"""
# 调用DeepSeek API
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://api.deepseek.com"
)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": "你是一个资深的微信公众号内容作者。"},
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=3000
)
return response.choices[0].message.content内容质量控制
AI生成的内容不能直接发，需要自动检查：
def quality_check(content):
"""内容质量检查"""
issues = []
# 长度检查
if len(content) < 500:
issues.append("内容过短，不足500字")
if len(content) > 5000:
issues.append("内容过长，超过5000字")
# 敏感词检查（简单示例，生产环境建议用专业敏感词库）
sensitive_words = ["作为AI", "我是一个语言模型", "作为人工智能"]
for word in sensitive_words:
if word in content:
issues.append(f"包含敏感表述：{word}")
# 结构检查
if content.count("【") < 2:
issues.append("文章结构不完整，小节过少")
return {"pass": len(issues) == 0, "issues": issues}四、多账号矩阵管理
管理多个公众号的核心是配置隔离和内容差异化。
# accounts.json - 账号配置
{
"accounts": [
{
"name": "主号-科技前沿",
"appid": "wx_xxx_1",
"appsecret": "secret_1",
"style": "深度分析",
"topics": ["AI技术", "大模型", "科技趋势"],
"publish_time": "10:00"
},
{
"name": "子号-轻松科普",
"appid": "wx_xxx_2",
"appsecret": "secret_2",
"style": "轻松活泼",
"topics": ["AI入门", "工具推荐", "效率提升"],
"publish_time": "14:30"
},
{
"name": "子号-行业观察",
"appid": "wx_xxx_3",
"appsecret": "secret_3",
"style": "专业",
"topics": ["行业动态", "商业分析", "技术落地"],
"publish_time": "19:00"
}
]
}差异化内容生成
同一个主题，不同账号用不同风格改写，避免内容雷同：
def generate_matrix_content(topic, accounts):
"""为多个账号生成差异化内容"""
results = []
for account in accounts:
content = generate_article(
topic=topic,
style=account["style"],
word_count=1200 if account["style"] == "轻松活泼" else 1800
)
results.append({
"account": account["name"],
"content": content,
"publish_time": account["publish_time"]
})
return results五、定时发布调度
使用Python的schedule库或系统crontab实现定时发布：
import schedule
import threading
def scheduled_publish(account_config, content):
"""定时发布任务"""
api = WechatAPI(account_config["appid"], account_config["appsecret"])
# 1. 上传封面图
thumb_id = api.upload_thumb("default_cover.jpg")
# 2. 格式化内容为HTML（公众号使用HTML格式）
html_content = format_to_wechat_html(content)
# 3. 创建草稿
title = content.split("\n")[0] # 第一行作为标题
draft_id = api.create_draft(title, html_content, thumb_id)
# 4. 发布
result = api.publish(draft_id)
log(f"[{account_config['name']}] 发布结果: {result}")
# 注册定时任务
for account in config["accounts"]:
schedule.every().day.at(account["publish_time"]).do(
scheduled_publish, account, generated_content[account["name"]]
)六、微信指令控制
通过微信服务号的消息推送接口，实现"发一条消息就触发发布"：
# Flask接收微信消息推送
from flask import Flask, request
import xml.etree.ElementTree as ET
app = Flask(__name__)
@app.route("/wechat", methods=["POST"])
def wechat_webhook():
xml_data = request.data
root = ET.fromstring(xml_data)
msg_type = root.find("MsgType").text
content = root.find("Content").text if msg_type == "text" else ""
from_user = root.find("FromUserName").text
# 验证是否为管理员
if from_user not in ADMIN_OPENIDS:
return reply_text(root, "无权限操作")
# 解析指令
if content.startswith("发布"):
topic = content.replace("发布", "").strip()
# 异步执行，避免微信5秒超时
threading.Thread(target=handle_publish, args=(topic,)).start()
return reply_text(root, f"收到，正在为所有账号生成关于「{topic}」的内容...")
return reply_text(root, "指令格式：发布 [主题]")微信服务号的消息推送有5秒响应超时。内容生成和发布是耗时操作，必须放到异步线程/任务队列中执行，先立即回复"收到"，完成后再主动推送结果。
七、注意事项与坑
- Access Token并发问题：多个进程同时刷新token会导致旧token失效。建议用Redis做分布式锁，或单独一个定时任务刷新。
- 图文内容HTML格式：公众号编辑器使用HTML，不是Markdown。需要做格式转换，且不支持所有HTML标签。
- 图片必须用微信域名：图文中引用的图片必须先通过uploadimg接口上传到微信服务器，不能直接引用外部URL。
- 草稿和发布是两个步骤：先创建草稿（draft/add），再提交发布（freepublish/submit），发布是异步的。
- 群发次数限制：超过限制后当月无法再群发。务必在代码中做次数统计和预警。