import asyncio from datetime import datetime import re import aiofiles import os import sys from typing import List, Dict, Any from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel from typing import List from llama_index.core.program import LLMTextCompletionProgram from llama_index.core.output_parsers import PydanticOutputParser from llama_index.program.lmformatenforcer import ( LMFormatEnforcerPydanticProgram, ) from llama_index.core.prompts import PromptTemplate from llama_index.llms.llama_cpp import LlamaCPP from llama_index.llms.openai import OpenAI from llama_index.llms.litellm import LiteLLM from llama_index.core.llms.llm import LLM from src.models.product_model import Product from src.manager.template_manager import TemplateManager, TemplateService, TemplateType from src.models.ai_execution_record import MarketingInfo, LLMConfig, SuperPromptMixin, AgentConfig, AgentContent, AICompetitorAnalyzeMainKeywords, AICompetitorAnalyzeMainKeywordsResult, MarketingContentGeneration from ai.base_agent import BaseAgent from utils.file import save_to_file, read_file from config.settings import MONGO_URL, MONGO_DB_NAME,LITELLM_API_BASE, LITELLM_API_KEY,OPENAI_API_KEY,OPENAI_API_BASE from utils.logu import get_logger logger = get_logger("ai") class MarketingAgent(BaseAgent): async def get_mainkeys_tailkeys_prompt(self, product_name, prompt: str='',output_type='markdown', verbose=False) -> PromptTemplate: base_prompt = "```\n{{product_info}}\n{{competitor_info}}\n```\n" prompt_mainkyes = prompt or '''\ 你是日本站的亚马逊运营，请你根据产品信息为用户选出主要关键词和长尾关键词。 请根据竞品数据，按以下规则分析： - 选出搜索量在1万以上的相同关键词作为主要关键词3个。 - 如果竞品的搜索量都不足1万，则从排名前十的关键词中筛选3个搜索量最大且相关性最强的词。 - 结合日本市场特点分析 - 根据我的产品基本信息，从竞品的主要信息和同类竞品的相似关键词中，筛选出最符合我产品的长尾关键词 10个以上 筛选长尾词的示例： - 假设我的产品是电线保护，那么竞品关键词中，“隐藏排线管” 就不符合长尾词 - 假设我的产品是“防老化、防动物咬”用途，你就不能在竞品数据中选择不属于我这个使用场景的长尾关键词。 生成的内容满足以下要求： - reason 、 suggestions 必须写成中文 - monthly_searches 必须是 int ，按照从大到小排序，别的字段按照源数据填写即可。 - 内容格式必须是 {output_type} ''' variables = {'product_name': product_name} product_info = await self.template_manager.execute_template("product_info", variables) competitor_info = await self.template_manager.execute_template("competitor_for_llm", variables) prompt_tmpl = PromptTemplate(template=base_prompt + prompt_mainkyes, ) return prompt_tmpl.partial_format( product_info=product_info, competitor_info=competitor_info, output_type=output_type, ) def llm_mainkeys_tailkeys_to_model(self, prompt_template:PromptTemplate, prompt_kwargs:Dict={}, verbose=False): program = LLMTextCompletionProgram.from_defaults( output_parser=PydanticOutputParser(output_cls=MarketingInfo), prompt_template_str=prompt_template.format(prompt_kwargs), llm=self.llm, verbose=verbose, ) competitor = program(**prompt_kwargs) logger.info(f"{competitor}") return competitor async def gen_mainkeys_tailkeys(self, product_name, prompt: str='',output_type='markdown', verbose=False, overwrite=False): logger.info(f"start llm model: {self.llm._get_model_name()}, output_type: {output_type}") agent_model = await AgentContent.find_one(AgentContent.model_name == self.llm._get_model_name(),AgentContent.product_name == product_name) if agent_model and not overwrite: logger.info(f"agent_model exist {agent_model}") return agent_model elif not agent_model: agent_model = AgentContent(model_name=self.llm._get_model_name(), product_name=product_name) prompt_template = await self.get_mainkeys_tailkeys_prompt(product_name, prompt,output_type, verbose) logger.info(f"{prompt_template.format()}") if output_type == 'json': competitor = self.llm_mainkeys_tailkeys_to_model(prompt_template, verbose) elif output_type == 'markdown': response = await self.llm.acomplete(prompt_template.format()) pattern = r'```markdown(.*?)```' matches = re.findall(pattern, response.text, re.DOTALL) if not matches: competitor = response.text else: competitor = matches[0] agent_model.product = await Product.find_one(Product.basic_info.name == product_name) agent_model.competitor[output_type] = competitor agent_model.update_time = datetime.now() return await agent_model.save() async def get_marketing_prompt(self, product_name, prompt: str='', verbose=False, llm=None): prompt_marketing = prompt or '''\ 你是日本站的亚马逊运营，请你根据产品信息、竞品相似主关键词、竞品长尾词，为用户生成营销文案和5个卖点文案，需要用日文。 请根据竞品数据，按以下规则分析： - 选出搜索量在1万以上的相同关键词作为主要关键词3个。 - 如果竞品的搜索量都不足1万，则从排名前十的关键词中筛选3个搜索量最大且相关性最强的词。 - 结合日本市场特点分析 - 根据我的产品基本信息，从竞品的主要信息和同类竞品的相似关键词中，筛选出最符合我产品的长尾关键词 10个以上 筛选长尾词的示例： - 假设我的产品是电线保护，那么竞品关键词中，“隐藏排线管” 就不符合长尾词 - 假设我的产品是“防老化、防动物咬”用途，你就不能在竞品数据中选择不属于我这个使用场景的长尾关键词。 生成的内容满足以下要求： - reason 、 suggestions 必须写成中文 - monthly_searches 必须是 int ，按照从大到小排序，别的字段按照源数据填写即可。 ''' all_keywords = self.template_manager.execute_template('agent.mainkeys_tailkeys') variables = {'product_name': product_name} product_info = await self.template_manager.execute_template("product_info", variables) program = LMFormatEnforcerPydanticProgram( output_cls=MarketingInfo, llm=llm or self.llm, prompt_template_str="{{product_info}}\n{{all_keywords}}\n" + prompt_marketing, verbose=verbose, ) competitor = program(product_info=product_info, all_keywords=all_keywords) return competitor async def gen_marketing_file(self, product_name, output_path: str, llm_models: List[str] = []): models = await AgentContent.find(AgentContent.product_name == product_name).to_list() # 创建一个字典来存储模型及其基础名称 model_dict = {} unsorted_models = [] for model in models: # 提取基础名称（去掉前面的xxx/xxx） base_name = model.model_name.split('/')[-1] if '/' in model.model_name else model.model_name # 检查是否在优先级列表中 found = False for priority_model in llm_models: priority_base = priority_model.split('/')[-1] if '/' in priority_model else priority_model if base_name == priority_base: model_dict[priority_model] = (model, base_name) # 存储模型和显示名称 found = True break if not found: # 对于不在列表中的模型，也存储基础名称 unsorted_models.append((model, base_name)) # 按照llm_models顺序生成内容 sorted_content = '' for priority_model in llm_models: if priority_model in model_dict: model, display_name = model_dict[priority_model] markdown = model.competitor.get('markdown', '') logger.info(f"llm_name: {model.model_name} , {markdown[:100]}") sorted_content += f"# {display_name}\n{markdown}\n\n" # 使用基础名称 # 添加未排序的模型 unsorted_content = '' for model, display_name in unsorted_models: markdown = model.competitor.get('markdown', '') logger.info(f"llm_name: {model.model_name} , {markdown[:100]}") unsorted_content += f"# {display_name}\n{markdown}\n\n" # 使用基础名称 prompt_template = await self.get_mainkeys_tailkeys_prompt(product_name, output_type='markdown') content = sorted_content + unsorted_content + f'# 提示词 \n{prompt_template.format()}\n\n' return save_to_file(content, output_path) async def llm_task(product_name): m = TemplateManager(MONGO_URL, MONGO_DB_NAME) await m.initialize() model = 'openai/groq/llama-3.1-8b-instant' # model = 'groq/groq/qwen-2.5-coder-32b' # model = 'openai/glm-4-flash' # model = 'openai/deepseek-v3' # model = 'openai/groq/qwen-2.5-32b' # model = 'openai/deepseek-chat' model = 'openai/deepseek-reasoner' # model = 'openai/doubao-pro-32k-241215' llm_list = [ # 'openai/doubao-pro-32k-241215', LiteLLM(model='openai/doubao-pro-32k-241215', api_key=OPENAI_API_KEY, api_base=OPENAI_API_BASE), # 'openai/deepseek-reasoner', LiteLLM(model='openai/deepseek-reasoner', api_key=OPENAI_API_KEY, api_base=OPENAI_API_BASE), # 'openai/deepseek-v3', LiteLLM(model='openai/deepseek-v3', api_key=OPENAI_API_KEY, api_base=OPENAI_API_BASE), # 'openai/QwQ-32B', ] task_list = [] for llm in llm_list: agent = MarketingAgent(llm=llm, template_manager=m) agent_model = agent.gen_mainkeys_tailkeys(product_name=product_name, verbose=True, overwrite=True) task_list.append(agent_model) # logger.info(f"{agent_model.competitor.items()}") await asyncio.gather(*task_list) await gen_marketing_file(product_name) # llm = LiteLLM(model=model, api_key=OPENAI_API_KEY, api_base=OPENAI_API_BASE) # agent = MarketingAgent(llm=llm, template_manager=m) # agent_model = await agent.gen_mainkeys_tailkeys(product_name='大尺寸厚款卸妆棉240片', verbose=True, overwrite=True) # logger.info(f"{agent_model.competitor.items()}") async def gen_marketing_file(product_name): m = TemplateManager(MONGO_URL, MONGO_DB_NAME) await m.initialize() model = 'openai/deepseek-reasoner' llm = LiteLLM(model=model, api_key=OPENAI_API_KEY, api_base=OPENAI_API_BASE) # product_name = '大尺寸厚款卸妆棉240片' agent = MarketingAgent(llm=llm, template_manager=m) output_path = r'G:\code\amazone\copywriting_production\output\temp' + f"\\{product_name}-营销文案.md" llm_models = [ 'openai/doubao-pro-32k-241215', 'openai/deepseek-reasoner', 'openai/deepseek-v3', # 'openai/QwQ-32B', ] await agent.gen_marketing_file(product_name=product_name, output_path=output_path, llm_models=llm_models) logger.info(f"{output_path}") def main(): product_name = '狗刷牙指套' # product_name = '园艺镊子套装2件套' asyncio.run(llm_task(product_name=product_name)) if __name__ == "__main__": main()