diff --git a/media_3c_agent.py b/media_3c_agent.py new file mode 100644 index 0000000000000000000000000000000000000000..522f968e6e2241c8e3048ea48f3a12749a52609f --- /dev/null +++ b/media_3c_agent.py @@ -0,0 +1,567 @@ +import os +import json +from datetime import datetime +from lazyllm import OnlineChatModule, WebModule +import re + +# 跨境3C配件选品智能助手Agent +class CrossBorder3CAgent: + def __init__(self): + # 初始化LazyLLM聊天模块 - 只使用在线模型 + try: + # 从环境变量获取API密钥 + api_key = os.environ.get("LAZYLLM_DOUBAO_API_KEY") + if not api_key: + raise ValueError("api_key is required") + + print("检测到环境变量中的API密钥,使用在线模型") + try: + if os.environ.get("LAZYLLM_DOUBAO_API_KEY"): + self.chat_module = OnlineChatModule( + source="doubao", + api_key=api_key + ) + # 测试API密钥是否有效 + try: + test_response = self.chat_module("测试") + print("API密钥验证成功,使用在线模型") + except Exception as test_error: + raise Exception(f"API密钥验证失败: {str(test_error)}") + except Exception as api_error: + raise Exception(f"使用API密钥初始化在线模型失败: {str(api_error)}") + except Exception as e: + print(f"无法初始化在线模型: {e}") + # 不使用任何后备模型,直接抛出异常 + raise Exception(f"模型初始化失败:{e}") + + # 3C配件类别数据 + self.product_categories = { + "手机配件": { + "subcategories": ["手机壳", "手机膜", "充电器", "数据线", "移动电源", "无线充电器", "手机支架"], + "features": ["便携性", "兼容性", "耐用性", "设计感"] + }, + "电脑配件": { + "subcategories": ["笔记本支架", "键盘", "鼠标", "散热器", "USB集线器", "网线", "摄像头"], + "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 + } + } + + # 构建选品流程 + self.selection_pipeline = self._build_selection_pipeline() + + # 构建对话流程 + self.chat_pipeline = self._build_chat_pipeline() + + def _build_selection_pipeline(self): + """构建3C配件选品流程""" + + 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}",从以下3C配件类别中选择最合适的一个: + {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": "工厂直销" + } + + # 计算利润率 + # 确保cost_analysis中的值是数字类型 + try: + production_cost = float(analysis_data.get("production_cost", 0)) + # 如果production_cost是字典,尝试获取其中的值 + if isinstance(production_cost, dict): + production_cost = float(production_cost.get("min", production_cost.get("max", 0))) + except (ValueError, TypeError): + production_cost = 0 + + try: + logistics_cost = float(analysis_data.get("logistics_cost", 0)) + # 如果logistics_cost是字典,尝试获取其中的值 + if isinstance(logistics_cost, dict): + logistics_cost = float(logistics_cost.get("min", logistics_cost.get("max", 0))) + except (ValueError, TypeError): + logistics_cost = 0 + + try: + tariffs = float(analysis_data.get("tariffs", 0)) + # 如果tariffs是字典,尝试获取其中的值 + if isinstance(tariffs, dict): + tariffs = float(tariffs.get("percentage", tariffs.get("min", tariffs.get("max", 0)))) + 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_process(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_process + + 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""" + 作为3C配件产品专家,请回答以下产品咨询问题: + 用户问题:{user_input} + + 请提供专业、准确的产品信息和建议。 + """ + return self.chat_module(prompt) + + else: + prompt = f""" + 作为跨境3C配件选品智能助手,请回答以下问题: + 用户问题:{user_input} + + 请提供专业、有帮助的回答。如果问题超出范围,请礼貌说明。 + """ + return self.chat_module(prompt) + + # 构建对话流程 + def chat_process(user_input): + intent = intent_recognition(user_input) + response = response_generation(user_input, intent) + return { + "user_input": user_input, + "intent": intent, + "response": response, + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S") + } + + return chat_process + + def _format_selection_result(self, result): + """格式化选品结果为易读的文本""" + # 检查result是否为字典类型 + if not isinstance(result, dict): + return f"选品结果格式错误:期望字典类型,收到{type(result).__name__}类型\n\n原始内容:\n{str(result)}" + + # 检查必要的键是否存在 + required_keys = ['market', 'category', 'generation_time', 'top_recommendations', 'detailed_recommendation'] + for key in required_keys: + if key not in result: + return f"选品结果格式错误:缺少必要字段 {key}" + + response = f""" +## 🌍 {result['market']}市场{result['category']}选品分析报告 + +**生成时间:** {result['generation_time']} + +### 📊 推荐产品 + +""" + + # 检查top_recommendations是否为列表 + if not isinstance(result['top_recommendations'], list): + response += "推荐产品数据格式错误:期望列表类型\n\n" + else: + for i, product in enumerate(result['top_recommendations'], 1): + # 检查product是否为字典且包含必要字段 + if not isinstance(product, dict): + response += f"产品 {i} 数据格式错误:期望字典类型,收到{type(product).__name__}类型\n\n" + continue + + # 检查必要字段是否存在 + product_required_keys = ['name', 'target_audience', 'price_range', 'profit_margin', 'features', 'competition_analysis'] + missing_keys = [key for key in product_required_keys if key not in product] + if missing_keys: + response += f"产品 {i} 缺少必要字段: {', '.join(missing_keys)}\n\n" + continue + + # 检查price_range是否为字典且包含min和max + if not isinstance(product['price_range'], dict) or 'min' not in product['price_range'] or 'max' not in product['price_range']: + response += f"产品 {i} 价格范围格式错误\n\n" + continue + + # 检查competition_analysis是否为字典且包含differentiation和entry_difficulty + if not isinstance(product['competition_analysis'], dict) or 'differentiation' not in product['competition_analysis'] or 'entry_difficulty' not in product['competition_analysis']: + response += f"产品 {i} 竞争分析格式错误\n\n" + continue + + response += f""" +**{i}. {product['name']}** +- 目标客户:{product['target_audience']} +- 价格区间:${product['price_range']['min']}-${product['price_range']['max']} +- 预估利润率:{product['profit_margin']:.1f}% +- 产品特点:{', '.join(product['features'])} +- 差异化优势:{', '.join(product['competition_analysis']['differentiation'])} +- 市场进入难度:{product['competition_analysis']['entry_difficulty']} + +""" + + response += f""" +### 💡 专业建议 + +{result['detailed_recommendation']} + +--- +*本分析由跨境3C配件选品智能助手基于LazyLLM框架生成* + """ + + return response + + def chat(self, user_input): + """处理用户输入并返回回复""" + result = self.chat_pipeline(user_input) + # 如果结果是字典,返回response字段;否则直接返回结果 + if isinstance(result, dict): + return result.get("response", str(result)) + return result + + def start_web_chat(self): + """启动Web聊天界面""" + # 创建Web模块 + web_module = WebModule(self.chat_pipeline) + # 启动Web服务 + web_module.start().wait() + +# 主函数 +if __name__ == "__main__": + # 创建3C配件选品智能助手 + agent = CrossBorder3CAgent() + + # 启动Web聊天界面 + print("正在启动跨境3C配件选品智能助手Web界面...") + agent.start_web_chat() \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..43965d2211e944a7ccb855ee30bc563a8fb1fa04 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +lazyllm>=0.1.0 +flask>=2.0.0 +requests>=2.31.0 \ No newline at end of file diff --git a/web_app.py b/web_app.py new file mode 100644 index 0000000000000000000000000000000000000000..f59500885a72603ee5d775982e170fcf3d9d84af --- /dev/null +++ b/web_app.py @@ -0,0 +1,585 @@ +from flask import Flask, render_template_string, request, jsonify +from media_3c_agent import CrossBorder3CAgent + +app = Flask(__name__) + +# 创建3C配件选品智能助手实例 +try: + agent = CrossBorder3CAgent() +except Exception as e: + print(f"创建智能助手实例时出错: {str(e)}") + import traceback + traceback.print_exc() + agent = None + +# HTML模板 +html_template = ''' + + +
+ + +基于LazyLLM框架的智能选品顾问,助您开拓全球市场
+基于市场数据和AI分析,为您提供精准的3C配件选品建议
+覆盖北美、欧洲、东南亚等主要市场,了解各地消费者偏好
+全面分析产品成本、物流费用和预期利润率,优化定价策略
+深入分析竞争对手和市场机会,找到差异化优势
+基于LazyLLM的智能顾问
+