登陆 注册
职贝云数AI新零售门户 › 新媒体矩阵玩法 › 抖音私域 › 如何用 Multi-Agent 自动运营抖音账号：从脚本到内容闭 ...
美业门店全域流量运营系统拆解：三步骤构建

在消费互联网流量增长放缓的背景下，线下服务业的数字化转型进入新阶段。对于高度分散、依赖人工的美业门店 详细

返回列表 发新帖
开启左侧

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

[复制链接]
在线会员 aZgrMu 发表于 前天 07:40 | 显示全部楼层 |阅读模式 打印 上一主题 下一主题
现在，让咱们一一完毕那些智能体战组件。
情况拆修战依靠装置

起首，让咱们树立开辟情况并装置须要的依靠：
  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, ConversationSu妹妹aryMemory
  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.su妹妹ary_memory = ConversationSu妹妹aryMemory(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.co妹妹it()
  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.co妹妹it()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.co妹妹it()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.co妹妹it()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_co妹妹ents_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,"co妹妹ent_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_co妹妹ents(self, video_id:str, limit:int=50)-> List[Dict]:"""获得望频批评"""# 正在理论使用中，那里该当挪用抖音API# 那里咱们使用模仿数据
  25.         mock_co妹妹ents =[{"co妹妹ent_id":"1","user_id":"user1","content":"那个望频太有效了！","like_count":45,"timestamp":"2023-10-01T10:30:00"},{"co妹妹ent_id":"2","user_id":"user2","content":"叨教第两步如何干？","like_count":12,"timestamp":"2023-10-01T11:15:00"},{"co妹妹ent_id":"3","user_id":"user3","content":"已经珍藏，目前垂垂瞅","like_count":8,"timestamp":"2023-10-01T12:45:00"},{"co妹妹ent_id":"4","user_id":"user4","content":"没有附和您的概念，尔觉得...","like_count":23,"timestamp":"2023-10-01T14:20:00"},{"co妹妹ent_id":"5","user_id":"user5","content":"能不克不及出一期更具体的？","like_count":31,"timestamp":"2023-10-01T16:05:00"}]return mock_co妹妹ents[:limit]defreply_to_co妹妹ent(self, co妹妹ent_id:str, reply_content:str)->bool:"""复兴批评"""# 正在理论使用中，那里该当挪用抖音APIprint(f"Replying to co妹妹ent {co妹妹ent_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
复造代码
回复

使用道具 举报

返回列表 发表新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册 qq_login

本版积分规则

avatar
aZgrMu注册会员

关注0

粉丝0

帖子17

发布主题
推荐阅读更多+
阅读排行更多+
用专业创造成效
400-778-7781
周一至周五 9:00-18:00
意见反馈：server@mailiao.group
紧急联系：181-67184787
ftqrcode

扫一扫关注我们

Powered by 职贝云数A新零售门户 X3.5© 2004-2025 职贝云数 Inc.( 蜀ICP备2024104722号 )