职贝云数AI新零售门户

标题: 如何用 Multi-Agent 自动运营抖音账号：从脚本到内容闭环 [打印本页]
作者: aZgrMu    时间: 前天 07:40
标题: 如何用 Multi-Agent 自动运营抖音账号：从脚本到内容闭环
如今，让我们逐一完成这些智能体和组件。
环境搭建和依赖安装

首先，让我们设置开发环境并安装必要的依赖：
  1. # 首先，创建一个虚拟环境（引荐）# python -m venv douyin_agent_env# source douyin_agent_env/bin/activate  # Linux/Mac# .\douyin_agent_env\Scripts\activate  # Windows# 安装必要的包
  2. !pip install langchain openai python-dotenv moviepy pillow pandas matplotlib sqlite3 requests schedule
复制代码
接上去，创建一个 .env 文件来存储我们的API密钥和配置：
  1. OPENAI_API_KEY=your_openai_api_key_here
  2. DOUYIN_CLIENT_KEY=your_douyin_client_key_here
  3. DOUYIN_CLIENT_SECRET=your_douyin_client_secret_here
复制代码
基础组件完成

首先，让我们完成一些基础组件，这些将被一切智能体运用。
记忆系统（Memory System）

记忆系统允许智能体记住过去的交互、决策和结果，这对于保持分歧性和持续学习至关重要。
  1. from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
  2. from langchain.schema import BaseMessage
  3. import sqlite3
  4. from datetime import datetime
  5. import json
  6. from typing import List, Dict, Any
  8. classMemorySystem:def__init__(self, db_path:str="douyin_agent_memory.db"):
  9.         self.db_path = db_path
  10.         self._init_db()
  11.         self.conversation_memory = ConversationBufferMemory(return_messages=True)
  12.         self.summary_memory = ConversationSummaryMemory(llm=None)# 我们会在运用时设置LLMdef_init_db(self):"""初始化数据库表"""
  13.         conn = sqlite3.connect(self.db_path)
  14.         cursor = conn.cursor()# 创建内容历史表
  15.         cursor.execute('''
  16.         CREATE TABLE IF NOT EXISTS content_history (
  17.             id INTEGER PRIMARY KEY AUTOINCREMENT,
  18.             content_id TEXT UNIQUE,
  19.             content_type TEXT,
  20.             title TEXT,
  21.             description TEXT,
  22.             publish_date TIMESTAMP,
  23.             performance_data TEXT,
  24.             created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  25.         )
  26.         ''')# 创建决策记录表
  27.         cursor.execute('''
  28.         CREATE TABLE IF NOT EXISTS decision_history (
  29.             id INTEGER PRIMARY KEY AUTOINCREMENT,
  30.             agent_name TEXT,
  31.             decision_type TEXT,
  32.             decision_content TEXT,
  33.             context TEXT,
  34.             outcome TEXT,
  35.             timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  36.         )
  37.         ''')# 创建用户互动表
  38.         cursor.execute('''
  39.         CREATE TABLE IF NOT EXISTS user_interactions (
  40.             id INTEGER PRIMARY KEY AUTOINCREMENT,
  41.             interaction_type TEXT,
  42.             user_id TEXT,
  43.             content_id TEXT,
  44.             interaction_content TEXT,
  45.             response TEXT,
  46.             timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
  47.         )
  48.         ''')
  49.         
  50.         conn.commit()
  51.         conn.close()defadd_conversation_message(self, message: BaseMessage):"""添加对话音讯到记忆"""
  52.         self.conversation_memory.chat_memory.add_message(message)defget_conversation_history(self)-> List[BaseMessage]:"""获取对话历史"""return self.conversation_memory.chat_memory.messages
  53.    
  54.     defrecord_content(self, content_id:str, content_type:str, title:str,
  55.                       description:str, publish_date: datetime =None,
  56.                       performance_data: Dict =None):"""记录内容发布信息"""
  57.         conn = sqlite3.connect(self.db_path)
  58.         cursor = conn.cursor()try:
  59.             cursor.execute('''
  60.             INSERT OR REPLACE INTO content_history
  61.             (content_id, content_type, title, description, publish_date, performance_data)
  62.             VALUES (?, ?, ?, ?, ?, ?)
  63.             ''',(
  64.                 content_id,
  65.                 content_type,
  66.                 title,
  67.                 description,
  68.                 publish_date.isoformat()if publish_date elseNone,
  69.                 json.dumps(performance_data)if performance_data elseNone))
  70.             conn.commit()except Exception as e:print(f"Error recording content: {e}")finally:
  71.             conn.close()defrecord_decision(self, agent_name:str, decision_type:str,
  72.                        decision_content:str, context: Dict =None,
  73.                        outcome:str=None):"""记录智能体决策"""
  74.         conn = sqlite3.connect(self.db_path)
  75.         cursor = conn.cursor()try:
  76.             cursor.execute('''
  77.             INSERT INTO decision_history
  78.             (agent_name, decision_type, decision_content, context, outcome)
  79.             VALUES (?, ?, ?, ?, ?)
  80.             ''',(
  81.                 agent_name,
  82.                 decision_type,
  83.                 decision_content,
  84.                 json.dumps(context)if context elseNone,
  85.                 outcome
  86.             ))
  87.             conn.commit()except Exception as e:print(f"Error recording decision: {e}")finally:
  88.             conn.close()defrecord_user_interaction(self, interaction_type:str, user_id:str,
  89.                                content_id:str, interaction_content:str,
  90.                                response:str=None):"""记录用户互动"""
  91.         conn = sqlite3.connect(self.db_path)
  92.         cursor = conn.cursor()try:
  93.             cursor.execute('''
  94.             INSERT INTO user_interactions
  95.             (interaction_type, user_id, content_id, interaction_content, response)
  96.             VALUES (?, ?, ?, ?, ?)
  97.             ''',(
  98.                 interaction_type,
  99.                 user_id,
  100.                 content_id,
  101.                 interaction_content,
  102.                 response
  103.             ))
  104.             conn.commit()except Exception as e:print(f"Error recording user interaction: {e}")finally:
  105.             conn.close()defget_content_history(self, limit:int=10)-> List[Dict]:"""获取内容历史"""
  106.         conn = sqlite3.connect(self.db_path)
  107.         cursor = conn.cursor()
  108.         
  109.         cursor.execute('''
  110.         SELECT content_id, content_type, title, description, publish_date, performance_data
  111.         FROM content_history
  112.         ORDER BY publish_date DESC
  113.         LIMIT ?
  114.         ''',(limit,))
  115.         
  116.         results = cursor.fetchall()
  117.         conn.close()
  118.         
  119.         content_list =[]for row in results:
  120.             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]elseNone})return content_list
  121.    
  122.     defget_top_performing_content(self, metric:str='view_count', limit:int=5)-> List[Dict]:"""获取表现最好的内容"""
  123.         conn = sqlite3.connect(self.db_path)
  124.         cursor = conn.cursor()
  125.         
  126.         cursor.execute('''
  127.         SELECT content_id, content_type, title, description, publish_date, performance_data
  128.         FROM content_history
  129.         WHERE performance_data IS NOT NULL
  130.         ORDER BY publish_date DESC
  131.         LIMIT 100
  132.         ''')
  133.         
  134.         results = cursor.fetchall()
  135.         conn.close()# 解析功能数据并排序
  136.         content_list =[]for row in results:
  137.             perf_data = json.loads(row[5])if row[5]else{}if metric in perf_data:
  138.                 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]})# 按目的值排序
  139.         content_list.sort(key=lambda x: x['metric_value'], reverse=True)return content_list[:limit]
