diff --git a/maternal_agent.py b/maternal_agent.py new file mode 100644 index 0000000000000000000000000000000000000000..a2dc1949fa12e58682f10474062f3dc6345ec053 --- /dev/null +++ b/maternal_agent.py @@ -0,0 +1,688 @@ +import os +import json +from datetime import datetime +from lazyllm import OnlineChatModule, WebModule, pipeline, TrainableModule +import re + +# 母婴类选品智能助手Agent +class MaternalAgent: + def __init__(self): + # 初始化LazyLLM聊天模块 + try: + # 首先尝试使用OnlineChatModule + self.chat_module = self._init_local_model() + print(f"成功加载模型") + except Exception as e: + print(f"无法初始化模型: {e}") + try: + # 如果模型加载失败,尝试在线模型 + self.chat_module = OnlineChatModule() + print("成功加载在线模型") + except Exception as e2: + print(f"无法初始化在线模型: {e2}") + print("请设置API密钥,例如:export OPENAI_API_KEY=your_key") + # 使用模拟的聊天模块作为后备 + self.chat_module = self._create_mock_chat_module() + print("使用模拟聊天模块作为后备方案") + + # 母婴产品类别数据 + self.product_categories = { + "喂养用品": { + "subcategories": ["奶瓶", "奶嘴", "奶粉盒", "吸奶器", "辅食机", "餐具", "水杯"], + "features": ["安全性", "材质", "易用性", "清洁方便"] + }, + "护理用品": { + "subcategories": ["纸尿裤", "湿纸巾", "爽身粉", "护臀膏", "沐浴露", "洗发水", "体温计"], + "features": ["温和性", "吸收性", "无刺激", "安全性"] + }, + "服装服饰": { + "subcategories": ["婴儿服", "童装", "睡袋", "帽子", "袜子", "鞋子", "围兜"], + "features": ["舒适性", "透气性", "材质", "设计感"] + }, + "安全座椅": { + "subcategories": ["提篮式座椅", "双向安全座椅", "增高垫", "汽车安全座椅"], + "features": ["安全性", "安装方便", "舒适性", "耐用性"] + }, + "玩具教具": { + "subcategories": ["益智玩具", "毛绒玩具", "积木", "绘本", "音乐玩具", "运动玩具"], + "features": ["安全性", "教育性", "趣味性", "材质"] + }, + "出行用品": { + "subcategories": ["婴儿车", "背带", "腰凳", "妈咪包", "折叠椅", "遮阳伞"], + "features": ["便携性", "舒适性", "耐用性", "多功能"] + }, + "寝具用品": { + "subcategories": ["婴儿床", "床垫", "床上用品", "枕头", "被子", "蚊帐"], + "features": ["舒适性", "透气性", "安全性", "耐用性"] + } + } + + # 目标市场数据 + self.target_markets = { + "北美": { + "countries": ["美国", "加拿大"], + "preferences": ["高品质", "安全认证", "有机材料", "创新功能"], + "price_sensitivity": "中低", + "logistics_cost_factor": 1.2 + }, + "欧洲": { + "countries": ["德国", "英国", "法国", "意大利", "西班牙"], + "preferences": ["环保认证", "安全标准", "有机材料", "设计感"], + "price_sensitivity": "中", + "logistics_cost_factor": 1.3 + }, + "东南亚": { + "countries": ["新加坡", "马来西亚", "泰国", "越南", "菲律宾"], + "preferences": ["性价比", "耐用性", "多功能", "本地化适配"], + "price_sensitivity": "高", + "logistics_cost_factor": 0.8 + }, + "中东": { + "countries": ["阿联酋", "沙特阿拉伯", "卡塔尔"], + "preferences": ["高品质", "豪华设计", "品牌认知", "大尺寸"], + "price_sensitivity": "低", + "logistics_cost_factor": 1.5 + } + } + + # 构建选品流程 + self.selection_pipeline = self._build_selection_pipeline() + + # 构建对话流程 + self.chat_pipeline = self._build_chat_pipeline() + + def _init_local_model(self): + """初始化本地模型""" + try: + # 尝试从环境变量获取API密钥 + api_key = os.environ.get("DOUBAO_API_KEY") or os.environ.get("LAZYLLM_DOUBAO_API_KEY") or os.environ.get("OPENAI_API_KEY") + if api_key: + print("检测到环境变量中的API密钥,使用在线模型") + try: + # 优先尝试豆包模型 + if os.environ.get("DOUBAO_API_KEY") or os.environ.get("LAZYLLM_DOUBAO_API_KEY"): + model = OnlineChatModule( + source="doubao", + api_key=api_key + ) + else: + # 如果没有豆包密钥但有OpenAI密钥,尝试OpenAI + model = OnlineChatModule( + source="openai", + api_key=api_key + ) + + # 测试API密钥是否有效 + try: + test_response = model("测试") + print("API密钥验证成功,使用在线模型") + return model + except Exception as test_error: + print(f"API密钥验证失败: {str(test_error)}") + print("将尝试其他模型选项") + # 继续尝试其他选项 + except Exception as api_error: + print(f"使用API密钥初始化在线模型失败: {str(api_error)}") + print("将尝试其他模型选项") + # 继续尝试其他选项 + + # 尝试加载本地模型 + local_model_path = os.environ.get("LOCAL_MODEL_PATH") + if local_model_path: + if os.path.exists(local_model_path): + print(f"尝试加载本地模型: {local_model_path}") + try: + model = TrainableModule(local_model_path) + # 尝试启动模型 + try: + model.start() + print("本地模型启动成功") + except Exception as start_error: + print(f"本地模型启动失败: {str(start_error)}") + print("将尝试使用在线模型作为后备") + raise start_error + return model + except Exception as e: + print(f"加载本地模型失败: {str(e)}") + else: + print(f"本地模型路径不存在: {local_model_path}") + + # 尝试使用常见的本地模型路径 + common_model_paths = [ + os.path.expanduser("~/.cache/huggingface/hub"), + "/models", + "./models" + ] + + for path in common_model_paths: + if os.path.exists(path): + print(f"尝试从常见路径加载模型: {path}") + try: + # 尝试列出目录中的模型 + model_dirs = [d for d in os.listdir(path) + if os.path.isdir(os.path.join(path, d))] + if model_dirs: + # 尝试第一个模型目录 + model_path = os.path.join(path, model_dirs[0]) + print(f"找到模型目录: {model_path}") + model = TrainableModule(model_path) + # 尝试启动模型 + try: + model.start() + print("本地模型启动成功") + return model + except Exception as start_error: + print(f"本地模型启动失败: {str(start_error)}") + print("将尝试下一个模型或在线模型") + continue + except Exception as e: + print(f"从常见路径加载模型失败: {str(e)}") + continue + + # 尝试使用在线模型 + print("尝试使用在线模型...") + try: + model = OnlineChatModule(source="doubao") + # 测试模型是否可用 + try: + test_response = model("测试") + print("在线模型验证成功") + return model + except Exception as test_error: + print(f"在线模型验证失败: {str(test_error)}") + # 使用模拟聊天模块作为后备 + print("使用模拟聊天模块作为后备") + model = self._create_mock_chat_module() + return model + except Exception as online_error: + print(f"在线模型初始化失败: {str(online_error)}") + # 使用模拟聊天模块作为后备 + print("使用模拟聊天模块作为后备") + model = self._create_mock_chat_module() + return model + except Exception as e: + print(f"初始化模型失败: {str(e)}") + # 使用模拟聊天模块作为后备 + print("使用模拟聊天模块作为后备") + model = self._create_mock_chat_module() + return model + + def _create_mock_chat_module(self): + """创建模拟聊天模块作为后备方案""" + class MockChatModule: + def __call__(self, prompt): + # 简单的模拟响应,专注于母婴选品场景 + if "选品" in prompt or "推荐" in prompt: + return "我是跨境母婴类选品智能助手,我可以为您提供专业的母婴产品选品建议。请告诉我您感兴趣的产品类别和目标市场。" + elif "喂养用品" in prompt: + return "喂养用品是母婴产品中的重要类别,包括奶瓶、奶嘴、吸奶器、辅食机等。建议选择符合安全标准、易于清洁的产品。" + elif "护理用品" in prompt: + return "护理用品包括纸尿裤、湿纸巾、护臀膏等。建议选择温和无刺激、吸收性好的产品,特别要注意材质的安全性。" + elif "安全座椅" in prompt: + return "安全座椅是保障婴儿出行安全的重要产品。建议选择通过权威安全认证、安装方便、舒适性好的产品。" + elif "玩具" in prompt: + return "玩具不仅能给孩子带来乐趣,还能促进智力发展。建议选择安全无毒、适合孩子年龄、具有教育意义的玩具。" + elif "目标市场" in prompt: + return "不同的目标市场有不同的消费偏好。北美市场注重高品质和安全认证,欧洲市场注重环保和设计感,东南亚市场注重性价比。" + else: + return "您好!我是跨境母婴类选品智能助手,很高兴为您服务。我可以为您提供母婴产品的选品建议、市场分析和趋势预测。请告诉我您需要什么帮助。" + + return MockChatModule() + + def _build_selection_pipeline(self): + """构建母婴产品选品流程""" + + def market_analysis(user_input): + """市场分析:根据用户需求分析目标市场和产品类别""" + # 提取目标市场 + target_market = None + for market in self.target_markets: + if market in user_input: + target_market = market + break + + # 如果没有指定市场,默认为北美 + if not target_market: + target_market = "北美" + + # 提取产品类别 + product_category = None + for category in self.product_categories: + if category in user_input: + product_category = category + break + + # 如果没有指定类别,使用大模型分析 + if not product_category: + prompt = f""" + 根据用户需求"{user_input}",从以下母婴产品类别中选择最合适的一个: + {list(self.product_categories.keys())} + + 只回答类别名称,不需要解释。 + """ + product_category = self.chat_module(prompt).strip() + if product_category not in self.product_categories: + product_category = "喂养用品" # 默认类别 + + return { + "target_market": target_market, + "product_category": product_category, + "market_info": self.target_markets[target_market], + "category_info": self.product_categories[product_category] + } + + def product_research(analysis_result): + """产品研究:基于市场和类别研究热门产品""" + target_market = analysis_result["target_market"] + product_category = analysis_result["product_category"] + + # 构建产品研究提示 + prompt = f""" + 作为跨境电商母婴选品专家,请为{target_market}市场研究5款热门的{product_category}产品。 + + 要求: + 1. 每款产品包含:产品名称、主要特点、目标客户群体、价格区间(美元) + 2. 考虑{target_market}市场的消费者偏好:{', '.join(analysis_result['market_info']['preferences'])} + 3. 考虑价格敏感度:{analysis_result['market_info']['price_sensitivity']} + 4. 产品应符合母婴产品的安全标准和质量要求 + 5. 以JSON格式返回,包含products数组,每个产品包含name, features, target_audience, price_range字段 + + 示例格式: + {{ + "products": [ + {{ + "name": "产品名称", + "features": ["特点1", "特点2"], + "target_audience": "目标客户群体", + "price_range": {{ + "min": 10, + "max": 50 + }} + }} + ] + }} + """ + + try: + response = self.chat_module(prompt) + # 尝试解析JSON + if "```json" in response: + json_str = response.split("```json")[1].split("```")[0].strip() + else: + # 尝试提取JSON部分 + json_match = re.search(r'\{.*\}', response, re.DOTALL) + json_str = json_match.group(0) if json_match else response + + product_data = json.loads(json_str) + return {{ + **analysis_result, + "researched_products": product_data.get("products", []) + }} + except Exception as e: + print(f"解析产品研究数据失败: {e}") + # 返回默认产品数据 + default_products = [ + {{ + "name": f"智能{product_category}A", + "features": ["安全材质", "易用性", "高品质"], + "target_audience": "新手父母", + "price_range": {{"min": 20, "max": 60}} + }}, + {{ + "name": f"便携{product_category}B", + "features": ["轻便易携带", "多功能", "性价比高"], + "target_audience": "经常外出的父母", + "price_range": {{"min": 15, "max": 40}} + }} + ] + + return {{ + **analysis_result, + "researched_products": default_products + }} + + def supplier_analysis(research_result): + """供应商分析:分析潜在供应商和成本结构""" + products = research_result["researched_products"] + target_market = research_result["target_market"] + logistics_factor = research_result["market_info"]["logistics_cost_factor"] + + analyzed_products = [] + + for product in products: + # 构建供应商分析提示 + prompt = f""" + 作为供应链专家,请分析"{product['name']}"的供应商情况: + + 1. 估算生产成本(基于产品特点和价格区间) + 2. 估算物流成本(考虑{target_market}市场,物流系数{logistics_factor}) + 3. 估算关税和税费 + 4. 识别潜在供应链风险 + 5. 推荐供应商类型(工厂、贸易公司等) + + 以JSON格式返回,包含production_cost, logistics_cost, tariffs, risks, supplier_type字段 + """ + + try: + response = self.chat_module(prompt) + # 尝试解析JSON + if "```json" in response: + json_str = response.split("```json")[1].split("```")[0].strip() + else: + # 尝试提取JSON部分 + json_match = re.search(r'\{.*\}', response, re.DOTALL) + json_str = json_match.group(0) if json_match else response + + analysis_data = json.loads(json_str) + except Exception as e: + print(f"解析供应商分析数据失败: {e}") + # 返回默认分析数据 + price_mid = (product["price_range"]["min"] + product["price_range"]["max"]) / 2 + analysis_data = { + "production_cost": price_mid * 0.3, + "logistics_cost": price_mid * 0.1 * logistics_factor, + "tariffs": price_mid * 0.05, + "risks": ["供应链不稳定", "质量波动"], + "supplier_type": "工厂直销" + } + + # 计算利润率 + try: + prod_cost = analysis_data.get("production_cost", 0) + if isinstance(prod_cost, dict): + min_val = prod_cost.get("min", 0) + max_val = prod_cost.get("max", 0) + prod_cost = float(min_val) if min_val else float(max_val) + production_cost = float(prod_cost) + except (ValueError, TypeError): + production_cost = 0 + + try: + log_cost = analysis_data.get("logistics_cost", 0) + if isinstance(log_cost, dict): + min_val = log_cost.get("min", 0) + max_val = log_cost.get("max", 0) + log_cost = float(min_val) if min_val else float(max_val) + logistics_cost = float(log_cost) + except (ValueError, TypeError): + logistics_cost = 0 + + try: + tariff_cost = analysis_data.get("tariffs", 0) + if isinstance(tariff_cost, dict): + percentage = tariff_cost.get("percentage", 0) + min_val = tariff_cost.get("min", 0) + max_val = tariff_cost.get("max", 0) + if percentage: + tariff_cost = float(percentage) + else: + tariff_cost = float(min_val) if min_val else float(max_val) + tariffs = float(tariff_cost) + except (ValueError, TypeError): + tariffs = 0 + + total_cost = production_cost + logistics_cost + tariffs + avg_price = (product["price_range"]["min"] + product["price_range"]["max"]) / 2 + profit_margin = (avg_price - total_cost) / avg_price * 100 + + analyzed_products.append({ + **product, + "cost_analysis": analysis_data, + "total_cost": total_cost, + "average_price": avg_price, + "profit_margin": profit_margin + }) + + return { + **research_result, + "analyzed_products": analyzed_products + } + + def competition_analysis(analysis_result): + """竞争分析:分析市场竞争格局和差异化机会""" + products = analysis_result["analyzed_products"] + target_market = analysis_result["target_market"] + product_category = analysis_result["product_category"] + + competition_results = [] + + for product in products: + # 构建竞争分析提示 + prompt = f""" + 作为市场分析师,请分析"{product['name']}"在{target_market}{product_category}市场的竞争情况: + + 1. 识别主要竞争对手(2-3个) + 2. 分析竞争对手定价策略 + 3. 识别市场空白和差异化机会 + 4. 评估市场进入难度 + 5. 提供差异化建议 + + 以JSON格式返回,包含competitors, pricing_strategy, market_gaps, entry_difficulty, differentiation字段 + """ + + try: + response = self.chat_module(prompt) + # 尝试解析JSON + if "```json" in response: + json_str = response.split("```json")[1].split("```")[0].strip() + else: + # 尝试提取JSON部分 + json_match = re.search(r'\{.*\}', response, re.DOTALL) + json_str = json_match.group(0) if json_match else response + + competition_data = json.loads(json_str) + except Exception as e: + print(f"解析竞争分析数据失败: {e}") + # 返回默认分析数据 + competition_data = { + "competitors": ["品牌A", "品牌B"], + "pricing_strategy": "中高端定价", + "market_gaps": ["有机材料", "个性化定制"], + "entry_difficulty": "中等", + "differentiation": ["创新设计", "优质材料"] + } + + competition_results.append({ + **product, + "competition_analysis": competition_data + }) + + return { + **analysis_result, + "competition_results": competition_results + } + + def generate_recommendation(competition_result): + """生成最终推荐:基于所有分析生成选品推荐""" + products = competition_result["competition_results"] + target_market = competition_result["target_market"] + product_category = competition_result["product_category"] + + # 按利润率排序 + sorted_products = sorted(products, key=lambda x: x["profit_margin"], reverse=True) + + # 构建推荐提示 + top_products = sorted_products[:3] # 取前3个产品 + products_summary = "\n".join([ + f"{i+1}. {p['name']}: 利润率{p['profit_margin']:.1f}%, 市场差异化: {', '.join(p['competition_analysis'].get('differentiation', ['无差异化信息']))}" + for i, p in enumerate(top_products) + ]) + + prompt = f""" + 基于以下分析结果,为{target_market}{product_category}市场提供选品建议: + + 产品分析结果: + {products_summary} + + 请提供: + 1. 最推荐的产品及理由 + 2. 市场进入策略 + 3. 风险控制建议 + 4. 长期发展规划 + + 以专业、简洁的顾问口吻回答,突出关键建议。 + """ + + recommendation = self.chat_module(prompt) + + return { + "market": target_market, + "category": product_category, + "top_recommendations": sorted_products[:3], + "detailed_recommendation": recommendation, + "generation_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") + } + + # 构建流程管道 - 使用自定义函数替代pipeline,避免字典哈希错误 + def selection_pipeline(user_input): + # 依次执行各个分析步骤 + market_result = market_analysis(user_input) + research_result = product_research(market_result) + supplier_result = supplier_analysis(research_result) + competition_result = competition_analysis(supplier_result) + final_result = generate_recommendation(competition_result) + return final_result + + return selection_pipeline + + def _build_chat_pipeline(self): + """构建对话流程""" + + def intent_recognition(user_input): + """意图识别:识别用户意图是选品咨询还是其他问题""" + prompt = f""" + 请识别用户意图:"用户说:{user_input}" + + 意图类型: + 1. 选品咨询 - 用户想要获取产品选品建议 + 2. 市场咨询 - 用户想了解市场情况 + 3. 产品咨询 - 用户想了解特定产品信息 + 4. 其他 - 其他类型问题 + + 只回答意图类型编号(1-4),不需要解释。 + """ + + try: + intent = self.chat_module(prompt).strip() + intent_map = { + "1": "选品咨询", + "2": "市场咨询", + "3": "产品咨询", + "4": "其他" + } + return intent_map.get(intent, "其他") + except: + return "其他" + + def response_generation(user_input, intent): + """根据意图生成回复""" + if intent == "选品咨询": + # 调用选品流程 + try: + result = self.selection_pipeline(user_input) + return self._format_selection_result(result) + except Exception as e: + return f"选品分析过程中出现错误:{str(e)}。请尝试重新描述您的需求。" + + elif intent == "市场咨询": + prompt = f""" + 作为跨境电商母婴市场专家,请回答以下市场咨询问题: + 用户问题:{user_input} + + 请提供专业、简洁的回答,重点关注北美、欧洲和东南亚市场的母婴产品趋势。 + """ + return self.chat_module(prompt) + + elif intent == "产品咨询": + prompt = f""" + 作为母婴产品专家,请回答以下产品咨询问题: + 用户问题:{user_input} + + 请提供专业、准确的回答,重点关注产品的安全性、质量标准和使用建议。 + """ + return self.chat_module(prompt) + + else: + # 其他类型问题,直接使用聊天模块 + prompt = f""" + 作为跨境母婴类选品智能助手,请回答用户的问题: + 用户问题:{user_input} + + 请提供友好、专业的回答,重点关注母婴产品相关的内容。 + """ + return self.chat_module(prompt) + + def chat_pipeline(input_data): + """整合对话流程""" + user_input = input_data.get("user_input", "") + + # 识别用户意图 + intent = intent_recognition(user_input) + + # 生成响应 + response = response_generation(user_input, intent) + + return { + "response": response + } + + return chat_pipeline + + def _format_selection_result(self, result): + """格式化选品结果""" + if "error" in result: + return f"选品分析过程中出现错误:{result['error']}" + + # 构建格式化的响应 + response = f"# 跨境母婴类选品建议\n\n" + response += f"## 市场与类别分析\n" + response += f"- 目标市场:{result['market']}\n" + response += f"- 产品类别:{result['category']}\n\n" + + response += f"## 热门产品推荐\n" + for i, product in enumerate(result['top_recommendations'], 1): + response += f"### {i}. {product['name']}\n" + response += f"- **主要特点**:{', '.join(product['features'])}\n" + response += f"- **目标客户群体**:{product['target_audience']}\n" + response += f"- **价格区间**:${product['price_range']['min']} - ${product['price_range']['max']}\n" + response += f"- **利润率**:{product['profit_margin']:.1f}%\n\n" + + response += f"## 详细选品建议\n" + response += result['detailed_recommendation'] + response += f"\n\n---\n" + response += f"生成时间:{result['generation_time']}\n" + response += f"如果您需要更详细的产品分析或市场调研,请随时告诉我!" + + return response + + def chat(self, user_input): + """与用户进行对话""" + try: + result = self.chat_pipeline({"user_input": user_input}) + return result.get("response", "抱歉,我无法理解您的请求。") + except Exception as e: + print(f"对话处理失败: {e}") + return "抱歉,我在处理您的请求时遇到了问题。" + + def get_product_categories(self): + """获取产品类别信息""" + return self.product_categories + + def get_target_markets(self): + """获取目标市场信息""" + return self.target_markets + +# 测试代码 +if __name__ == "__main__": + agent = CrossBorderMaternalAgent() + print("跨境母婴类选品智能助手已启动!") + print("您可以输入问题或需求,输入'quit'退出。") + + while True: + user_input = input("用户: ") + if user_input.lower() == "quit": + print("再见!") + break + + response = agent.chat(user_input) + print(f"助手: {response}") \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..d119c62c7522e0e936a557648ba201ed390b67cc --- /dev/null +++ b/requirements.txt @@ -0,0 +1,18 @@ +lazyllm>=0.1.0 +transformers>=4.30.0 +pydantic>=2.0.0 +requests>=2.31.0 +pandas>=2.0.0 +flask>=2.0.0 +numpy>=1.24.0 +matplotlib>=3.7.0 +openai>=1.0.0 # 用于真实API调用 +tiktoken>=0.5.0 # 用于token计算 +aiohttp>=3.8.0 # 用于异步HTTP请求 +asyncio # 用于异步处理 +torch>=2.0.0 # 本地模型所需 +accelerate>=0.20.0 # 本地模型加速 +sentencepiece>=0.1.99 # 本地模型分词 +protobuf>=3.20.0 # 本地模型依赖 +scipy>=1.10.0 # 本地模型依赖 +modelscope>=1.30.0 # 本地模型下载和管理 \ No newline at end of file diff --git a/web_agent.py b/web_agent.py new file mode 100644 index 0000000000000000000000000000000000000000..d25b48ff9120563b2e6d8adaa967ce37e6ceee57 --- /dev/null +++ b/web_agent.py @@ -0,0 +1,580 @@ +from flask import Flask, render_template_string, request, jsonify +from maternal_agent import MaternalAgent + +# 初始化Flask应用 +app = Flask(__name__) + +# 初始化母婴选品Agent +agent = MaternalAgent() + +# HTML模板 +html_template = ''' + + +
+ + +专业的母婴产品选品建议、市场分析和趋势预测
+