Tutorial 10: 完整自媒体工作流
项目概述
本教程将整合前面所有知识，构建一个完整的自媒体 AI Agent 系统，实现：
内容策划 - 发现热点、分析潜力、制定计划
内容创作 - 生成大纲、撰写文章、优化内容
多平台适配 - 自动适配不同平台风格
自动发布 - 一键发布到多个平台
数据追踪 - 监控发布效果、生成报告
系统架构
┌─────────────────────────────────────────────────────────────────────┐
│ Self-Media AI Agent System │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Orchestrator │ │
│ │ (工作流编排器 - 协调各个子Agent) │ │
│ └──────────────────────────┬──────────────────────────────────┘ │
│ │ │
│ ┌────────────────────────┼────────────────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │ Planner │ │ Creator │ │ Publisher │ │
│ │ Agent │───────►│ Agent │───────►│ Agent │ │
│ │ 内容策划 │ │ 内容创作 │ │ 多平台发布 │ │
│ └──────────┘ └──────────┘ └──────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ Shared Resources │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────────────────┐ │ │
│ │ │知识库 │ │ 模板库 │ │ 配置 │ │ 状态存储 │ │ │
│ │ │ RAG │ │Prompts │ │ Config │ │ State Store │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
完整实现
项目结构
selfmedia_agent/
├── __init__.py
├── config.py # 配置管理
├── models.py # 数据模型
├── tools/ # 工具集
│ ├── __init__.py
│ ├── research.py # 话题研究工具
│ ├── writing.py # 写作工具
│ └── publishing.py # 发布工具
├── agents/ # Agent 定义
│ ├── __init__.py
│ ├── planner.py # 策划 Agent
│ ├── creator.py # 创作 Agent
│ └── publisher.py # 发布 Agent
├── knowledge/ # 知识库
│ └── writing_tips.txt
└── main.py # 主入口
配置管理 (config.py)
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import os
from dotenv import load_dotenv
load_dotenv()
class Platform(str, Enum):
WECHAT = "微信公众号"
ZHIHU = "知乎"
XIAOHONGSHU = "小红书"
TOUTIAO = "今日头条"
@dataclass
class PlatformConfig:
name: str
max_length: int
style: str
title_max_length: int
best_publish_times: List[str]
credentials: Dict = field(default_factory=dict)
@dataclass
class AgentConfig:
model_name: str = "gpt-4o-mini"
temperature: float = 0.7
max_iterations: int = 15
verbose: bool = True
class Config:
"""全局配置"""
# API Keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
# Agent 配置
AGENT = AgentConfig()
# 平台配置
PLATFORMS = {
Platform.WECHAT: PlatformConfig(
name="微信公众号",
max_length=3000,
style="深度、有温度、引发共鸣",
title_max_length=64,
best_publish_times=["08:00", "12:00", "20:00"]
),
Platform.ZHIHU: PlatformConfig(
name="知乎",
max_length=5000,
style="专业、有理有据",
title_max_length=50,
best_publish_times=["10:00", "21:00"]
),
Platform.XIAOHONGSHU: PlatformConfig(
name="小红书",
max_length=1000,
style="活泼、口语化、emoji",
title_max_length=20,
best_publish_times=["12:00", "19:00"]
),
}
数据模型 (models.py)
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from enum import Enum
class ContentStatus(str, Enum):
DRAFT = "draft"
READY = "ready"
PUBLISHED = "published"
FAILED = "failed"
@dataclass
class ContentIdea:
topic: str
angle: str
target_audience: str
platforms: List[str]
priority: int = 1
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class ContentOutline:
title: str
hook: str
sections: List[Dict[str, str]]
conclusion: str
cta: str
@dataclass
class Article:
id: str
title: str
content: str
summary: str
tags: List[str]
outline: Optional[ContentOutline] = None
status: ContentStatus = ContentStatus.DRAFT
created_at: datetime = field(default_factory=datetime.now)
@dataclass
class AdaptedArticle:
original_id: str
platform: str
title: str
content: str
summary: str
tags: List[str]
cover_suggestion: str
status: ContentStatus = ContentStatus.READY
@dataclass
class PublishRecord:
article_id: str
platform: str
success: bool
post_url: Optional[str] = None
error_message: Optional[str] = None
published_at: Optional[datetime] = None
stats: Dict = field(default_factory=dict)
主系统实现 (main.py)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.memory import ConversationBufferWindowMemory
from typing import List, Dict, Optional
from datetime import datetime
import json
import uuid
# ========== 工具定义 ==========
@tool
def search_hot_topics(category: str, count: int = 5) -> str:
"""搜索热门话题
Args:
category: 分类（科技/职场/生活/教育）
count: 数量
"""
topics = {
"科技": [
{"topic": "AI Agent 开发", "heat": 95, "trend": "🔥上升"},
{"topic": "Cursor IDE 技巧", "heat": 88, "trend": "📈上升"},
{"topic": "RAG 最佳实践", "heat": 82, "trend": "➡️稳定"},
],
"职场": [
{"topic": "AI 时代职业转型", "heat": 92, "trend": "🔥上升"},
{"topic": "远程办公效率", "heat": 78, "trend": "➡️稳定"},
]
}
return json.dumps(topics.get(category, [])[:count], ensure_ascii=False)
@tool
def analyze_topic(topic: str) -> str:
"""分析话题潜力
Args:
topic: 话题名称
"""
return json.dumps({
"topic": topic,
"potential": "高",
"competition": "中等",
"recommended_angles": ["入门教程", "实战案例", "避坑指南"],
"best_platforms": ["知乎", "微信公众号"]
}, ensure_ascii=False)
@tool
def create_outline(topic: str, platform: str, angle: str) -> str:
"""创建文章大纲
Args:
topic: 话题
platform: 目标平台
angle: 切入角度
"""
return json.dumps({
"title": f"{topic}：{angle}",
"hook": "你是否也遇到过这样的困扰...",
"sections": [
{"title": "背景介绍", "points": ["问题描述", "为什么重要"]},
{"title": "核心方法", "points": ["方法1", "方法2", "方法3"]},
{"title": "实战案例", "points": ["案例分析", "代码演示"]},
{"title": "常见问题", "points": ["FAQ1", "FAQ2"]},
],
"conclusion": "总结核心要点",
"cta": "关注获取更多干货"
}, ensure_ascii=False)
@tool
def write_section(section_title: str, points: str, style: str) -> str:
"""撰写文章章节
Args:
section_title: 章节标题
points: 要点（逗号分隔）
style: 写作风格
"""
# 实际应用中这里会调用 LLM 生成内容
return f"""
## {section_title}
{points}
这是一段根据 {style} 风格生成的内容示例...
"""
@tool
def adapt_for_platform(content: str, source_platform: str, target_platform: str) -> str:
"""将内容适配到目标平台
Args:
content: 原始内容
source_platform: 来源平台
target_platform: 目标平台
"""
adaptations = {
"小红书": "已将内容改为口语化风格，添加了emoji 📚✨",
"知乎": "已将内容改为专业风格，添加了数据引用",
"微信公众号": "已优化排版，添加了引导关注"
}
return json.dumps({
"target_platform": target_platform,
"adaptation_note": adaptations.get(target_platform, "已完成适配"),
"content_preview": content[:100] + "..."
}, ensure_ascii=False)
@tool
def publish_content(platform: str, title: str, content: str) -> str:
"""发布内容到平台
Args:
platform: 目标平台
title: 标题
content: 内容
"""
# 模拟发布
post_id = str(uuid.uuid4())[:8]
return json.dumps({
"success": True,
"platform": platform,
"post_id": post_id,
"url": f"https://{platform.lower()}.com/p/{post_id}",
"published_at": datetime.now().isoformat()
}, ensure_ascii=False)
@tool
def get_content_stats(platform: str, post_id: str) -> str:
"""获取内容数据统计
Args:
platform: 平台
post_id: 文章ID
"""
import random
return json.dumps({
"platform": platform,
"post_id": post_id,
"views": random.randint(1000, 10000),
"likes": random.randint(50, 500),
"comments": random.randint(10, 100),
"shares": random.randint(5, 50)
}, ensure_ascii=False)
# ========== 主 Agent 系统 ==========
class SelfMediaAgentSystem:
"""自媒体 AI Agent 系统"""
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
# 所有工具
self.tools = [
search_hot_topics,
analyze_topic,
create_outline,
write_section,
adapt_for_platform,
publish_content,
get_content_stats
]
# 对话记忆
self.memory = ConversationBufferWindowMemory(
k=10,
return_messages=True,
memory_key="chat_history"
)
# 系统提示
self.system_prompt = """你是专业的自媒体 AI 助手，具备以下能力：
🎯 **内容策划**
- 发现热门话题和内容机会
- 分析话题潜力和最佳切入角度
- 制定内容发布计划
✍️ **内容创作**
- 创建结构化的文章大纲
- 撰写高质量的文章内容
- 优化标题、摘要和标签
📱 **多平台运营**
- 适配不同平台的内容风格
- 自动发布到多个平台
- 追踪内容表现数据
工作原则：
1. 内容要有价值，不做标题党
2. 适配平台特点，尊重用户习惯
3. 数据驱动，持续优化
4. 保持专业性和原创性
请根据用户需求，合理使用工具完成任务。每次任务完成后，给出清晰的总结和下一步建议。
"""
self.prompt = ChatPromptTemplate.from_messages([
("system", self.system_prompt),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
self.agent = create_tool_calling_agent(self.llm, self.tools, self.prompt)
self.executor = AgentExecutor(
agent=self.agent,
tools=self.tools,
verbose=True,
max_iterations=20,
handle_parsing_errors=True
)
def chat(self, message: str) -> str:
"""与 Agent 对话"""
history = self.memory.load_memory_variables({})["chat_history"]
result = self.executor.invoke({
"input": message,
"chat_history": history
})
# 保存对话历史
self.memory.save_context(
{"input": message},
{"output": result["output"]}
)
return result["output"]
def plan_content(self, category: str, platforms: List[str], days: int = 7) -> str:
"""制定内容计划"""
platforms_str = "、".join(platforms)
return self.chat(f"""
请帮我制定一个{days}天的内容计划：
- 内容领域：{category}
- 目标平台：{platforms_str}
要求：
1. 搜索当前热门话题
2. 分析每个话题的潜力
3. 为每天安排一个内容主题
4. 说明每个内容的切入角度和目标平台
""")
def create_content(self, topic: str, platform: str, style: str = None) -> str:
"""创作内容"""
style_info = f"，风格要求：{style}" if style else ""
return self.chat(f"""
请帮我创作一篇关于「{topic}」的文章，发布到「{platform}」{style_info}。
请执行以下步骤：
1. 分析话题，确定最佳切入角度
2. 创建详细的文章大纲
3. 逐章节撰写内容
4. 生成标题、摘要和标签
""")
def publish_to_platforms(self, content: str, title: str, platforms: List[str]) -> str:
"""发布到多个平台"""
platforms_str = "、".join(platforms)
return self.chat(f"""
请将以下内容发布到这些平台：{platforms_str}
标题：{title}
内容：
{content}
请执行以下步骤：
1. 为每个平台适配内容风格
2. 发布到各平台
3. 汇报发布结果
""")
def get_performance_report(self, post_ids: Dict[str, str]) -> str:
"""获取发布效果报告"""
posts_info = "\n".join([f"- {p}: {pid}" for p, pid in post_ids.items()])
return self.chat(f"""
请获取以下文章的数据表现：
{posts_info}
请生成一份数据报告，包括：
1. 各平台的阅读量、点赞、评论、分享数据
2. 数据对比分析
3. 优化建议
""")
def full_workflow(self, topic: str, platforms: List[str]) -> str:
"""执行完整工作流"""
platforms_str = "、".join(platforms)
return self.chat(f"""
请执行完整的自媒体内容工作流：
话题：{topic}
目标平台：{platforms_str}
完整流程：
1. 分析话题潜力和最佳切入角度
2. 创建文章大纲
3. 撰写完整文章
4. 为每个平台适配内容
5. 发布到所有平台
6. 生成发布报告
请逐步执行，每一步都给出详细输出。
""")
# ========== 使用示例 ==========
def main():
print("=" * 60)
print("自媒体 AI Agent 系统")
print("=" * 60)
agent = SelfMediaAgentSystem()
# 交互式对话
print("\n欢迎使用自媒体 AI Agent！")
print("你可以：")
print("1. 让我帮你发现热门话题")
print("2. 让我帮你创作文章")
print("3. 让我帮你发布到多个平台")
print("4. 输入 'quit' 退出")
print("-" * 60)
while True:
user_input = input("\n你: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("再见！")
break
if not user_input:
continue
print("\nAgent 正在处理...")
response = agent.chat(user_input)
print(f"\nAgent: {response}")
if __name__ == "__main__":
main()
使用示例
# 创建 Agent 系统
agent = SelfMediaAgentSystem()
# 场景1：制定内容计划
plan = agent.plan_content(
category="科技",
platforms=["微信公众号", "知乎"],
days=7
)
print(plan)
# 场景2：创作内容
content = agent.create_content(
topic="AI Agent 开发入门",
platform="知乎",
style="专业但易懂"
)
print(content)
# 场景3：多平台发布
result = agent.publish_to_platforms(
content="...",
title="AI Agent 开发入门指南",
platforms=["微信公众号", "知乎", "小红书"]
)
print(result)
# 场景4：完整工作流
full_result = agent.full_workflow(
topic="LangChain 实战教程",
platforms=["微信公众号", "知乎"]
)
print(full_result)
部署建议
环境配置
# 创建 .env 文件 OPENAI_API_KEY=your-api-key # 安装依赖 pip install langchain langchain-openai python-dotenv
生产环境优化
使用 Redis 存储对话历史
添加请求限流和重试机制
实现异步发布队列
添加监控和告警
扩展方向
接入真实的平台 API
添加图片生成能力
实现定时发布功能
添加数据分析仪表板
总结
通过本系列教程，我们学习了：
LangChain 基础 - LLM、Prompts、Chains
高级特性 - Memory、Agents、Tools
RAG 技术 - 知识库构建与检索
实战应用 - 自媒体内容创作与发布
现在你已经掌握了使用 LangChain 构建 AI Agent 的核心技能，可以将这些知识应用到更多场景中！
🎉 恭喜完成全部教程！