如何用 Multi-Agent 自动运营抖音账号：从脚本到内容闭环

---

一、 引言 (Introduction)

钩子 (The Hook)

你是否曾梦想过拥有一个24/7不间断运营的抖音账号？一个能够自动生成创意脚本、制作高质量内容、智能发布，甚至能够与粉丝互动的全自动化系统？在当今这个内容为王的时代，短视频平台已经成为了个人品牌建设和商业推广的主战场。然而，持续产出高质量内容、保持账号活跃度、与粉丝互动，这一切都需要投入巨大的时间和精力。

想象一下：当你还在梦乡时，你的账号已经自动完成了一个爆款视频的制作和发布；当你忙于工作时，系统正在智能地回复粉丝的评论和私信；当你在思考下一个创意时，AI已经为你准备好了几十个选题方案。这听起来像是科幻小说，但随着大语言模型和多智能体（Multi-Agent）技术的快速发展，这一切正在变成现实。

定义问题/阐述背景 (The “Why”)

在短视频运营领域，内容创作者和运营者面临着诸多挑战：

1. 内容生产压力：抖音算法偏好高频更新，创作者需要不断产出新内容，这对个人创作者来说是巨大的负担。
2. 创意枯竭：即使是最有创意的人，也会面临灵感枯竭的时刻。
3. 数据分析复杂：理解平台算法、分析数据表现、优化内容策略需要专业知识和大量时间。
4. 粉丝互动耗时：回复评论、处理私信、维护社区氛围占用了大量运营时间。
5. 多平台管理困难：如果同时运营多个平台或账号，工作量更是成倍增加。

传统的自动化工具往往只能解决单一问题，比如定时发布或简单的数据分析，无法形成完整的内容生产和运营闭环。而Multi-Agent系统，通过模拟人类团队协作的方式，能够将复杂的运营工作分解为多个专业化任务，由不同的智能体协同完成，从而实现真正意义上的全自动化运营。

亮明观点/文章目标 (The “What” & “How”)

在这篇文章中，我将带你从零开始，构建一个基于Multi-Agent的抖音账号全自动运营系统。我们将涵盖从内容创意生成、脚本撰写、视频制作、发布排期到粉丝互动的完整流程。

具体来说，你将学到：

1. Multi-Agent系统的核心概念：理解什么是多智能体系统，以及它如何应用于内容运营。
2. 系统架构设计：如何设计一个分工明确、协同高效的智能体团队。
3. 核心模块实现：
   • 创意生成Agent：如何利用AI生成爆款选题
   • 剧本创作Agent：如何写出吸引人的短视频脚本
   • 内容制作Agent：如何自动化生成视频内容
   • 数据分析Agent：如何智能分析内容表现并优化策略
   • 互动管理Agent：如何自动化回复评论和私信
4. 完整闭环构建：如何将这些模块连接起来，形成一个自学习、自优化的运营系统。
5. 实战案例：通过一个真实的账号运营案例，展示系统的实际效果。

我们将使用Python作为主要开发语言，结合LangChain等框架来构建我们的Multi-Agent系统。无论你是内容创作者、运营人员还是技术开发者，这篇文章都将为你打开一扇通往自动化内容运营的大门。

---

二、 基础知识/背景铺垫 (Foundational Concepts)

核心概念定义

在深入探讨如何构建Multi-Agent抖音运营系统之前，让我们先明确一些关键概念：

什么是Multi-Agent系统？

Multi-Agent系统（多智能体系统） 是由多个相互作用的智能体（Agent）组成的计算系统。每个智能体都是一个自主的实体，能够感知环境、做出决策并采取行动。这些智能体可以共享信息、协作完成任务，甚至可以通过学习不断改进自己的行为。

在我们的抖音运营场景中，每个智能体可以看作是团队中的一个专业角色：创意总监、编剧、剪辑师、数据分析师、社区管理员等。它们各自负责特定的任务，但又紧密协作，共同完成账号运营的整体目标。

大语言模型（LLM）在Multi-Agent中的角色

大语言模型（Large Language Model, LLM） 如GPT-4、Claude、文心一言等，是构建现代Multi-Agent系统的核心技术。这些模型具有强大的理解、推理和生成能力，使得智能体能够：

1. 理解复杂任务：解析自然语言指令，理解运营目标和用户需求。
2. 生成高质量内容：创作脚本、回复评论、撰写文案。
3. 逻辑推理：分析数据、制定策略、解决问题。
4. 工具使用：调用外部API、访问数据库、操作其他软件系统。

