邹平融媒客户端
193.88MB · 2025-11-08
原文
wemark平台(wemark.xinghegongyu.com)是专门免费为写作者打造的工作利器，它有多种markdown排版主题，而且支持图文编辑，只需要截图粘贴即可，同时也支持对话大模型润色文章，还有更多功能期待大家去发现，发过一篇介绍点击[这里](https://mp.*wei**xin.…
有了wemark平台，我们就可以搭建一个本地AI工作流，每天帮我们写一篇微信公众号文章发布到wemark平台，wemark平台支持实时预览并且修改，还可以支持一键复制内容到公众号，对于日日更的小伙伴非常方便，你只需要说出你要创作的主题，只需要一句话就可以创建你要的精美图文。
好了，接下来带大家开启这段愉快的旅程吧！
项目目录：
wechat_article_workflow/
├── main.py # 主程序入口
├── workflow.py # LangGraph工作流定义
├── models.py # 多模态模型接口
├── blog_publisher.py # 博客发布器
├── config.py # 配置管理
├── requirements.txt # 依赖包
├── .env.example # 环境变量示例
└── README.md # 说明文档
使用示例：
from workflow import WechatArticleWorkflow
# 创建工作流实例
workflow = WechatArticleWorkflow()
# 运行工作流
result = workflow.run("人工智能在教育领域的应用")
# 获取生成的Markdown内容
markdown_content = result["final_markdown"]
print(markdown_content)
工作流核心代码：
"""
LangGraph工作流实现
"""
import json
import uuid
import time
from typing import Dict, Any, List
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from typing_extensions import Annotated, TypedDict
from models import TextGenerator, ImageGenerator, ArticleSection
from blog_publisher import BlogPublisher
class WorkflowState(TypedDict):
"""工作流状态"""
topic: str
outline: Dict[str, Any]
sections: List[ArticleSection]
final_markdown: str
images_generated: bool
published: bool
error: str
class WechatArticleWorkflow:
"""微信公众号文章生成工作流"""
def __init__(self):
self.text_generator = TextGenerator()
self.image_generator = ImageGenerator()
self.blog_publisher = BlogPublisher()
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
"""构建LangGraph工作流"""
workflow = StateGraph(WorkflowState)
# 添加节点
workflow.add_node("generate_outline", self._generate_outline)
workflow.add_node("generate_content", self._generate_content)
workflow.add_node("generate_images", self._generate_images)
workflow.add_node("compile_markdown", self._compile_markdown)
workflow.add_node("publish_article", self._publish_article)
# 设置边
workflow.set_entry_point("generate_outline")
workflow.add_edge("generate_outline", "generate_content")
workflow.add_edge("generate_content", "generate_images")
workflow.add_edge("generate_images", "compile_markdown")
workflow.add_edge("compile_markdown", "publish_article")
workflow.add_edge("publish_article", END)
return workflow.compile()
def _generate_outline(self, state: WorkflowState) -> WorkflowState:
"""生成文章大纲"""
print(f"? 开始为主题 '{state['topic']}' 生成大纲...")
try:
outline_text = self.text_generator.generate_article_outline(state["topic"])
# 尝试解析JSON，如果失败则使用文本
try:
outline = json.loads(outline_text)
except:
# 如果不是JSON格式，创建基础结构
outline = {
"title": f"关于{state['topic']}的深度思考",
"sections": [
{
"title": "引言",
"content_points": ["背景介绍", "问题提出"],
"image_description": f"{state['topic']}相关的概念图"
},
{
"title": "核心内容",
"content_points": ["主要观点", "详细分析"],
"image_description": f"{state['topic']}的实践应用场景"
},
{
"title": "总结展望",
"content_points": ["要点总结", "未来展望"],
"image_description": f"{state['topic']}的未来发展趋势"
}
],
"conclusion": "总结全文要点，引导读者思考"
}
state["outline"] = outline
print(f"✅ 大纲生成完成: {outline['title']}")
except Exception as e:
state["error"] = f"大纲生成失败: {str(e)}"
print(f"❌ {state['error']}")
return state
def _generate_content(self, state: WorkflowState) -> WorkflowState:
"""生成文章内容"""
print("? 开始生成文章内容...")
sections = []
outline = state.get("outline", {})
try:
for section_data in outline.get("sections", []):
print(f" 生成段落: {section_data['title']}")
time.sleep(12);
content = self.text_generator.generate_section_content(
section_data["title"],
section_data.get("content_points", []),
state["topic"]
)
section = ArticleSection(
title=section_data["title"],
content=content,
image_prompt=section_data.get("image_description", "")
)
sections.append(section)
state["sections"] = sections
print(f"✅ 内容生成完成，共{len(sections)}个段落")
except Exception as e:
state["error"] = f"内容生成失败: {str(e)}"
print(f"❌ {state['error']}")
return state
def _generate_images(self, state: WorkflowState) -> WorkflowState:
"""生成配图"""
print("? 开始生成配图...")
try:
sections = state.get("sections", [])
for i, section in enumerate(sections):
if section.image_prompt:
print(f" 生成图片 {i+1}: {section.image_prompt}")
image_url = self.image_generator.generate_image(section.image_prompt)
if image_url:
section.image_url = image_url
print(f" ✅ 图片生成成功")
else:
print(f" ⚠️ 图片生成失败，将跳过")
state["images_generated"] = True
print("✅ 配图生成完成")
except Exception as e:
state["error"] = f"配图生成失败: {str(e)}"
print(f"❌ {state['error']}")
state["images_generated"] = False
return state
def _compile_markdown(self, state: WorkflowState) -> WorkflowState:
"""编译Markdown格式"""
print("? 开始编译Markdown...")
try:
outline = state.get("outline", {})
sections = state.get("sections", [])
# 构建Markdown内容
markdown_parts = []
# 标题
title = outline.get("title", f"关于{state['topic']}的思考")
markdown_parts.append(f"# {title}n")
# 各个段落
for section in sections:
markdown_parts.append(f"## {section.title}n")
# 添加图片（如果有）
if section.image_url:
markdown_parts.append(f"n")
# 添加内容
markdown_parts.append(f"{section.content}n")
# 结论
if outline.get("conclusion"):
markdown_parts.append(f"## 总结nn{outline['conclusion']}n")
# 互动引导
markdown_parts.append("n---nn? 你对这个话题有什么看法？欢迎在评论区分享你的想法！n")
final_markdown = "n".join(markdown_parts)
# 确保最终内容不包含markdown代码块标记
final_markdown = self._clean_final_markdown(final_markdown)
state["final_markdown"] = final_markdown
print("✅ Markdown编译完成")
except Exception as e:
state["error"] = f"Markdown编译失败: {str(e)}"
print(f"❌ {state['error']}")
return state
def _publish_article(self, state: WorkflowState) -> WorkflowState:
"""发布文章到博客"""
print("? 开始发布文章...")
try:
outline = state.get("outline", {})
title = outline.get("title", f"关于{state['topic']}的思考")
content = state.get("final_markdown", "")
if content:
success = self.blog_publisher.publish(title, content)
state["published"] = success
if success:
print("✅ 文章发布成功！")
else:
print("❌ 文章发布失败")
else:
state["error"] = "没有内容可发布"
print(f"❌ {state['error']}")
except Exception as e:
state["error"] = f"发布失败: {str(e)}"
print(f"❌ {state['error']}")
state["published"] = False
return state
def _clean_final_markdown(self, text: str) -> str:
"""智能清理最终markdown内容 - 只移除包装性的代码块标记，保留内容中的代码示例"""
import re
# 检查是否整个文本被```markdown包装
if text.strip().startswith('```markdown') and text.strip().endswith('```'):
# 移除外层的```markdown包装
text = re.sub(r'^```markdowns*n?', '', text.strip())
text = re.sub(r'n?```s*$', '', text)
# 检查是否整个文本被```包装（无语言标识）
elif text.strip().startswith('```') and text.strip().endswith('```'):
lines = text.strip().split('n')
if len(lines) >= 2:
# 检查第一行是否只是```（包装性的）
first_line = lines[0].strip()
if first_line == '```' or (first_line.startswith('```') and len(first_line) <= 10):
# 移除首尾的```包装
text = 'n'.join(lines[1:-1])
# 清理多余的空行，但保持适当的段落间距
text = re.sub(r'ns*ns*n+', 'nn', text)
return text.strip()
def run(self, topic: str) -> Dict[str, Any]:
"""运行工作流"""
print(f"? 启动微信公众号文章生成工作流")
print(f"? 主题: {topic}")
print("=" * 50)
initial_state = WorkflowState(
topic=topic,
outline={},
sections=[],
final_markdown="",
images_generated=False,
published=False,
error=""
)
try:
final_state = self.graph.invoke(initial_state)
print("n" + "=" * 50)
print("? 工作流执行完成！")
if final_state.get("error"):
print(f"⚠️ 执行过程中出现错误: {final_state['error']}")
if final_state.get("published"):
print("✅ 文章已成功发布到博客")
# 输出最终的Markdown内容
if final_state.get("final_markdown"):
print("n? 生成的Markdown内容:")
print("-" * 30)
print(final_state["final_markdown"])
return final_state
except Exception as e:
error_msg = f"工作流执行失败: {str(e)}"
print(f"❌ {error_msg}")
return {"error": error_msg}
启动程序：
"""
主程序入口 - 微信公众号文章生成工作流
"""
import os
import sys
from workflow import WechatArticleWorkflow
def main():
"""主函数"""
print("? 微信公众号文章AI生成工具")
print("=" * 50)
# 检查必要的环境变量（只需要文本生成API密钥）
kimi_key = os.getenv("KIMI_API_KEY")
if not kimi_key:
print("⚠️ 缺少必要的环境变量:")
print(" - KIMI_API_KEY: Kimi API密钥（用于文本生成）")
print("n请在.env文件中设置这个变量，或者通过export命令设置")
print("示例:")
print("export KIMI_API_KEY='your-kimi-key'")
print("n? 图片生成使用免费服务，无需额外API密钥！")
print("? 获取Kimi API密钥: https://platform.m**oon*shot.cn/")
return
# 显示当前配置的图片生成服务
image_model = os.getenv("IMAGE_MODEL", "pollinations")
print(f"? 当前图片生成服务: {image_model} (免费)")
# 检查可选的图片生成API密钥
optional_keys = {
"HUGGINGFACE_API_KEY": "Hugging Face (免费1000次/月)",
"STABILITY_API_KEY": "Stability AI (新用户免费额度)"
}
available_services = ["pollinations", "picsum", "unsplash"] # 无需API密钥的服务
for key, desc in optional_keys.items():
if os.getenv(key):
available_services.append(desc)
print(f"? 可用的图片生成服务: {', '.join(available_services)}")
# 获取用户输入的主题
if len(sys.argv) > 1:
topic = " ".join(sys.argv[1:])
else:
topic = input("请输入文章主题: ").strip()
if not topic:
print("❌ 请提供文章主题")
return
# 创建并运行工作流
workflow = WechatArticleWorkflow()
try:
result = workflow.run(topic)
if result.get("error"):
print(f"n❌ 执行失败: {result['error']}")
sys.exit(1)
else:
print(f"n? 任务完成！")
# 保存Markdown到本地文件
if result.get("final_markdown"):
filename = f"article_{topic.replace(' ', '_')}.md"
with open(filename, 'w', encoding='utf-8') as f:
f.write(result["final_markdown"])
print(f"? 文章已保存到: {filename}")
except KeyboardInterrupt:
print("nn⏹️ 用户中断执行")
except Exception as e:
print(f"n❌ 程序执行出错: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
我们在启东项目的时候只需要说主题即可，比如我们这里输入：写一篇关于Python处理excel文件的使用方式，然后回车，看下日志：
(venv) ➜ wechat_article_workflow python main.py "写一篇关于Python处理excel文件的使用方式"
? 微信公众号文章AI生成工具
==================================================
? 当前图片生成服务: pollinations (免费)
? 可用的图片生成服务: pollinations, picsum, unsplash
? 启动微信公众号文章生成工作流
? 主题: 写一篇关于Python处理excel文件的使用方式
==================================================
? 开始为主题 '写一篇关于Python处理excel文件的使用方式' 生成大纲...
✅ 大纲生成完成: Python处理Excel文件：让数据分析更高效
? 开始生成文章内容...
生成段落: 引言
生成段落: Python处理Excel的必要库
生成段落: 读取Excel文件
生成段落: 修改和写入Excel文件
生成段落: 高级操作：合并与拆分Excel文件
✅ 内容生成完成，共5个段落
? 开始生成配图...
生成图片 1: 一张展示Excel文件和Python图标的图片，背景为数据分析图表
? Pollinations生成中: illustration artwork 一张展示Excel文件和Python图标的图片,背景为数据分析图表,social media illustration style,high quality,professional digital art high quality
? 图片URL: https://image.pollinations.ai/prompt/illustration%20artwork%20%E4%B8%80%E5%BC%A0%E5%B1%95%E7%A4%BAExcel%E6%96%87%E4%BB%B6%E5%92%8CPython%E5%9B%BE%E6%A0%87%E7%9A%84%E5%9B%BE%E7%89%87%2C%E8%83%8C%E6%99%AF%E4%B8%BA%E6%95%B0%E6%8D%AE%E5%88%86%E6%9E%90%E5%9B%BE%E8%A1%A8%2Csocial%20media%20illustration%20style%2Chigh%20quality%2Cprofessional%20digital%20art%20high%20quality?width=1024&height=1024&nologo=true&enhance=true
✅ 图片生成成功 (尝试 1/3)
✅ 图片生成成功
生成图片 2: 一张展示pandas、openpyxl、xlrd和xlsxwriter库的图标和名称的拼图
? Pollinations生成中: illustration artwork 一张展示pandas、openpyxl、xlrd和xlsxwriter库的图标和名称的拼图,social media illustration style,high quality,professional digital art high quality
? 图片URL: https://image.pollinations.ai/prompt/illustration%20artwork%20%E4%B8%80%E5%BC%A0%E5%B1%95%E7%A4%BApandas%E3%80%81openpyxl%E3%80%81xlrd%E5%92%8Cxlsxwriter%E5%BA%93%E7%9A%84%E5%9B%BE%E6%A0%87%E5%92%8C%E5%90%8D%E7%A7%B0%E7%9A%84%E6%8B%BC%E5%9B%BE%2Csocial%20media%20illustration%20style%2Chigh%20quality%2Cprofessional%20digital%20art%20high%20quality?width=1024&height=1024&nologo=true&enhance=true
✅ 图片生成成功 (尝试 1/3)
✅ 图片生成成功
生成图片 3: 一张代码截图，展示使用pandas读取Excel文件的示例代码
? Pollinations生成中: illustration artwork 一张代码截图,展示使用pandas读取Excel文件的示例代码,social media illustration style,high quality,professional digital art high quality
? 图片URL: https://image.pollinations.ai/prompt/illustration%20artwork%20%E4%B8%80%E5%BC%A0%E4%BB%A3%E7%A0%81%E6%88%AA%E5%9B%BE%2C%E5%B1%95%E7%A4%BA%E4%BD%BF%E7%94%A8pandas%E8%AF%BB%E5%8F%96Excel%E6%96%87%E4%BB%B6%E7%9A%84%E7%A4%BA%E4%BE%8B%E4%BB%A3%E7%A0%81%2Csocial%20media%20illustration%20style%2Chigh%20quality%2Cprofessional%20digital%20art%20high%20quality?width=1024&height=1024&nologo=true&enhance=true
⚠️ HTTP状态码: 502
? 等待3秒后重试...
⚠️ HTTP状态码: 502
? 等待3秒后重试...
✅ 图片生成成功 (尝试 3/3)
✅ 图片生成成功
生成图片 4: 一张图表，展示修改前后Excel文件的数据对比
? Pollinations生成中: illustration artwork 一张图表,展示修改前后Excel文件的数据对比,social media illustration style,high quality,professional digital art high quality
? 图片URL: https://image.pollinations.ai/prompt/illustration%20artwork%20%E4%B8%80%E5%BC%A0%E5%9B%BE%E8%A1%A8%2C%E5%B1%95%E7%A4%BA%E4%BF%AE%E6%94%B9%E5%89%8D%E5%90%8EExcel%E6%96%87%E4%BB%B6%E7%9A%84%E6%95%B0%E6%8D%AE%E5%AF%B9%E6%AF%94%2Csocial%20media%20illustration%20style%2Chigh%20quality%2Cprofessional%20digital%20art%20high%20quality?width=1024&height=1024&nologo=true&enhance=true
⏱️ 请求超时 (尝试 1/3)
? 等待3秒后重试...
✅ 图片生成成功 (尝试 2/3)
✅ 图片生成成功
生成图片 5: 一张流程图，展示合并和拆分Excel文件的步骤
? Pollinations生成中: illustration artwork 一张流程图,展示合并和拆分Excel文件的步骤,social media illustration style,high quality,professional digital art high quality
? 图片URL: https://image.pollinations.ai/prompt/illustration%20artwork%20%E4%B8%80%E5%BC%A0%E6%B5%81%E7%A8%8B%E5%9B%BE%2C%E5%B1%95%E7%A4%BA%E5%90%88%E5%B9%B6%E5%92%8C%E6%8B%86%E5%88%86Excel%E6%96%87%E4%BB%B6%E7%9A%84%E6%AD%A5%E9%AA%A4%2Csocial%20media%20illustration%20style%2Chigh%20quality%2Cprofessional%20digital%20art%20high%20quality?width=1024&height=1024&nologo=true&enhance=true
✅ 图片生成成功 (尝试 1/3)
✅ 图片生成成功
✅ 配图生成完成
? 开始编译Markdown...
✅ Markdown编译完成
? 开始发布文章...
? 正在发布文章: Python处理Excel文件：让数据分析更高效
? 文章ID: 207ef96c-6e6b-4339-8680-9215b6b78131
✅ 文章发布成功！
从开始理解主题->生成大纲->生成段落内容->图片->生成markdown->发布文章，整个过程都是自动完成的，我们可以看下最终发布到wemark平台中长啥样。
以下都是来自wemark平台中此工作流自动生成上传的文章截图：
惊艳到大家了吗，反正是让我如沐春风，耳目一新啊。从此解放双手，让你的热情和精力真正发挥在创造方面，其他的交给AI工作流吧！
最后想要学习的欢迎关注公众号并且加微信，需要资料的或者想要一起成长的欢迎打扰我。
历史大模型系列文章：
