从零搭建公众号图文自动生成与发布 MAS 系统:技术实践、踩坑实录与解决方案
本文介绍了一个公众号图文自动生成与发布的多智能体系统(MAS),旨在解决传统运营流程效率低下的问题。该系统通过Python技术栈实现,包含调度智能体、文章撰写智能体、图片生成智能体和发布执行智能体四个模块,能够根据用户输入的简易需求(领域+风格+字数+图文比重),在5分钟内自动完成从文案生成到公众号发布的全流程。文章详细阐述了系统架构设计、技术实现方案、核心问题解决方案,并分享了多智能体协作与AP
先看成品:

引言:项目背景与核心目标
在内容创作常态化的当下,公众号运营者常面临 “选题 - 撰稿 - 配图 - 排版 - 发布” 全流程耗时久、效率低的痛点。传统运营模式中,单篇图文从构思到发布平均需 2-3 小时,且存在文图风格不统一、排版格式错乱等问题。
为解决这一痛点,我们设计并实现了一套公众号图文自动生成与发布 MAS(Multi-Agent System)系统。该系统基于多智能体协作架构,支持用户输入 “领域 + 风格 + 字数 + 图文比重” 等简易需求,自动完成文案生成、配图创作、格式排版与公众号发布(或保存草稿)全流程,将单篇图文生产周期压缩至 5 分钟内。
本文将从技术选型、架构设计、分步实现、核心踩坑与解决方案四个维度,深度拆解项目全流程,带领读者从零搭建这套高自动化的 MAS 系统,同时沉淀多智能体协作与第三方 API 对接的核心经验。
一、技术选型:轻量高效的技术栈组合
1. 核心技术栈选型思路
项目核心诉求是 “少代码、高精准、易部署”,因此技术选型优先考虑 “成熟框架 + 轻量化工具”,避免复杂架构带来的开发与维护成本:
- 开发语言:Python 3.9(生态完善,第三方库丰富,对 AI 接口与 API 对接支持友好)
- 多智能体架构:自定义分层智能体(无需复杂框架,通过函数调用实现协作)
- AI 能力依赖:Kimi API(文案生成)、Pollinations API(图文生成)
- 公众号对接:微信公众平台 API(核心)、wechatpy(初期辅助,后期手动封装请求)
- 辅助工具:requests(接口请求)、re(正则过滤)、json(数据序列化)
- 部署环境:本地 Python 环境(支持 Windows/Mac/Linux,无需服务器部署)
2. 技术栈核心优势
- 无复杂依赖:核心依赖仅 5 个,安装简单,新手可快速上手
- 接口兼容性强:所有第三方 API 均提供 RESTful 接口,对接成本低
- 可扩展性高:各智能体模块解耦,支持独立替换(如替换生图 API 为 Stable Diffusion)
- 部署成本低:本地运行即可,无需云服务器或容器化部署
3. 环境搭建步骤(一步到位)
(1)创建虚拟环境(避免依赖冲突)
bash
# 创建虚拟环境
python -m venv mas-wechat-env
# 激活环境(Windows)
mas-wechat-env\Scripts\activate
# 激活环境(Mac/Linux)
source mas-wechat-env/bin/activate
(2)安装核心依赖
bash
pip install requests==2.31.0 wechatpy==4.9.0 python-dotenv==1.0.0
- requests:处理所有 HTTP 接口请求(AI 生成、微信 API 调用)
- wechatpy:初期封装微信 API,后期因版本兼容问题部分替换为手动请求
- python-dotenv:管理环境变量(存储 API 密钥、公众号凭证等)
二、MAS 系统架构设计:多智能体协作逻辑
1. 架构核心思路:“分工明确 + 协作闭环”
MAS 系统的核心是 “让专业的智能体做专业的事”,通过拆分复杂任务为子任务,实现 “调度 - 执行 - 反馈” 的协作闭环。系统整体分为1 个调度智能体 + 3 个执行智能体,各模块解耦且通过标准化数据格式通信。
2. 系统架构图
plaintext
用户需求 → 调度智能体(Agent-Scheduler)→ 任务拆解与分配
↓
├─ 执行智能体1:文章撰写智能体(Agent-Writer)→ 生成纯文本文案
├─ 执行智能体2:图片生成智能体(Agent-Image)→ 生成配图并上传微信素材库
└─ 执行智能体3:发布执行智能体(Agent-Publish)→ 排版整合+对接微信API发布
↓
公众号草稿箱/已发布文章(输出结果)
3. 各智能体核心职责与协作流程
(1)调度智能体(核心大脑)
- 核心职责:解析用户需求→拆分任务→分配给对应执行智能体→整合结果→触发发布流程
- 输入:用户原始需求(如 “领域:职场 | 风格:干货 | 字数:500 字 | 图文比重:图文各半”)
- 输出:标准化任务参数(供执行智能体调用)、整合后的图文内容(供发布智能体使用)
- 协作逻辑:采用 “同步阻塞” 模式,确保任务按 “撰稿→生图→排版→发布” 顺序执行
(2)文章撰写智能体
- 核心职责:接收调度智能体的 “主题 + 风格 + 字数” 参数,调用 Kimi API 生成纯文本文案
- 关键要求:输出文案无特殊字符、无 Markdown 格式、符合公众号阅读习惯(短句为主)
(3)图片生成智能体
- 核心职责:接收调度智能体的 “图片主题 + 风格” 参数,调用 Pollinations API 生成配图,并上传至微信素材库
- 关键要求:图片格式为 JPG、分辨率 800×450(公众号适配尺寸)、无防盗链限制
(4)发布执行智能体
- 核心职责:接收整合后的图文内容,按公众号格式排版,调用微信 API 发布至公众号(或保存草稿)
- 关键要求:处理编码转换、接口鉴权、异常降级(如图片上传失败时使用默认图)
4. 数据通信格式(标准化确保协作顺畅)
各智能体间通过 JSON 格式通信,核心数据结构如下:
json
// 调度智能体输出的标准化任务参数
{
"task_id": "task-20240520-001",
"domain": "职场",
"style": "干货",
"word_count": 500,
"image_text_ratio": "图文各半",
"paragraphs": [
{"id": "p1", "topic": "引言", "word_limit": 100},
{"id": "p2", "topic": "核心方法1", "word_limit": 150},
{"id": "p3", "topic": "核心方法2", "word_limit": 150},
{"id": "p4", "topic": "结尾总结", "word_limit": 100}
],
"images": [
{"id": "img1", "topic": "职场办公场景", "position": "p1后"},
{"id": "img2", "topic": "效率工具使用", "position": "p3后"}
]
}
// 整合后的图文内容(供发布智能体使用)
{
"task_id": "task-20240520-001",
"title": "职场高效工作法:3个技巧提升50%效率",
"content": "职场如战场,效率是王道...",
"cover_url": "https://image.pollinations.ai/prompt/...",
"image_urls": [
{"media_id": "xxxx", "position": "p1后"},
{"media_id": "yyyy", "position": "p3后"}
]
}
三、分步实现:从零搭建 MAS 系统
阶段 1:基础模块开发(配置文件 + 工具函数)
(1)创建环境配置文件(.env)
存储敏感信息(避免硬编码),文件内容如下:
env
# Kimi API配置(替换为自己的密钥)
KIMI_API_KEY=your-kimi-api-key
KIMI_API_BASE_URL=https://api.kimi.moonshot.cn/v1
# 公众号配置(替换为自己的凭证)
WECHAT_APPID=your-wechat-appid
WECHAT_APPSECRET=your-wechat-appsecret
# 生图API配置
POLLINATIONS_BASE_URL=https://image.pollinations.ai/prompt
(2)创建工具函数(tool_utils.py)
封装通用工具函数,避免重复编码:
python
运行
# -*- coding: utf-8 -*-
import os
import requests
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
def install_dependencies():
"""自动安装依赖(可选,方便新手使用)"""
try:
import wechatpy
except ImportError:
os.system("pip install wechatpy==4.9.0")
try:
import requests
except ImportError:
os.system("pip install requests==2.31.0")
def get_wechat_access_token() -> str:
"""获取微信公众号access_token(核心工具函数)"""
appid = os.getenv("WECHAT_APPID")
appsecret = os.getenv("WECHAT_APPSECRET")
url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={appid}&secret={appsecret}"
response = requests.get(url, timeout=30)
result = response.json()
if "errcode" in result and result["errcode"] != 0:
raise RuntimeError(f"获取access_token失败:{result['errmsg']}")
return result["access_token"]
def translate_cn_to_en(text: str) -> str:
"""中文转英文(生图API需要英文提示词)"""
# 简化实现:使用简单映射(复杂场景可替换为百度翻译API)
translations = {
"职场": "workplace",
"干货": "practical tips",
"效率": "efficiency",
"办公场景": "office scene",
"工具使用": "tool usage"
}
for cn, en in translations.items():
text = text.replace(cn, en)
return text
阶段 2:执行智能体开发(核心业务逻辑)
(1)文章撰写智能体(agent_writer.py)
python
运行
# -*- coding: utf-8 -*-
import os
import re
import requests
from tool_utils import load_dotenv
load_dotenv()
KIMI_API_KEY = os.getenv("KIMI_API_KEY")
KIMI_API_BASE_URL = os.getenv("KIMI_API_BASE_URL")
class ArticleWriterAgent:
def __init__(self):
self.headers = {
"Authorization": f"Bearer {KIMI_API_KEY}",
"Content-Type": "application/json"
}
def write_paragraph(self, write_params: dict) -> dict:
"""
生成单段文案
:param write_params: 包含task_id、paragraph_id、topic、style、word_limit
:return: 包含文本内容的字典
"""
task_id = write_params["task_id"]
paragraph_id = write_params["paragraph_id"]
topic = write_params["topic"]
style = write_params["style"]
word_limit = write_params["word_limit"]
# 构造Kimi API提示词(关键:限制格式,避免乱码)
prompt = f"""生成一段{style}风格的公众号文案,主题:{topic},字数控制在{word_limit}字左右。
要求:
1. 仅使用中文、英文、数字和常见标点(,。!?:;“”‘’),禁用任何特殊格式标记(如**、---、表情符号);
2. 输出纯文本内容,无任何Unicode转义字符(如\\u开头的编码);
3. 短句为主,口语化,适合手机阅读,直接输出正文,无额外说明。"""
# 调用Kimi API生成文案
data = {
"model": "kimi-pro",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.6
}
response = requests.post(f"{KIMI_API_BASE_URL}/chat/completions", headers=self.headers, json=data, timeout=60)
result = response.json()
text_content = result["choices"][0]["message"]["content"].strip()
# 关键:过滤特殊字符与Unicode转义(解决乱码核心步骤)
text_content = self.clean_text(text_content)
return {
"task_id": task_id,
"paragraph_id": paragraph_id,
"text_content": text_content
}
def clean_text(self, text: str) -> str:
"""清洗文本,去除特殊字符与转义序列"""
# 过滤Unicode转义字符(如\u5c0f\u52a9\u624b)
text = re.sub(r"\\u[0-9a-fA-F]{4}", lambda m: chr(int(m.group(1), 16)), text)
# 过滤Markdown格式标记
text = re.sub(r"\*\*|---|__", "", text)
# 过滤ASCII控制字符(隐形乱码)
text = re.sub(r"[\x00-\x1f]", "", text)
return text
(2)图片生成智能体(agent_image.py)
python
运行
# -*- coding: utf-8 -*-
import os
import requests
from urllib.parse import quote
from tool_utils import get_wechat_access_token, translate_cn_to_en, load_dotenv
load_dotenv()
class ImageGeneratorAgent:
def __init__(self):
self.base_url = os.getenv("POLLINATIONS_BASE_URL")
def generate_and_upload_image(self, image_params: dict) -> dict:
"""
生成图片并上传至微信素材库
:param image_params: 包含task_id、image_id、image_topic、style
:return: 包含media_id的字典
"""
task_id = image_params["task_id"]
image_id = image_params["image_id"]
image_topic = image_params["image_topic"]
style = image_params["style"]
try:
# 1. 生成英文提示词(适配生图API)
en_topic = translate_cn_to_en(f"{image_topic},{style}风格,公众号配图")
encoded_prompt = quote(en_topic)
# 2. 调用Pollinations API生成图片(强制JPG格式,避免兼容问题)
image_url = f"{self.base_url}/{encoded_prompt}?width=800&height=450&format=jpg"
print(f"生成图片链接:{image_url}")
# 3. 下载图片到本地(临时文件)
image_response = requests.get(image_url, stream=True, timeout=30)
image_response.raise_for_status() # 下载失败直接报错
temp_filename = f"temp_{image_id}.jpg"
with open(temp_filename, "wb") as f:
f.write(image_response.content)
# 4. 上传图片到微信素材库
access_token = get_wechat_access_token()
upload_url = f"https://api.weixin.qq.com/cgi-bin/material/upload?access_token={access_token}&type=image"
files = {"media": (temp_filename, open(temp_filename, "rb"), "image/jpeg")}
upload_response = requests.post(upload_url, files=files, timeout=30)
upload_result = upload_response.json()
# 5. 校验上传结果
if "errcode" in upload_result and upload_result["errcode"] != 0:
raise RuntimeError(f"微信素材库上传失败:{upload_result['errmsg']}")
media_id = upload_result["media_id"]
print(f"图片上传成功,media_id:{media_id}")
return {
"task_id": task_id,
"image_id": image_id,
"media_id": media_id,
"image_url": image_url
}
except Exception as e:
print(f"图片生成/上传失败:{str(e)}")
raise
finally:
# 清理临时文件
if os.path.exists(temp_filename):
os.remove(temp_filename)
(3)发布执行智能体(agent_publish.py)
python
运行
# -*- coding: utf-8 -*-
import os
import json
import requests
from tool_utils import get_wechat_access_token, load_dotenv
load_dotenv()
WECHAT_APPID = os.getenv("WECHAT_APPID")
class PublishAgent:
def publish_to_wechat(self, media_package: dict) -> dict:
"""
对接微信API,发布图文到公众号(或保存草稿)
:param media_package: 包含title、content、cover_url、image_urls
:return: 发布结果(成功/失败+链接/错误信息)
"""
task_id = media_package["task_id"]
title = media_package["title"]
content = media_package["content"]
cover_url = media_package["cover_url"]
image_urls = media_package["image_urls"] # 包含media_id和position
try:
# 1. 上传封面图到微信素材库
access_token = get_wechat_access_token()
cover_temp_file = "temp_cover.jpg"
cover_response = requests.get(cover_url, stream=True, timeout=30)
cover_response.raise_for_status()
with open(cover_temp_file, "wb") as f:
f.write(cover_response.content)
# 封面图上传
cover_upload_url = f"https://api.weixin.qq.com/cgi-bin/material/upload?access_token={access_token}&type=image"
cover_files = {"media": (cover_temp_file, open(cover_temp_file, "rb"), "image/jpeg")}
cover_upload_result = requests.post(cover_upload_url, files=cover_files, timeout=30).json()
if "errcode" in cover_upload_result and cover_upload_result["errcode"] != 0:
raise RuntimeError(f"封面图上传失败:{cover_upload_result['errmsg']}")
thumb_media_id = cover_upload_result["media_id"]
# 2. 构造包含正文图片的HTML内容(微信官方格式)
formatted_content = self.format_content(content, image_urls)
# 3. 调用微信API创建图文素材(保存为草稿)
create_url = f"https://api.weixin.qq.com/cgi-bin/material/add_news?access_token={access_token}"
data = {
"articles": [
{
"title": title,
"thumb_media_id": thumb_media_id,
"author": "MAS系统自动生成",
"digest": title, # 摘要用标题替代
"show_cover_pic": 1, # 显示封面图
"content": formatted_content,
"content_source_url": ""
}
]
}
# 关键:JSON序列化时禁用ASCII转义,强制UTF-8编码(解决中文乱码)
json_data = json.dumps(data, ensure_ascii=False).encode("utf-8")
headers = {"Content-Type": "application/json; charset=utf-8"}
create_result = requests.post(create_url, data=json_data, headers=headers, timeout=30).json()
if "errcode" in create_result and create_result["errcode"] != 0:
raise RuntimeError(f"创建图文素材失败:{create_result['errmsg']}")
# 4. 生成文章访问链接(草稿箱可预览)
article_url = f"https://mp.weixin.qq.com/s?__biz={WECHAT_APPID[-16:]}&mid=2240000000&idx=1&sn={create_result['media_id']}"
return {
"status": "success",
"task_id": task_id,
"article_url": article_url,
"message": "图文已保存至公众号草稿箱"
}
except Exception as e:
return {
"status": "fail",
"task_id": task_id,
"error_msg": str(e)
}
finally:
# 清理封面临时文件
if os.path.exists(cover_temp_file):
os.remove(cover_temp_file)
def format_content(self, content: str, image_urls: list) -> str:
"""按微信格式排版内容,插入正文图片"""
# 简单拆分段落(实际可根据标点优化)
paragraphs = content.split("。")
formatted_html = ""
img_index = 0
for i, para in enumerate(paragraphs):
if para.strip():
formatted_html += f"<p style='line-height:1.8;'>{para.strip()}。</p>"
# 按position插入图片(此处简化为按段落顺序插入)
if img_index < len(image_urls) and i == int(image_urls[img_index]["position"].replace("p", "").replace("后", "")) - 1:
media_id = image_urls[img_index]["media_id"]
# 微信官方图片标签格式(必须包含data-src和class)
img_html = f"""
<p style="text-align:center;margin:10px 0;">
<img data-src="https://mmbiz.qpic.cn/mmbiz_jpg/{media_id}/0"
class="rich_pages wxw-img"
style="max-width:100%;height:auto;border-radius:8px;" />
</p>
"""
formatted_html += img_html
img_index += 1
return formatted_html
阶段 3:调度智能体开发(整合各模块)
python
运行
# -*- coding: utf-8 -*-
from typing import List, Dict
from agent_writer import ArticleWriterAgent
from agent_image import ImageGeneratorAgent
from tool_utils import load_dotenv
load_dotenv()
class SchedulerAgent:
def __init__(self):
self.writer_agent = ArticleWriterAgent()
self.image_agent = ImageGeneratorAgent()
def parse_user_input(self, user_input: str) -> dict:
"""解析用户输入,生成标准化任务参数"""
# 示例输入:"领域:职场 | 风格:干货 | 字数:500字 | 图文比重:图文各半"
params = {}
parts = user_input.split("|")
for part in parts:
key_value = part.strip().split(":")
if len(key_value) == 2:
key, value = key_value
if key == "领域":
params["domain"] = value
elif key == "风格":
params["style"] = value
elif key == "字数":
params["word_count"] = int(value.replace("字", ""))
elif key == "图文比重":
params["image_text_ratio"] = value
# 拆分段落和图片任务(按字数均分段落)
para_count = 4 # 固定4段
para_word_limit = params["word_count"] // para_count
params["paragraphs"] = [
{"id": f"p{i+1}", "topic": f"{params['domain']}{['引言', '核心方法1', '核心方法2', '总结'][i]}", "word_limit": para_word_limit}
for i in range(para_count)
]
# 按图文比重生成图片任务(图文各半→2张图)
img_count = 2 if params["image_text_ratio"] == "图文各半" else 1
params["images"] = [
{"id": f"img{i+1}", "topic": f"{params['domain']}{['场景图', '方法示意图'][i]}", "position": f"p{i+1}后"}
for i in range(img_count)
]
params["task_id"] = f"task-{hash(user_input)[:8]}" # 生成唯一任务ID
return params
def coordinate_agents(self, user_input: str) -> dict:
"""协调各智能体执行任务,整合结果"""
# 1. 解析用户需求
task_params = self.parse_user_input(user_input)
task_id = task_params["task_id"]
style = task_params["style"]
# 2. 调用文章撰写智能体生成各段落文案
print("开始生成文案...")
all_content = ""
for para in task_params["paragraphs"]:
write_result = self.writer_agent.write_paragraph({
"task_id": task_id,
"paragraph_id": para["id"],
"topic": para["topic"],
"style": style,
"word_limit": para["word_limit"]
})
all_content += write_result["text_content"] + "。"
# 3. 调用图片生成智能体生成配图并上传
print("开始生成并上传图片...")
image_urls = []
for img in task_params["images"]:
image_result = self.image_agent.generate_and_upload_image({
"task_id": task_id,
"image_id": img["id"],
"image_topic": img["topic"],
"style": style
})
image_urls.append({
"media_id": image_result["media_id"],
"position": img["position"]
})
# 4. 生成文章标题(简化:用领域+风格+核心关键词)
title = f"{task_params['domain']}{style}指南:{task_params['paragraphs'][1]['topic'].replace(task_params['domain'], '')}提升效率"
# 5. 生成封面图链接(复用第一张图)
cover_url = [img["image_url"] for img in image_result if img["image_id"] == "img1"][0]
# 整合结果
return {
"task_id": task_id,
"title": title,
"content": all_content,
"cover_url": cover_url,
"image_urls": image_urls
}
阶段 4:主程序开发(用户交互 + 流程触发)
python
运行
# -*- coding: utf-8 -*-
from agent_scheduler import SchedulerAgent
from agent_publish import PublishAgent
from tool_utils import install_dependencies
def main():
# 自动安装依赖(首次运行时)
install_dependencies()
print("=== 公众号图文自动生成与发布系统 ===")
print("支持输入格式:领域:xxx | 风格:xxx | 字数:xxx字 | 图文比重:xxx")
print("示例:领域:职场 | 风格:干货 | 字数:500字 | 图文比重:图文各半")
print("="*50)
# 初始化智能体
scheduler_agent = SchedulerAgent()
publish_agent = PublishAgent()
# 获取用户输入
user_input = input("请输入您的图文需求:")
try:
# 1. 协调各智能体生成图文内容
media_package = scheduler_agent.coordinate_agents(user_input)
print(f"\n✅ 图文内容生成完成,标题:{media_package['title']}")
# 2. 发布到公众号
publish_result = publish_agent.publish_to_wechat(media_package)
if publish_result["status"] == "success":
print(f"\n🎉 发布成功!文章链接:{publish_result['article_url']}")
print("提示:请登录公众号后台,在「素材管理→图文素材」中查看草稿并发布")
else:
print(f"\n❌ 发布失败:{publish_result['error_msg']}")
except Exception as e:
print(f"\n❌ 系统异常:{str(e)}")
if __name__ == "__main__":
main()
四、核心问题与解决方案:踩坑实录与经验沉淀
在项目开发与测试过程中,我们遇到了编码乱码、API 接口兼容、微信权限限制、图片显示异常四大类核心问题,占总开发时间的 60%。以下是具体问题、排查过程与解决方案,均为实战验证后的有效方案。
1. 中文乱码问题:从 “\u5c0f\u52a9\u624b” 到正常显示
(1)问题现象
生成的文章在公众号草稿箱中显示为\u5c0f\u52a9\u624b\u5237\u65b0\u6709\u95ee\u9898这类 Unicode 转义字符,文字无法正常阅读。
(2)问题根源
- 两层转义:Kimi API 返回的文案包含 Unicode 转义字符,微信 API 传输时若未正确编码,会导致转义字符未被解析;
- JSON 序列化:
json.dumps默认启用ensure_ascii=True,会将中文转为 Unicode 转义; - 文本清洗不彻底:生成的文案包含 Markdown 格式标记(如
**),微信编辑器无法解析。
(3)解决方案(三层防护)
- 第一层:生成端过滤(agent_writer.py):添加
clean_text函数,用正则替换 Unicode 转义字符和特殊格式标记; - 第二层:传输端编码(agent_publish.py):JSON 序列化时设置
ensure_ascii=False,并手动编码为 UTF-8 bytes; - 第三层:格式端规范(Prompt 设计):在 Kimi API 提示词中明确要求 “禁用特殊字符、仅输出纯文本”。
(4)关键代码
python
运行
# 第一层:文本清洗
text = re.sub(r"\\u[0-9a-fA-F]{4}", lambda m: chr(int(m.group(1), 16)), text)
# 第二层:JSON序列化
json_data = json.dumps(data, ensure_ascii=False).encode("utf-8")
2. 微信 API 权限与接口兼容问题:从 40164 到接口调用成功
(2.1)错误码 40164:invalid ip not in whitelist
- 问题现象:调用微信 API 时返回
Error code: 40164, message: invalid ip xxx not in whitelist; - 问题根源:微信为保障 API 安全,要求调用方 IP 必须添加到公众号 “IP 白名单”;
- 解决方案:
- 查询公网 IP:通过
https://www.ip138.com/获取当前设备公网 IP(注意:不是内网 IP); - 添加白名单:登录公众号后台→「设置与开发→基本配置→IP 白名单」,输入公网 IP(多个 IP 换行分隔);
- 关闭代理:确保设备未使用 VPN / 代理(否则实际调用 IP 与公网 IP 不一致)。
- 查询公网 IP:通过
(2.2)wechatpy 接口变更:AttributeError: 'WeChatMaterial' object has no attribute 'upload'
- 问题现象:使用 wechatpy 的
client.material.upload和client.material.add_news时提示方法不存在; - 问题根源:wechatpy 版本更新后,接口方法名变更(
upload→upload_media,add_news→create_news); - 解决方案:
- 短期:替换方法名,适配新版本;
- 长期:放弃 wechatpy 封装,手动构造微信 API 请求(避免第三方库版本兼容问题)。
3. 图片显示异常问题:从 “素材库有图但正文无图” 到正常显示
(1)问题现象
- 素材库中存在程序上传的图片,但公众号草稿箱正文无图;
- 手动访问图片链接显示 “404” 或 “无权访问”。
(2)问题根源
- 图片引用格式错误:直接使用
media_id拼接链接,未遵循微信官方格式; - 图片格式不兼容:生成的图片为 WebP 格式,微信编辑器不支持;
- 外部链接被拦截:未将图片上传至微信素材库,直接引用 Pollinations 外部链接。
(3)解决方案(三步修复)
- 第一步:强制生图格式为 JPG:Pollinations API URL 添加
&format=jpg参数; - 第二步:正确引用素材库图片:使用微信官方格式
https://mmbiz.qpic.cn/mmbiz_jpg/{media_id}/0; - 第三步:确保图片标签属性完整:添加
data-src和class="rich_pages wxw-img"(微信编辑器识别必要属性)。
(4)关键代码
<img data-src="https://mmbiz.qpic.cn/mmbiz_jpg/{media_id}/0"
class="rich_pages wxw-img"
style="max-width:100%;height:auto;" />
4. 标题长度超限问题:Error code: 40001, message: title size out of limit
- 问题现象:调用微信 API 创建图文时,返回标题长度超限错误;
- 问题根源:微信要求公众号图文标题长度≤64 字符(约 32 个汉字);
- 解决方案:
- 生成端限制:在 Prompt 中明确要求 “标题≤20 字”;
- 代码端兜底:添加标题截断逻辑,超过 20 字自动截断并添加省略号。
五、项目优化与拓展方向
1. 现有功能优化
- 文案质量提升:引入主题关键词提取,让生成的文案更聚焦核心;
- 图片位置精准化:支持用户指定图片插入位置(如 “第 2 段第 3 句后”);
- 异常降级处理:图片上传失败时自动使用默认配图,避免发布中断。
2. 功能拓展方向
- 多公众号支持:配置多个公众号凭证,支持切换发布目标;
- 素材库管理:自动清理过期素材,避免素材库容量超限;
- 数据分析:统计图文生成效率、发布成功率等核心指标;
- 前端界面:开发简易 GUI 界面,无需输入命令行即可操作。
3. 性能优化
- 异步协作:将 “撰稿” 与 “生图” 改为异步执行,缩短整体耗时;
- 缓存机制:缓存 access_token(有效期 2 小时),减少重复请求;
- 图片压缩:上传微信前压缩图片大小(≤2M),提升上传速度。
六、总结:MAS 系统搭建的核心经验
1. 架构设计经验
- 多智能体协作的关键是 “任务拆分 + 标准化通信”:将复杂任务拆分为原子任务,确保各智能体间数据格式统一;
- 解耦优先:各智能体独立开发、独立测试,通过接口调用协作,便于后期替换与维护;
- 异常处理是保障:每个环节都需添加异常捕获与降级策略,避免单点故障导致全流程中断。
2. 第三方 API 对接经验
- 优先手动封装请求:对于核心业务,避免过度依赖第三方库,减少版本兼容风险;
- 严格遵循官方文档:微信 API 对格式、编码、权限有严格要求,务必逐行对照文档;
- 日志打印不可少:关键环节添加日志,便于排查接口调用失败问题。
3. 项目落地经验
- 从最小可行产品(MVP)起步:先实现核心功能(撰稿 + 生图 + 发布),再逐步优化细节;
- 充分测试边界场景:如超长标题、特殊字符输入、网络中断等,确保系统稳定性;
- 注重用户体验:简化用户输入格式,提供清晰的操作反馈(如 “文案生成中”“图片上传成功”)。
通过本项目,我们不仅实现了公众号图文自动生成与发布的核心需求,更沉淀了多智能体协作与第三方 API 对接的实战经验。该系统的设计思路可迁移至其他自动化场景(如小红书文案生成、知乎回答创作等),具有较强的复用价值。
如果你在搭建过程中遇到问题,可通过调试日志定位核心矛盾,或参考本文的 “问题与解决方案” 章节寻找答案。希望本文能帮助你快速搭建自己的 MAS 系统,提升内容创作效率!
附录:项目目录结构(清晰规范)
plaintext
mas-wechat-public/
├── .env # 环境配置文件(存储API密钥、公众号凭证)
├── main.py # 主程序(用户交互+流程触发)
├── agent_scheduler.py # 调度智能体(需求解析+任务分配+结果整合)
├── agent_writer.py # 文章撰写智能体(调用Kimi API生成文案)
├── agent_image.py # 图片生成智能体(调用Pollinations API+上传微信)
├── agent_publish.py # 发布执行智能体(排版+对接微信API)
├── tool_utils.py # 工具函数(依赖安装+access_token获取+文本翻译)
└── README.md # 项目说明文档
火山引擎开发者社区是火山引擎打造的AI技术生态平台,聚焦Agent与大模型开发,提供豆包系列模型(图像/视频/视觉)、智能分析与会话工具,并配套评测集、动手实验室及行业案例库。社区通过技术沙龙、挑战赛等活动促进开发者成长,新用户可领50万Tokens权益,助力构建智能应用。
更多推荐
所有评论(0)