抖音平台运营的关键要素

要成功运营一个抖音账号，需要理解以下关键要素：

1. 内容质量：视频的创意、制作水准、信息价值是基础。
2. 发布时机：在用户活跃度高的时段发布可以获得更多曝光。
3. 互动率：评论、点赞、分享、完播率等指标影响算法推荐。
4. 账号定位：清晰的垂直领域定位有助于吸引精准粉丝。
5. 热点追踪：及时结合热点话题可以获得额外流量。

相关工具/技术概览

在构建我们的系统之前，让我们先了解一些将用到的核心工具和技术：

LangChain

LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它提供了一套工具和抽象，使我们能够更容易地：

• 连接LLM到其他数据源
• 创建链式的LLM调用
• 构建具有记忆和推理能力的应用
• 开发Multi-Agent系统

LangChain的Agent框架特别适合我们的场景，因为它允许我们创建能够使用工具、具有推理能力的智能体。

OpenAI API / 其他LLM API

我们需要访问强大的LLM来驱动我们的智能体。OpenAI的GPT-4是一个很好的选择，因为它具有强大的理解和生成能力。当然，你也可以选择其他模型，如Anthropic的Claude、Google的PaLM，或者国内的文心一言、通义千问等。

抖音开放平台API

要实现真正的自动化，我们需要与抖音平台进行交互。抖音开放平台提供了一系列API，允许我们：

• 上传和发布视频
• 获取账号数据和视频表现
• 读取和回复评论
• 管理私信

需要注意的是，使用这些API通常需要申请开发者账号并获得相应的权限。

视频处理工具

自动化制作视频需要一些视频处理工具：

• MoviePy：一个用于视频编辑的Python库，可以用来剪辑视频、添加音频、文本等。
• FFmpeg：一个强大的多媒体处理工具，可以处理几乎所有的视频和音频格式。
• Pillow：用于图像处理的Python库，可以用来创建和编辑图片。

数据存储和分析

• SQLite/PostgreSQL：用于存储账号数据、内容历史、用户互动等信息。
• Pandas：用于数据分析和处理的Python库。
• Matplotlib/Plotly：用于数据可视化。

---

三、 核心内容/实战演练 (The Core - “How-To”)

系统整体架构设计

在开始编写代码之前，让我们先设计整个系统的架构。我们的Multi-Agent抖音运营系统将由以下几个核心智能体组成：

1. 创意总监Agent（Creative Director Agent）
2. 剧本作家Agent（Script Writer Agent）
3. 内容制作Agent（Content Producer Agent）
4. 发布策略Agent（Publishing Strategist Agent）
5. 数据分析Agent（Data Analyst Agent）
6. 社区管理Agent（Community Manager Agent）

这些智能体将通过一个中央协调器（Central Coordinator）进行通信和协作。此外，我们还需要一些共享组件，如记忆系统（Memory System）、工具库（Tool Library）和数据存储（Data Storage）。

让我们用Mermaid图来可视化这个架构：

现在，让我们逐一实现这些智能体和组件。

环境搭建和依赖安装

首先，让我们设置开发环境并安装必要的依赖：

# 首先，创建一个虚拟环境（推荐）
# python -m venv douyin_agent_env
# source douyin_agent_env/bin/activate  # Linux/Mac
# .\douyin_agent_env\Scripts\activate  # Windows

# 安装必要的包
!pip install langchain openai python-dotenv moviepy pillow pandas matplotlib sqlite3 requests schedule

接下来，创建一个 .env 文件来存储我们的API密钥和配置：

OPENAI_API_KEY=your_openai_api_key_here
DOUYIN_CLIENT_KEY=your_douyin_client_key_here
DOUYIN_CLIENT_SECRET=your_douyin_client_secret_here

基础组件实现

首先，让我们实现一些基础组件，这些将被所有智能体使用。

记忆系统（Memory System）

记忆系统允许智能体记住过去的交互、决策和结果，这对于保持一致性和持续学习至关重要。

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain.schema import BaseMessage
import sqlite3
from datetime import datetime
import json
from typing import List, Dict, Any