复制代码
工具库（Tool Library）

接上去，让我们创建一个工具库，包含各种智能体能够需求运用的工具：
  1. import requests
  2. import json
  3. from datetime import datetime, timedelta
  4. from typing import List, Dict, Any, Optional
  5. import os
  6. from dotenv import load_dotenv
  8. # 加载环境变量
  9. load_dotenv()classToolLibrary:def__init__(self):
  10.         self.douyin_api_base ="https://open.douyin.com"
  11.         self.douyin_client_key = os.getenv("DOUYIN_CLIENT_KEY")
  12.         self.douyin_client_secret = os.getenv("DOUYIN_CLIENT_SECRET")
  13.         self.access_token =None
  14.         self._refresh_access_token()def_refresh_access_token(self):"""刷新抖音API访问令牌"""
  15.         url =f"{self.douyin_api_base}/oauth/client_token/"
  16.         params ={"client_key": self.douyin_client_key,"client_secret": self.douyin_client_secret,"grant_type":"client_credential"}try:
  17.             response = requests.get(url, params=params)
  18.             response.raise_for_status()
  19.             data = response.json()if data.get("data"):
  20.                 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}")defget_trending_topics(self, category:str="general")-> List[Dict]:"""获取当前抢手话题"""# 在实践运用中，这里可以衔接到抖音的热点API或第三方热点服务# 这里我们运用模拟数据作为示例
  21.         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
  22.    
  23.     defsearch_content_ideas(self, keyword:str, limit:int=10)-> List[Dict]:"""搜索内容创意"""# 在实践运用中，这里可以衔接到抖音的搜索API或第三方内容分析服务# 这里我们运用模拟数据作为示例
  24.         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]defget_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}defget_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"# 表现趋向}defget_comments(self, video_id:str, limit:int=50)-> List[Dict]:"""获取视频评论"""# 在实践运用中，这里应该调用抖音API# 这里我们运用模拟数据
  25.         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]defreply_to_comment(self, comment_id:str, reply_content:str)->bool:"""回复评论"""# 在实践运用中，这里应该调用抖音APIprint(f"Replying to comment {comment_id}: {reply_content}")returnTrue# 假设回复成功defupload_video(self, video_path:str, title:str, description:str,
  26.                    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
  27.         video_id =f"video_{datetime.now().strftime('%Y%m%d%H%M%S')}"return video_id
  28.    
  29.     defschedule_video_publish(self, video_id:str, publish_time: datetime)->bool:"""安排视频发布工夫"""# 在实践运用中，这里应该调用抖音APIprint(f"Scheduling video {video_id} to be published at {publish_time}")returnTrue# 假设操作成功
复制代码
智能体完成

如今，让我们末尾完成各个智能体。我们将运用LangChain框架来构建这些智能体。
创意总监Agent（Creative Director Agent）

创意总监Agent担任生成内容创意和选题方向。它会分析抢手话题、账号历史表现和粉丝兴味，提出有潜力的内容创意。
  1. from langchain.llms import OpenAI
  2. from langchain.prompts import PromptTemplate
  3. from langchain.chains import LLMChain
  4. from langchain.agents import AgentType, initialize_agent, Tool
  5. from typing import List, Dict, Any
  6. import json
  7. from datetime import datetime
  9. classCreativeDirectorAgent:def__init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
  10.         self.memory = memory_system
  11.         self.tools = tool_library
  12.         self.llm = OpenAI(temperature=0.7, max_tokens=2000)# 较高的温度以获得更多创意# 定义创意总监运用的工具
  13.         self.agent_tools =[
  14.             Tool(
  15.                 name="获取抢手话题",
  16.                 func=self.tools.get_trending_topics,
  17.                 description="获取当前抖音上的抢手话题，可用于内容创意"),
  18.             Tool(
  19.                 name="搜索内容创意",
  20.                 func=lambda keyword: json.dumps(self.tools.search_content_ideas(keyword), ensure_ascii=False),
  21.                 description="基于关键词搜索相关的内容创意"),
  22.             Tool(
  23.                 name="获取账号历史内容",
  24.                 func=lambda limit=10: json.dumps(self.memory.get_content_history(limit), ensure_ascii=False),
  25.                 description="获取账号历史发布的内容，用于参考和避免反复"),
  26.             Tool(
  27.                 name="获取高表现内容",
  28.                 func=lambda metric='view_count', limit=5: json.dumps(self.memory.get_top_performing_content(metric, limit), ensure_ascii=False),
  29.                 description="获取账号历史上表现最好的内容，分析成功要素")]# 创建创意总监Agent
  30.         self.agent = initialize_agent(
  31.             self.agent_tools,
  32.             self.llm,
  33.             agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
  34.             verbose=True)# 创意生成提示模板
  35.         self.creative_prompt = PromptTemplate(
  36.             input_variables=["account_data","content_objectives","constraints"],
  37.             template="""你是一位专业的抖音内容创意总监，担任为账号制定内容策略和生成创意选题。
  38.             
  39.             账号数据：
  40.             {account_data}
  41.             
  42.             内容目的：
  43.             {content_objectives}
  44.             
  45.             约束条件：
  46.             {constraints}
  47.             
  48.             请按照以下步骤停止思索和工作：
  49.             1. 首先，了解当前的抢手话题和趋向
  50.             2. 查看账号历史表现最好的内容，分析成功要素
  51.             3. 确保新创意与账号定位分歧，同时避免与历史内容反复
  52.             4. 生成5-7个有潜力的内容创意，每个创意包含：标题、类型、简要描画、预期效果、预估难度
  53.             
  54.             请以JSON格式前往你的创意建议。
  55.             """)defgenerate_content_ideas(self, content_objectives:str="添加粉丝互动和账号曝光",
  56.                               constraints:str="保持账号垂直范畴定位，每视频时长控制在15-60秒")-> List[Dict]:"""生成内容创意"""
  57.         account_data = self.tools.get_account_data()
  58.         account_data_str = json.dumps(account_data, ensure_ascii=False)
  59.         
  60.         prompt = self.creative_prompt.format(
  61.             account_data=account_data_str,
  62.             content_objectives=content_objectives,
  63.             constraints=constraints
  64.         )try:
  65.             result = self.agent.run(prompt)# 记录决策
  66.             self.memory.record_decision(
  67.                 agent_name="CreativeDirectorAgent",
  68.                 decision_type="content_ideas_generation",
  69.                 decision_content=result,
  70.                 context={"content_objectives": content_objectives,"constraints": constraints,"timestamp": datetime.now().isoformat()})# 尝试解析JSON结果try:
  71.                 ideas = json.loads(result)return ideas ifisinstance(ideas,list)else[ideas]except json.JSONDecodeError:# 假如不是有效的JSON，前往原始结果return[{"raw_result": result}]except Exception as e:print(f"Error generating content ideas: {e}")return[]defrefine_content_idea(self, idea: Dict, feedback:str)-> Dict:"""根据反馈优化内容创意"""
  72.         refine_prompt =f"""
  73.         原始创意：
  74.         {json.dumps(idea, ensure_ascii=False)}
  75.         
  76.         反馈意见：
  77.         {feedback}
  78.         
  79.         请根据反馈意见优化这个内容创意，保留其核心价值的同时，处理反馈中提到的成绩。
  80.         请以JSON格式前往优化后的创意。
  81.         """try:
  82.             result = self.agent.run(refine_prompt)# 记录决策
  83.             self.memory.record_decision(
  84.                 agent_name="CreativeDirectorAgent",
  85.                 decision_type="content_idea_refinement",
  86.                 decision_content=result,
  87.                 context={"original_idea": idea,"feedback": feedback,"timestamp": datetime.now().isoformat()})try:
  88.                 refined_idea = json.loads(result)return refined_idea
  89.             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担任将创意转化为可拍摄的详细脚本，包括场景描画、台词、镜头建议等。
  1. from langchain.llms import OpenAI
  2. from langchain.prompts import PromptTemplate
  3. from langchain.chains import LLMChain
  4. from langchain.agents import AgentType, initialize_agent, Tool
  5. from typing import List, Dict, Any
  6. import json
  7. from datetime import datetime
  9. classScriptWriterAgent:def__init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
  10.         self.memory = memory_system
  11.         self.tools = tool_library
  12.         self.llm = OpenAI(temperature=0.8, max_tokens=3000)# 较高的温度以获得更有创意的脚本# 定义剧本作家运用的工具
  13.         self.agent_tools =[
  14.             Tool(
  15.                 name="获取高表现内容",
  16.                 func=lambda metric='view_count', limit=5: json.dumps(self.memory.get_top_performing_content(metric, limit), ensure_ascii=False),
  17.                 description="获取账号历史上表现最好的内容，分析其脚本结构和风格"),
  18.             Tool(
  19.                 name="搜索内容创意",
  20.                 func=lambda keyword: json.dumps(self.tools.search_content_ideas(keyword), ensure_ascii=False),
  21.                 description="基于关键词搜索相关的内容创意，为脚本提供更多灵感")]# 创建剧本作家Agent
  22.         self.agent = initialize_agent(
  23.             self.agent_tools,
  24.             self.llm,
  25.             agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
  26.             verbose=True)# 脚本写作提示模板
  27.         self.script_prompt = PromptTemplate(
  28.             input_variables=["content_idea","account_style","duration_constraint"],
  29.             template="""你是一位专业的抖音短视频编剧，擅长创作引人入胜、合适平台算法的短视频脚本。
  30.             
  31.             内容创意：
  32.             {content_idea}
  33.             
  34.             账号风格定位：
  35.             {account_style}
  36.             
  37.             时长约束：
  38.             {duration_constraint}
  39.             
  40.             请按照以下步骤停止思索和工作：
  41.             1. 首先，分析这个内容创意的核心价值和吸引点
  42.             2. 查看账号历史上表现最好的内容，分析其成功的脚本结构
  43.             3. 创作一个残缺的短视频脚本，包含以下元素：
  44.                - 视频概述：简要描画视频内容和目的
  45.                - 钩子设计：前3秒如何吸引观众留意力
  46.                - 详细脚本：按工夫线分解，包含场景、画面描画、台词/旁白、音乐/音效建议
  47.                - 互动设计：如何引导用户点赞、评论、分享
  48.                - 话题标签建议：相关的抢手话题和垂直范畴标签
  49.             
  50.             请以JSON格式前往你的脚本。
  51.             """)defwrite_script(self, content_idea: Dict,
  52.                     account_style:str="知识科普类，专业但不失兴趣，语速适中，画面明晰",
  53.                     duration_constraint:str="15-30秒")-> Dict:"""编写脚本"""
  54.         content_idea_str = json.dumps(content_idea, ensure_ascii=False)
  55.         
  56.         prompt = self.script_prompt.format(
  57.             content_idea=content_idea_str,
  58.             account_style=account_style,
  59.             duration_constraint=duration_constraint
  60.         )try:
  61.             result = self.agent.run(prompt)# 记录决策
  62.             self.memory.record_decision(
  63.                 agent_name="ScriptWriterAgent",
  64.                 decision_type="script_writing",
  65.                 decision_content=result,
  66.                 context={"content_idea": content_idea,"account_style": account_style,"duration_constraint": duration_constraint,"timestamp": datetime.now().isoformat()})# 尝试解析JSON结果try:
  67.                 script = json.loads(result)return script
  68.             except json.JSONDecodeError:# 假如不是有效的JSON，前往原始结果return{"raw_result": result}except Exception as e:print(f"Error writing script: {e}")return{}defrefine_script(self, script: Dict, feedback:str)-> Dict:"""根据反馈优化脚本"""
  69.         refine_prompt =f"""
  70.         原始脚本：
  71.         {json.dumps(script, ensure_ascii=False)}
  72.         
  73.         反馈意见：
  74.         {feedback}
  75.         
  76.         请根据反馈意见优化这个脚本，保留其核心价值的同时，处理反馈中提到的成绩。
  77.         请以JSON格式前往优化后的脚本。
  78.         """try:
  79.             result = self.agent.run(refine_prompt)# 记录决策
  80.             self.memory.record_decision(
  81.                 agent_name="ScriptWriterAgent",
  82.                 decision_type="script_refinement",
  83.                 decision_content=result,
  84.                 context={"original_script": script,"feedback": feedback,"timestamp": datetime.now().isoformat()})try:
  85.                 refined_script = json.loads(result)return refined_script
  86.             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担任根据脚本制造视频。它可以运用模板、素材库、文本转语音等工具来自动化生成视频。
  1. from langchain.llms import OpenAI
  2. from langchain.prompts import PromptTemplate
  3. from langchain.chains import LLMChain
  4. from langchain.agents import AgentType, initialize_agent, Tool
  5. from typing import List, Dict, Any, Optional
  6. import json
  7. from datetime import datetime
  8. import os
  9. from moviepy.editor import*from gtts import gTTS
  10. import requests
  11. from PIL import Image, ImageDraw, ImageFont
  13. classContentProducerAgent:def__init__(self, memory_system: MemorySystem, tool_library: ToolLibrary):
  14.         self.memory = memory_system
  15.         self.tools = tool_library
  16.         self.llm = OpenAI(temperature=0.3, max_tokens=1500)# 较低的温度以获得更准确的执行# 素材库途径
  17.         self.background_videos_path ="assets/background_videos"
  18.         self.music_path ="assets/music"
  19.         self.images_path ="assets/images"
  20.         self.output_path ="output/videos"# 确保目录存在for path in[self.background_videos_path, self.music_path, self.images_path, self.output_path]:
  21.             os.makedirs(path, exist_ok=True)# 定义内容制造运用的工具
  22.         self.agent_tools =[
  23.             Tool(
  24.                 name="文本转语音",
  25.                 func=self._text_to_speech,
  26.                 description="将文本转换为语音，参数为文本内容和可选的言语代码"),
  27.             Tool(
  28.                 name="搜索素材",
  29.                 func=self._search_materials,
  30.                 description="根据关键词搜索相关的素材（图片、视频、音乐）")]# 创建内容制造Agent
  31.         self.agent = initialize_agent(
  32.             self.agent_tools,
  33.             self.llm,
  34.             agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
  35.             verbose=True)# 视频制造计划提示模板
  36.         self.production_plan_prompt = PromptTemplate(
  37.             input_variables=["script","production_constraints"],
  38.             template="""你是一位专业的视频制造专家，担任根据脚本制定详细的视频制造计划。
  39.             
  40.             脚本内容：
  41.             {script}
  42.             
  43.             制造约束：
  44.             {production_constraints}
  45.             
  46.             请制定一个详细的视频制造计划，包含以下内容：
  47.             1. 素材需求清单：需求哪些图片、视频、音频素材
  48.             2. 视频结构：场景顺序、转场效果
  49.             3. 文本元素：需求添加的文本、字幕、标题，及其样式和地位
  50.             4. 音频计划：背景音乐、音效、旁白的安排
  51.             5. 制造步骤：详细的制造流程
  52.             
  53.             请以JSON格式前往你的制造计划。
  54.             """)def_text_to_speech(self, text:str, lang:str='zh')->str:"""将文本转换为语音"""try:
  55.             tts = gTTS(text=text, lang=lang, slow=False)
  56.             filename =f"temp_audio_{datetime.now().strftime('%Y%m%d%H%M%S')}.mp3"
  57.             filepath = os.path.join(self.output_path, filename)
  58.             tts.save(filepath)return filepath
  59.         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或本地素材库# 这里我们前往模拟数据
  60.         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":
  61.             results =[]for materials in mock_materials.values():
  62.                 results.extend(materials)return results
  63.         
  64.         return mock_materials.get(material_type,[])def_create_text_image(self, text:str, size:tuple=(1080,1920),
  65.                           font_size:int=80, text_color:tuple=(255,255,255),
  66.                           bg_color:tuple=(0,0,0,180))->str:"""创建包含文本的图片"""try:# 创建透明背景
  67.             img = Image.new('RGBA', size, bg_color)
  68.             draw = ImageDraw.Draw(img)# 加载字体（这里运用系统默许字体，实践运用中可以指定字体文件）try:
  69.                 font = ImageFont.truetype("arial.ttf", font_size)except:
  70.                 font = ImageFont.load_default()# 计算文本地位（居中）
  71.             text_bbox = draw.textbbox((0,0), text, font=font)
  72.             text_width = text_bbox[2]- text_bbox[0]
  73.             text_height = text_bbox[3]- text_bbox[1]
  74.             x =(size[0]- text_width)//2
  75.             y =(size[1]- text_height)//2# 绘制文本
  76.             draw.text((x, y), text, font=font, fill=text_color)# 保存图片
  77.             filename =f"text_image_{datetime.now().strftime('%Y%m%d%H%M%S')}.png"
  78.             filepath = os.path.join(self.output_path, filename)
  79.             img.save(filepath)return filepath
  80.         except Exception as e:print(f"Error creating text image: {e}")return""defcreate_production_plan(self, script: Dict,
  81.                               production_constraints:str="运用现有的素材库，保持竖屏9:16格式，视频明晰流利")-> Dict:"""创建视频制造计划"""
  82.         script_str = json.dumps(script, ensure_ascii=False)
  83.         
  84.         prompt = self.production_plan_prompt.format(
  85.             script=script_str,
  86.             production_constraints=production_constraints
  87.         )try:
  88.             result = self.agent.run(prompt)# 记录决策
  89.             self.memory.record_decision(
  90.                 agent_name="ContentProducerAgent",
  91.                 decision_type="production_plan_creation",
  92.                 decision_content=result,
  93.                 context={"script": script,"production_constraints": production_constraints,"timestamp": datetime.now().isoformat()})# 尝试解析JSON结果try:
  94.                 plan = json.loads(result)return plan
  95.             except json.JSONDecodeError:# 假如不是有效的JSON，前往原始结果return{"raw_result": result}except Exception as e:print(f"Error creating production plan: {e}")return{}defproduce_video(self, script: Dict, production_plan: Dict =None)-> Optional[str]:"""根据脚本和制造计划制造视频"""# 假如没有提供制造计划，先创建一个ifnot production_plan:
  96.             production_plan = self.create_production_plan(script)try:# 这是一个简化的视频制造流程# 在实践运用中，你能够需求根据脚本和制造计划的详细结构来完成# 1. 预备素材# 这里我们运用一些简单的素材作为示例
  97.             bg_video_path = os.path.join(self.background_videos_path,"sample_bg.mp4")# 假如没有示例背景视频，创建一个简单的纯色背景ifnot os.path.exists(bg_video_path):# 创建一个5秒的黑色背景视频
  98.                 bg_clip = ColorClip(size=(1080,1920), color=(0,0,0
复制代码




欢迎光临 职贝云数AI新零售门户 (https://www.taojin168.com/cloud/) Powered by Discuz! X3.5