class MemorySystem:
    def __init__(self, db_path: str = "douyin_agent_memory.db"):
        self.db_path = db_path
        self._init_db()
        self.conversation_memory = ConversationBufferMemory(return_messages=True)
        self.summary_memory = ConversationSummaryMemory(llm=None)  # 我们会在使用时设置LLM
        
    def _init_db(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # 创建内容历史表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS content_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            content_id TEXT UNIQUE,
            content_type TEXT,
            title TEXT,
            description TEXT,
            publish_date TIMESTAMP,
            performance_data TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        # 创建决策记录表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS decision_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            agent_name TEXT,
            decision_type TEXT,
            decision_content TEXT,
            context TEXT,
            outcome TEXT,
            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        # 创建用户互动表
        cursor.execute('''
        CREATE TABLE IF NOT EXISTS user_interactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            interaction_type TEXT,
            user_id TEXT,
            content_id TEXT,
            interaction_content TEXT,
            response TEXT,
            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        ''')
        
        conn.commit()
        conn.close()
    
    def add_conversation_message(self, message: BaseMessage):
        """添加对话消息到记忆"""
        self.conversation_memory.chat_memory.add_message(message)
    
    def get_conversation_history(self) -> List[BaseMessage]:
        """获取对话历史"""
        return self.conversation_memory.chat_memory.messages
    
    def record_content(self, content_id: str, content_type: str, title: str, 
                      description: str, publish_date: datetime = None, 
                      performance_data: Dict = None):
        """记录内容发布信息"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
            INSERT OR REPLACE INTO content_history 
            (content_id, content_type, title, description, publish_date, performance_data)
            VALUES (?, ?, ?, ?, ?, ?)
            ''', (
                content_id, 
                content_type, 
                title, 
                description, 
                publish_date.isoformat() if publish_date else None,
                json.dumps(performance_data) if performance_data else None
            ))
            conn.commit()
        except Exception as e:
            print(f"Error recording content: {e}")
        finally:
            conn.close()
    
    def record_decision(self, agent_name: str, decision_type: str, 
                       decision_content: str, context: Dict = None, 
                       outcome: str = None):
        """记录智能体决策"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
            INSERT INTO decision_history 
            (agent_name, decision_type, decision_content, context, outcome)
            VALUES (?, ?, ?, ?, ?)
            ''', (
                agent_name,
                decision_type,
                decision_content,
                json.dumps(context) if context else None,
                outcome
            ))
            conn.commit()
        except Exception as e:
            print(f"Error recording decision: {e}")
        finally:
            conn.close()
    
    def record_user_interaction(self, interaction_type: str, user_id: str, 
                               content_id: str, interaction_content: str, 
                               response: str = None):
        """记录用户互动"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        try:
            cursor.execute('''
            INSERT INTO user_interactions 
            (interaction_type, user_id, content_id, interaction_content, response)
            VALUES (?, ?, ?, ?, ?)
            ''', (
                interaction_type,
                user_id,
                content_id,
                interaction_content,
                response
            ))
            conn.commit()
        except Exception as e:
            print(f"Error recording user interaction: {e}")
        finally:
            conn.close()
    
    def get_content_history(self, limit: int = 10) -> List[Dict]:
        """获取内容历史"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
        SELECT content_id, content_type, title, description, publish_date, performance_data
        FROM content_history
        ORDER BY publish_date DESC
        LIMIT ?
        ''', (limit,))
        
        results = cursor.fetchall()
        conn.close()
        
        content_list = []
        for row in results:
            content_list.append({
                'content_id': row[0],
                'content_type': row[1],
                'title': row[2],
                'description': row[3],
                'publish_date': row[4],
                'performance_data': json.loads(row[5]) if row[5] else None
            })
        
        return content_list
    
    def get_top_performing_content(self, metric: str = 'view_count', limit: int = 5) -> List[Dict]:
        """获取表现最好的内容"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
        SELECT content_id, content_type, title, description, publish_date, performance_data
        FROM content_history
        WHERE performance_data IS NOT NULL
        ORDER BY publish_date DESC
        LIMIT 100
        ''')
        
        results = cursor.fetchall()
        conn.close()
        
        # 解析性能数据并排序
        content_list = []
        for row in results:
            perf_data = json.loads(row[5]) if row[5] else {}
            if metric in perf_data:
                content_list.append({
                    'content_id': row[0],
                    'content_type': row[1],
                    'title': row[2],
                    'description': row[3],
                    'publish_date': row[4],
                    'performance_data': perf_data,
                    'metric_value': perf_data[metric]
                })
        
        # 按指标值排序
        content_list.sort(key=lambda x: x['metric_value'], reverse=True)
        return content_list[:limit]

工具库（Tool Library）

接下来，让我们创建一个工具库，包含各种智能体可能需要使用的工具：

import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class ToolLibrary:
    def __init__(self):
        self.douyin_api_base = "https://open.douyin.com"
        self.douyin_client_key = os.getenv("DOUYIN_CLIENT_KEY")
        self.douyin_client_secret = os.getenv("DOUYIN_CLIENT_SECRET")
        self.access_token = None
        self._refresh_access_token()
    
    def _refresh_access_token(self):
        """刷新抖音API访问令牌"""
        url = f"{self.douyin_api_base}/oauth/client_token/"
        params = {
            "client_key": self.douyin_client_key,
            "client_secret": self.douyin_client_secret,
            "grant_type": "client_credential"
        }
        
        try:
            response = requests.get(url, params=params)
            response.raise_for_status()
            data = response.json()
            
            if data.get("data"):
                self.access_token = data["data"].get("access_token")
                print("Access token refreshed successfully")
            else:
                print(f"Failed to refresh access token: {data}")
        except Exception as e:
            print(f"Error refreshing access token: {e}")
    
    def get_trending_topics(self, category: str = "general") -> List[Dict]:
        """获取当前热门话题"""
        # 在实际应用中，这里可以连接到抖音的热点API或第三方热点服务
        # 这里我们使用模拟数据作为示例
        mock_trending_topics = [
            {"id": "1", "title": "人工智能新突破", "hot_value": 9800000, "category": "科技"},
            {"id": "2", "title": "健康饮食小贴士", "hot_value": 7500000, "category": "生活"},
            {"id": "3", "title": "旅行推荐地", "hot_value": 6200000, "category": "旅行"},
            {"id": "4", "title": "职场技能提升", "hot_value": 5800000, "category": "教育"},
            {"id": "5", "title": "搞笑日常", "hot_value": 8900000, "category": "娱乐"}
        ]
        
        if category != "general":
            return [topic for topic in mock_trending_topics if topic["category"] == category]
        
        return mock_trending_topics
    
    def search_content_ideas(self, keyword: str, limit: int = 10) -> List[Dict]:
        """搜索内容创意"""
        # 在实际应用中，这里可以连接到抖音的搜索API或第三方内容分析服务
        # 这里我们使用模拟数据作为示例
        mock_ideas = [
            {"title": f"{keyword}的5个技巧", "type": "教程", "potential": "高"},
            {"title": f"{keyword}常见错误", "type": "纠错", "potential": "中"},
            {"title": f"{keyword}VS 其他方法", "type": "对比", "potential": "高"},
            {"title": f"我是如何学习{keyword}的", "type": "个人经验", "potential": "中"},
            {"title": f"{keyword}的未来发展", "type": "趋势分析", "potential": "中高"}
        ]
        
        return mock_ideas[:limit]
    
    def get_account_data(self) -> Dict:
        """获取账号数据"""
        # 在实际应用中，这里应该调用抖音API
        # 这里我们使用模拟数据
        return {
            "fans_count": 52000,
            "following_count": 320,
            "total_likes": 1200000,
            "total_views": 45000000,
            "avg_views_per_video": 150000,
            "avg_likes_per_video": 4000,
            "avg_comments_per_video": 150,
            "avg_shares_per_video": 80
        }
    
    def get_video_performance(self, video_id: str) -> Dict:
        """获取视频表现数据"""
        # 在实际应用中，这里应该调用抖音API
        # 这里我们使用模拟数据
        return {
            "video_id": video_id,
            "view_count": 125000,
            "like_count": 3200,
            "comment_count": 180,
            "share_count": 95,
            "completion_rate": 0.68,  # 完播率
            "average_watch_time": 12.5,  # 平均观看时长（秒）
            "publish_time": (datetime.now() - timedelta(days=3)).isoformat(),
            "performance_trend": "increasing"  # 表现趋势
        }
    
    def get_comments(self, video_id: str, limit: int = 50) -> List[Dict]:
        """获取视频评论"""
        # 在实际应用中，这里应该调用抖音API
        # 这里我们使用模拟数据
        mock_comments = [
            {"comment_id": "1", "user_id": "user1", "content": "这个视频太有用了！", "like_count": 45, "timestamp": "2023-10-01T10:30:00"},
            {"comment_id": "2", "user_id": "user2", "content": "请问第二步怎么做？", "like_count": 12, "timestamp": "2023-10-01T11:15:00"},
            {"comment_id": "3", "user_id": "user3", "content": "已收藏，以后慢慢看", "like_count": 8, "timestamp": "2023-10-01T12:45:00"},
            {"comment_id": "4", "user_id": "user4", "content": "不同意你的观点，我认为...", "like_count": 23, "timestamp": "2023-10-01T14:20:00"},
            {"comment_id": "5", "user_id": "user5", "content": "能不能出一期更详细的？", "like_count": 31, "timestamp": "2023-10-01T16:05:00"}
        ]
        
        return mock_comments[:limit]
    
    def reply_to_comment(self, comment_id: str, reply_content: str) -> bool:
        """回复评论"""
        # 在实际应用中，这里应该调用抖音API
        print(f"Replying to comment {comment_id}: {reply_content}")
        return True  # 假设回复成功
    
    def upload_video(self, video_path: str, title: str, description: str, 
                   tags: List[str] = None, publish_time: datetime = None) -> Optional[str]:
        """上传视频（并可选地定时发布）"""
        # 在实际应用中，这里应该调用抖音API上传视频
        print(f"Uploading video: {title}")
        print(f"Description: {description}")
        print(f"Tags: {tags}")
        
        # 模拟生成一个视频ID
        video_id = f"video_{datetime.now().strftime('%Y%m%d%H%M%S')}"
        return video_id
    
    def schedule_video_publish(self, video_id: str, publish_time: datetime) -> bool:
        """安排视频发布时间"""
        # 在实际应用中，这里应该调用抖音API
        print(f"Scheduling video {video_id} to be published at {publish_time}")
        return True  # 假设操作成功

智能体实现

现在，让我们开始实现各个智能体。我们将使用LangChain框架来构建这些智能体。

创意总监Agent（Creative Director Agent）

创意总监Agent负责生成内容创意和选题方向。它会分析热门话题、账号历史表现和粉丝兴趣，提出有潜力的内容创意。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.agents import AgentType, initialize_agent, Tool
from typing import List, Dict, Any
import json
from datetime import datetime

class CreativeDirectorAgent:
    def __init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
        self.memory = memory_system
        self.tools = tool_library
        self.llm = OpenAI(temperature=0.7, max_tokens=2000)  # 较高的温度以获得更多创意
        
        # 定义创意总监使用的工具
        self.agent_tools = [
            Tool(
                name="获取热门话题",
                func=self.tools.get_trending_topics,
                description="获取当前抖音上的热门话题，可用于内容创意"
            ),
            Tool(
                name="搜索内容创意",
                func=lambda keyword: json.dumps(self.tools.search_content_ideas(keyword), ensure_ascii=False),
                description="基于关键词搜索相关的内容创意"
            ),
            Tool(
                name="获取账号历史内容",
                func=lambda limit=10: json.dumps(self.memory.get_content_history(limit), ensure_ascii=False),
                description="获取账号历史发布的内容，用于参考和避免重复"
            ),
            Tool(
                name="获取高表现内容",
                func=lambda metric='view_count', limit=5: json.dumps(self.memory.get_top_performing_content(metric, limit), ensure_ascii=False),
                description="获取账号历史上表现最好的内容，分析成功因素"
            )
        ]
        
        # 创建创意总监Agent
        self.agent = initialize_agent(
            self.agent_tools,
            self.llm,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True
        )
        
        # 创意生成提示模板
        self.creative_prompt = PromptTemplate(
            input_variables=["account_data", "content_objectives", "constraints"],
            template="""你是一位专业的抖音内容创意总监，负责为账号制定内容策略和生成创意选题。
            
            账号数据：
            {account_data}
            
            内容目标：
            {content_objectives}
            
            约束条件：
            {constraints}
            
            请按照以下步骤进行思考和工作：
            1. 首先，了解当前的热门话题和趋势
            2. 查看账号历史表现最好的内容，分析成功因素
            3. 确保新创意与账号定位一致，同时避免与历史内容重复
            4. 生成5-7个有潜力的内容创意，每个创意包含：标题、类型、简要描述、预期效果、预估难度
            
            请以JSON格式返回你的创意建议。
            """
        )
    
    def generate_content_ideas(self, content_objectives: str = "增加粉丝互动和账号曝光", 
                              constraints: str = "保持账号垂直领域定位，每视频时长控制在15-60秒") -> List[Dict]:
        """生成内容创意"""
        account_data = self.tools.get_account_data()
        account_data_str = json.dumps(account_data, ensure_ascii=False)
        
        prompt = self.creative_prompt.format(
            account_data=account_data_str,
            content_objectives=content_objectives,
            constraints=constraints
        )
        
        try:
            result = self.agent.run(prompt)
            
            # 记录决策
            self.memory.record_decision(
                agent_name="CreativeDirectorAgent",
                decision_type="content_ideas_generation",
                decision_content=result,
                context={
                    "content_objectives": content_objectives,
                    "constraints": constraints,
                    "timestamp": datetime.now().isoformat()
                }
            )
            
            # 尝试解析JSON结果
            try:
                ideas = json.loads(result)
                return ideas if isinstance(ideas, list) else [ideas]
            except json.JSONDecodeError:
                # 如果不是有效的JSON，返回原始结果
                return [{"raw_result": result}]
                
        except Exception as e:
            print(f"Error generating content ideas: {e}")
            return []
    
    def refine_content_idea(self, idea: Dict, feedback: str) -> Dict:
        """根据反馈优化内容创意"""
        refine_prompt = f"""
        原始创意：
        {json.dumps(idea, ensure_ascii=False)}
        
        反馈意见：
        {feedback}
        
        请根据反馈意见优化这个内容创意，保留其核心价值的同时，解决反馈中提到的问题。
        请以JSON格式返回优化后的创意。
        """
        
        try:
            result = self.agent.run(refine_prompt)
            
            # 记录决策
            self.memory.record_decision(
                agent_name="CreativeDirectorAgent",
                decision_type="content_idea_refinement",
                decision_content=result,
                context={
                    "original_idea": idea,
                    "feedback": feedback,
                    "timestamp": datetime.now().isoformat()
                }
            )
            
            try:
                refined_idea = json.loads(result)
                return refined_idea
            except json.JSONDecodeError:
                return {"raw_result": result, "original_idea": idea}
                
        except Exception as e:
            print(f"Error refining content idea: {e}")
            return idea

剧本作家Agent（Script Writer Agent）

剧本作家Agent负责将创意转化为可拍摄的详细脚本，包括场景描述、台词、镜头建议等。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.agents import AgentType, initialize_agent, Tool
from typing import List, Dict, Any
import json
from datetime import datetime

class ScriptWriterAgent:
    def __init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
        self.memory = memory_system
        self.tools = tool_library
        self.llm = OpenAI(temperature=0.8, max_tokens=3000)  # 较高的温度以获得更有创意的脚本
        
        # 定义剧本作家使用的工具
        self.agent_tools = [
            Tool(
                name="获取高表现内容",
                func=lambda metric='view_count', limit=5: json.dumps(self.memory.get_top_performing_content(metric, limit), ensure_ascii=False),
                description="获取账号历史上表现最好的内容，分析其脚本结构和风格"
            ),
            Tool(
                name="搜索内容创意",
                func=lambda keyword: json.dumps(self.tools.search_content_ideas(keyword), ensure_ascii=False),
                description="基于关键词搜索相关的内容创意，为脚本提供更多灵感"
            )
        ]
        
        # 创建剧本作家Agent
        self.agent = initialize_agent(
            self.agent_tools,
            self.llm,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True
        )
        
        # 脚本写作提示模板
        self.script_prompt = PromptTemplate(
            input_variables=["content_idea", "account_style", "duration_constraint"],
            template="""你是一位专业的抖音短视频编剧，擅长创作引人入胜、适合平台算法的短视频脚本。
            
            内容创意：
            {content_idea}
            
            账号风格定位：
            {account_style}
            
            时长约束：
            {duration_constraint}
            
            请按照以下步骤进行思考和工作：
            1. 首先，分析这个内容创意的核心价值和吸引点
            2. 查看账号历史上表现最好的内容，分析其成功的脚本结构
            3. 创作一个完整的短视频脚本，包含以下元素：
               - 视频概述：简要描述视频内容和目标
               - 钩子设计：前3秒如何吸引观众注意力
               - 详细脚本：按时间线分解，包含场景、画面描述、台词/旁白、音乐/音效建议
               - 互动设计：如何引导用户点赞、评论、分享
               - 话题标签建议：相关的热门话题和垂直领域标签
            
            请以JSON格式返回你的脚本。
            """
        )
    
    def write_script(self, content_idea: Dict, 
                    account_style: str = "知识科普类，专业但不失趣味，语速适中，画面清晰",
                    duration_constraint: str = "15-30秒") -> Dict:
        """编写脚本"""
        content_idea_str = json.dumps(content_idea, ensure_ascii=False)
        
        prompt = self.script_prompt.format(
            content_idea=content_idea_str,
            account_style=account_style,
            duration_constraint=duration_constraint
        )
        
        try:
            result = self.agent.run(prompt)
            
            # 记录决策
            self.memory.record_decision(
                agent_name="ScriptWriterAgent",
                decision_type="script_writing",
                decision_content=result,
                context={
                    "content_idea": content_idea,
                    "account_style": account_style,
                    "duration_constraint": duration_constraint,
                    "timestamp": datetime.now().isoformat()
                }
            )
            
            # 尝试解析JSON结果
            try:
                script = json.loads(result)
                return script
            except json.JSONDecodeError:
                # 如果不是有效的JSON，返回原始结果
                return {"raw_result": result}
                
        except Exception as e:
            print(f"Error writing script: {e}")
            return {}
    
    def refine_script(self, script: Dict, feedback: str) -> Dict:
        """根据反馈优化脚本"""
        refine_prompt = f"""
        原始脚本：
        {json.dumps(script, ensure_ascii=False)}
        
        反馈意见：
        {feedback}
        
        请根据反馈意见优化这个脚本，保留其核心价值的同时，解决反馈中提到的问题。
        请以JSON格式返回优化后的脚本。
        """
        
        try:
            result = self.agent.run(refine_prompt)
            
            # 记录决策
            self.memory.record_decision(
                agent_name="ScriptWriterAgent",
                decision_type="script_refinement",
                decision_content=result,
                context={
                    "original_script": script,
                    "feedback": feedback,
                    "timestamp": datetime.now().isoformat()
                }
            )
            
            try:
                refined_script = json.loads(result)
                return refined_script
            except json.JSONDecodeError:
                return {"raw_result": result, "original_script": script}
                
        except Exception as e:
            print(f"Error refining script: {e}")
            return script

内容制作Agent（Content Producer Agent）

内容制作Agent负责根据脚本制作视频。它可以使用模板、素材库、文本转语音等工具来自动化生成视频。

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.agents import AgentType, initialize_agent, Tool
from typing import List, Dict, Any, Optional
import json
from datetime import datetime
import os
from moviepy.editor import *
from gtts import gTTS
import requests
from PIL import Image, ImageDraw, ImageFont

class ContentProducerAgent:
    def __init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
        self.memory = memory_system
        self.tools = tool_library
        self.llm = OpenAI(temperature=0.3, max_tokens=1500)  # 较低的温度以获得更精确的执行
        
        # 素材库路径
        self.background_videos_path = "assets/background_videos"
        self.music_path = "assets/music"
        self.images_path = "assets/images"
        self.output_path = "output/videos"
        
        # 确保目录存在
        for path in [self.background_videos_path, self.music_path, self.images_path, self.output_path]:
            os.makedirs(path, exist_ok=True)
        
        # 定义内容制作使用的工具
        self.agent_tools = [
            Tool(
                name="文本转语音",
                func=self._text_to_speech,
                description="将文本转换为语音，参数为文本内容和可选的语言代码"
            ),
            Tool(
                name="搜索素材",
                func=self._search_materials,
                description="根据关键词搜索相关的素材（图片、视频、音乐）"
            )
        ]
        
        # 创建内容制作Agent
        self.agent = initialize_agent(
            self.agent_tools,
            self.llm,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True
        )
        
        # 视频制作计划提示模板
        self.production_plan_prompt = PromptTemplate(
            input_variables=["script", "production_constraints"],
            template="""你是一位专业的视频制作专家，负责根据脚本制定详细的视频制作计划。
            
            脚本内容：
            {script}
            
            制作约束：
            {production_constraints}
            
            请制定一个详细的视频制作计划，包含以下内容：
            1. 素材需求清单：需要哪些图片、视频、音频素材
            2. 视频结构：场景顺序、转场效果
            3. 文本元素：需要添加的文本、字幕、标题，及其样式和位置
            4. 音频计划：背景音乐、音效、旁白的安排
            5. 制作步骤：详细的制作流程
            
            请以JSON格式返回你的制作计划。
            """
        )
    
    def _text_to_speech(self, text: str, lang: str = 'zh') -> str:
        """将文本转换为语音"""
        try:
            tts = gTTS(text=text, lang=lang, slow=False)
            filename = f"temp_audio_{datetime.now().strftime('%Y%m%d%H%M%S')}.mp3"
            filepath = os.path.join(self.output_path, filename)
            tts.save(filepath)
            return filepath
        except Exception as e:
            print(f"Error in text-to-speech: {e}")
            return ""
    
    def _search_materials(self, keyword: str, material_type: str = "all") -> List[str]:
        """搜索素材（模拟函数）"""
        # 在实际应用中，这里可以连接到素材库API或本地素材库
        # 这里我们返回模拟数据
        mock_materials = {
            "background": ["assets/background_videos/bg1.mp4", "assets/background_videos/bg2.mp4"],
            "music": ["assets/music/music1.mp3", "assets/music/music2.mp3"],
            "image": ["assets/images/img1.jpg", "assets/images/img2.jpg"]
        }
        
        if material_type == "all":
            results = []
            for materials in mock_materials.values():
                results.extend(materials)
            return results
        
        return mock_materials.get(material_type, [])
    
    def _create_text_image(self, text: str, size: tuple = (1080, 1920), 
                          font_size: int = 80, text_color: tuple = (255, 255, 255),
                          bg_color: tuple = (0, 0, 0, 180)) -> str:
        """创建包含文本的图片"""
        try:
            # 创建透明背景
            img = Image.new('RGBA', size, bg_color)
            draw = ImageDraw.Draw(img)
            
            # 加载字体（这里使用系统默认字体，实际应用中可以指定字体文件）
            try:
                font = ImageFont.truetype("arial.ttf", font_size)
            except:
                font = ImageFont.load_default()
            
            # 计算文本位置（居中）
            text_bbox = draw.textbbox((0, 0), text, font=font)
            text_width = text_bbox[2] - text_bbox[0]
            text_height = text_bbox[3] - text_bbox[1]
            x = (size[0] - text_width) // 2
            y = (size[1] - text_height) // 2
            
            # 绘制文本
            draw.text((x, y), text, font=font, fill=text_color)
            
            # 保存图片
            filename = f"text_image_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
            filepath = os.path.join(self.output_path, filename)
            img.save(filepath)
            return filepath
        except Exception as e:
            print(f"Error creating text image: {e}")
            return ""
    
    def create_production_plan(self, script: Dict, 
                              production_constraints: str = "使用现有的素材库，保持竖屏9:16格式，视频清晰流畅") -> Dict:
        """创建视频制作计划"""
        script_str = json.dumps(script, ensure_ascii=False)
        
        prompt = self.production_plan_prompt.format(
            script=script_str,
            production_constraints=production_constraints
        )
        
        try:
            result = self.agent.run(prompt)
            
            # 记录决策
            self.memory.record_decision(
                agent_name="ContentProducerAgent",
                decision_type="production_plan_creation",
                decision_content=result,
                context={
                    "script": script,
                    "production_constraints": production_constraints,
                    "timestamp": datetime.now().isoformat()
                }
            )
            
            # 尝试解析JSON结果
            try:
                plan = json.loads(result)
                return plan
            except json.JSONDecodeError:
                # 如果不是有效的JSON，返回原始结果
                return {"raw_result": result}
                
        except Exception as e:
            print(f"Error creating production plan: {e}")
            return {}
    
    def produce_video(self, script: Dict, production_plan: Dict = None) -> Optional[str]:
        """根据脚本和制作计划制作视频"""
        # 如果没有提供制作计划，先创建一个
        if not production_plan:
            production_plan = self.create_production_plan(script)
        
        try:
            # 这是一个简化的视频制作流程
            # 在实际应用中，你可能需要根据脚本和制作计划的具体结构来实现
            
            # 1. 准备素材
            # 这里我们使用一些简单的素材作为示例
            bg_video_path = os.path.join(self.background_videos_path, "sample_bg.mp4")
            
            # 如果没有示例背景视频，创建一个简单的纯色背景
            if not os.path.exists(bg_video_path):
                # 创建一个5秒的黑色背景视频
                bg_clip = ColorClip(size=(1080, 1920), color=(0, 0, 0