随着云原生技术和AI能力的不断成熟，如何将传统数据库、无服务器计算和人工智能有机结合，构建现代化的数据分析平台，并借助阿里云提供的基础组件构建应用。本文基于微软经典的WorldWideImports示例数据库，设计并实现了一个智能销售分析平台的技术原型，并使用通义AI在数据洞察和内容生成方面的智能化能力。通过该Demo项目，我们不仅验证了各个组件的协同可行性，还展示了从购物篮分析、商品关联挖掘到AI营销素材生成的完整技术链路，为类似的企业级应用提供了参考和实现思路。

整体架构

架构图

整个DEMO架构相对简单：

• 数据存储层 阿里云RDS for SQL Server作为核心数据库，采用微软官方示例数据库
• 计算编排层：使用阿里云函数计算作为核心引擎，承担了数据查询、业务逻辑处理和服务编排的职责。函数计算的按需付费模式很适合这种DEMO分析场景，既能处理数据库查询，又能对接百炼大模型服务进行智能分析。
• Web前端：采用纯静态页面的方式，通过HTTP请求直接调用函数计算接口。整体较为简单

使用到的云产品简介

阿里云RDS for SQL Server：企业级数据库服务

阿里云RDS for SQL Server是基于微软SQL Server引擎的托管数据库服务，提供了完整的企业级数据管理能力。相比自建数据库，RDS具备自动备份恢复、高可用架构、性能监控等开箱即用的特性，大幅降低了数据库运维复杂度。

阿里云函数计算：无服务器计算平台

函数计算（Function Compute）是阿里云提供的事件驱动的全托管计算服务，开发者无需管理服务器即可运行代码。该服务支持多种编程语言运行时，具备毫秒级冷启动、按量付费、自动弹性伸缩等特性。在本项目中，我们使用Python运行时构建了统一的处理函数，通过参数路由实现数据查询、AI分析等多种功能。函数计算的无状态特性和按需执行模式，使得整个系统具备了良好的成本控制能力和高并发处理能力，非常适合构建轻量级的API服务。

百炼平台与通义大模型：AI能力的统一入口

百炼是阿里云推出的大模型服务平台，为开发者提供了便捷的AI能力调用接口。通过百炼平台，我们能够轻松接入通义千问和通义万像两大核心模型。通义千问作为大语言模型，具备强大的自然语言理解和生成能力，能够将复杂的数据分析结果转化为易懂的商业洞察和建议。通义万像则专注于图像生成领域，支持根据文本描述创建高质量的营销图片。

数据准备

在开始构建智能销售分析平台之前，我们需要先准备好示例数据。WorldWideImports是微软官方提供的SQL Server示例数据库，包含了完整的销售、库存、客户等业务数据，非常适合用于数据分析场景的演示。

数据库准备

在https://github.com/Microsoft/sql-server-samples 下载官方示例数据库，并上传到与RDS for SQL Server同一可用区的OSS Bucket中。

在阿里云RDS控制台中

两分钟后，我们在控制台看到该数据库已经Ready。

数据库账号准备

在控制台创建账号，并给与WWI数据库读写权限，如图所示，该账号用于函数计算连接RDS SQL Server

SQL准备

通过一些复杂的分析SQL，我们可以实现5种与销售分析有关的SQL，如下

# 1. 热销商品分析查询 # 用途：显示销售额最高的商品及其详细信息，包括收入、客户数量、平均价格等 TOP_SELLING_PRODUCTS_QUERY = """ -- 热销商品分析查询（已验证可用） -- 显示销售额最高的商品及其详细信息 WITH TopProducts AS (SELECT TOP 10 si.StockItemName, si.SearchDetails, \ ISNULL(sc.ColorName, 'Default') as ColorName, \ si.Brand, \ si.Size, \ SUM(ol.Quantity * ol.UnitPrice) as Revenue, \ COUNT(DISTINCT o.CustomerID) as CustomerCount, \ AVG(ol.UnitPrice) as AvgPrice, \ SUM(ol.Quantity) as TotalQuantitySold \ FROM Warehouse.StockItems si \ LEFT JOIN Sales.OrderLines ol ON si.StockItemID = ol.StockItemID \ LEFT JOIN Sales.Orders o ON ol.OrderID = o.OrderID \ LEFT JOIN Warehouse.Colors sc ON si.ColorID = sc.ColorID \ WHERE o.OrderDate >= '2016-01-01' \ GROUP BY si.StockItemID, si.StockItemName, si.SearchDetails, \ sc.ColorName, si.Brand, si.Size \ HAVING SUM(ol.Quantity * ol.UnitPrice) > 0 \ ORDER BY Revenue DESC) SELECT * \ FROM TopProducts; \ """ # 2. 商品关联分析查询（购物篮分析） # 用途：分析哪些商品经常一起被购买，用于推荐系统和交叉销售 PRODUCT_AFFINITY_ANALYSIS_QUERY = """ -- 商品关联分析查询（购物篮分析） -- 分析哪些商品经常一起被购买 WITH ProductAffinityAnalysis AS (SELECT ol1.StockItemID AS Product1, \ ol2.StockItemID AS Product2, \ p1.StockItemName AS Product1Name, \ p2.StockItemName AS Product2Name, \ COUNT(*) AS CoOccurrence, \ COUNT(*) * 1.0 / \ (SELECT COUNT(DISTINCT OrderID) \ FROM Sales.OrderLines \ WHERE StockItemID = ol1.StockItemID) AS SupportScore \ FROM Sales.OrderLines ol1 \ INNER JOIN Sales.OrderLines ol2 \ ON ol1.OrderID = ol2.OrderID AND ol1.StockItemID < ol2.StockItemID \ INNER JOIN Warehouse.StockItems p1 ON ol1.StockItemID = p1.StockItemID \ INNER JOIN Warehouse.StockItems p2 ON ol2.StockItemID = p2.StockItemID \ INNER JOIN Sales.Orders o ON ol1.OrderID = o.OrderID \ WHERE o.OrderDate >= '2016-01-01' \ GROUP BY ol1.StockItemID, ol2.StockItemID, \ p1.StockItemName, p2.StockItemName \ HAVING COUNT(*) >= 3 -- 至少同时出现3次 ) SELECT TOP 20 Product1Name, Product2Name, CoOccurrence, SupportScore FROM ProductAffinityAnalysis ORDER BY SupportScore DESC; \ """ # 3. 库存状态分析查询 # 用途：分析当前库存状态和商品需求，识别库存风险 INVENTORY_STATUS_ANALYSIS_QUERY = """ -- 库存状态分析查询 -- 分析当前库存状态和商品需求 SELECT TOP 30 si.StockItemID, si.StockItemName, si.Brand, si.Size, ISNULL(sih.QuantityOnHand, 0) as CurrentStock, ISNULL(sih.LastCostPrice, 0) as LastCostPrice, COUNT(ol.OrderLineID) as OrderFrequency, AVG(ol.Quantity) AS AvgDemand, SUM(ol.Quantity * ol.UnitPrice) as TotalRevenue, CASE WHEN ISNULL(sih.QuantityOnHand, 0) <= 50 THEN 'Critical' WHEN ISNULL(sih.QuantityOnHand, 0) <= 200 THEN 'Low' ELSE 'Normal' END AS StockStatus FROM Warehouse.StockItems si LEFT JOIN Warehouse.StockItemHoldings sih ON si.StockItemID = sih.StockItemID LEFT JOIN Sales.OrderLines ol ON si.StockItemID = ol.StockItemID LEFT JOIN Sales.Orders o ON ol.OrderID = o.OrderID WHERE o.OrderDate >= '2015-06-01' \ OR o.OrderDate IS NULL GROUP BY si.StockItemID, si.StockItemName, si.Brand, si.Size, sih.QuantityOnHand, \ sih.LastCostPrice ORDER BY OrderFrequency DESC; \ """ # 4. 客户购买行为分析查询 # 用途：分析客户购买模式和价值，用于客户分层和营销策略 CUSTOMER_BEHAVIOR_ANALYSIS_QUERY = """ -- 客户购买行为分析查询 -- 分析客户购买模式和价值 WITH CustomerAnalysis AS (SELECT c.CustomerID, \ c.CustomerName, \ c.DeliveryAddressLine1, \ c.DeliveryAddressLine2, \ COUNT(DISTINCT o.OrderID) as OrderCount, \ SUM(ol.Quantity * ol.UnitPrice) as TotalSpent, \ AVG(ol.Quantity * ol.UnitPrice) as AvgOrderValue, \ MAX(o.OrderDate) as LastOrderDate, \ DATEDIFF(DAY, MAX(o.OrderDate), '2016-05-31') as DaysSinceLastOrder, \ COUNT(DISTINCT ol.StockItemID) as UniqueProductsPurchased \ FROM Sales.Customers c \ INNER JOIN Sales.Orders o ON c.CustomerID = o.CustomerID \ INNER JOIN Sales.OrderLines ol ON o.OrderID = ol.OrderID \ WHERE o.OrderDate >= '2016-01-01' \ GROUP BY c.CustomerID, c.CustomerName, \ c.DeliveryAddressLine1, c.DeliveryAddressLine2) SELECT TOP 20 CustomerName, DeliveryAddressLine2 as Region, OrderCount, TotalSpent, AvgOrderValue, UniqueProductsPurchased, CASE WHEN TotalSpent > 50000 THEN 'VIP' WHEN TotalSpent > 20000 THEN 'Premium' WHEN TotalSpent > 5000 THEN 'Regular' ELSE 'Basic' END AS CustomerTier, CASE WHEN DaysSinceLastOrder <= 30 THEN 'Active' WHEN DaysSinceLastOrder <= 90 THEN 'At Risk' ELSE 'Inactive' END AS CustomerStatus FROM CustomerAnalysis ORDER BY TotalSpent DESC; \ """ # 5. 季节性销售趋势分析查询 # 用途：分析商品在不同月份的销售表现，识别季节性模式 SEASONAL_SALES_TREND_QUERY = """ -- 季节性销售趋势分析查询 -- 分析商品在不同月份的销售表现 SELECT MONTH (o.OrderDate) as OrderMonth, DATENAME(MONTH, o.OrderDate) as MonthName, si.StockItemName, SUM (ol.Quantity * ol.UnitPrice) as MonthlyRevenue, COUNT (DISTINCT o.OrderID) as OrderCount, AVG (ol.Quantity) as AvgQuantityPerOrder, SUM (ol.Quantity) as TotalQuantitySold FROM Sales.Orders o INNER JOIN Sales.OrderLines ol \ ON o.OrderID = ol.OrderID INNER JOIN Warehouse.StockItems si ON ol.StockItemID = si.StockItemID WHERE o.OrderDate >= '2016-01-01' AND si.StockItemName IN ( -- 只分析前10个热销商品的季节性趋势 SELECT TOP 10 si2.StockItemName FROM Warehouse.StockItems si2 INNER JOIN Sales.OrderLines ol2 ON si2.StockItemID = ol2.StockItemID INNER JOIN Sales.Orders o2 ON ol2.OrderID = o2.OrderID WHERE o2.OrderDate >= '2016-01-01' GROUP BY si2.StockItemName ORDER BY SUM (ol2.Quantity * ol2.UnitPrice) DESC ) GROUP BY MONTH (o.OrderDate), DATENAME(MONTH, o.OrderDate), si.StockItemName ORDER BY OrderMonth, MonthlyRevenue DESC; \ """

示例结果如图

配置大模型访问

进入阿里云百炼控制台https://bailian.console.aliyun.com/ ，获取用于调用大模型的API Key，这里我们的文本模式使用qwen-max，图片生成模型使用 wanx2.1-t2i-plus

对于这5种销售分析场景，我们分别将提示词封装到类中，将SQL Server返回的数据配合固定的系统提示词与用户提示词传给代码，核心代码如下：

class AIEnhance: """封装对大语言模型的调用，用于数据分析""" def __init__(self, model='qwen-max', image_model='wanx2.1-t2i-plus'): self.model = model self.image_model = image_model def analyze_top_selling_products(self, products_data): system_prompt = "你是一位资深的数据分析师。" user_prompt = ("请根据以下热销商品数据，生成一段简洁、深刻的分析报告。\n" "报告应包括：\n" "1. 总结关键的销售趋势。\n" "2. 指出表现最突出的几款商品及其特点。\n" "3. 基于数据，提出1-2条具体的商业建议。") return self._execute_analysis_prompt(system_prompt, user_prompt, products_data) def analyze_product_affinity(self, affinity_data): system_prompt = "你是一位零售策略师，擅长通过购物篮分析提升交叉销售。" user_prompt = ("请分析以下商品关联数据（哪些商品经常被一起购买）。\n" "分析应包括：\n" "1. 识别出关联性最强的几组商品（基于SupportScore和CoOccurrence）。\n" "2. 提出具体的营销建议，如商品捆绑、组合折扣或在商品页面进行推荐。\n" "3. 指出是否有任何非直观的、有趣的商品组合值得关注。") return self._execute_analysis_prompt(system_prompt, user_prompt, affinity_data) def analyze_inventory(self, inventory_data): system_prompt = "你是一位供应链和库存管理专家。" user_prompt = ("请根据下面的库存状态和销售数据进行分析。\n" "分析应包括：\n" "1. 识别出有风险的商品，特别是库存状态为'Critical'且订单频率(OrderFrequency)较高的商品，并建议立即采取行动。\n" "2. 找出库存积压的商品（库存量大但订单频率低），并建议处理方式（如促销、清仓）。\n" "3. 对整体库存健康状况给出一个简要评价。") return self._execute_analysis_prompt(system_prompt, user_prompt, inventory_data) def analyze_customer_behavior(self, customer_data): system_prompt = "你是一位客户关系管理（CRM）与营销专家。" user_prompt = ("请分析以下客户行为和分层数据。\n" "分析应包括：\n" "1. 总结客户群体的构成（如VIP, Premium客户的占比和特点）。\n" "2. 为不同客户状态('Active', 'At Risk', 'Inactive')的客户群体提出差异化的营销策略。\n" "3. 描述高价值客户的核心特征，为精准营销提供依据。") return self._execute_analysis_prompt(system_prompt, user_prompt, customer_data) def analyze_seasonal_trends(self, seasonal_data): system_prompt = "你是一位市场分析师，精通销售预测和季节性趋势分析。" user_prompt = ("请分析以下热销商品的月度销售数据，识别季节性趋势。\n" "分析应包括：\n" "1. 指出哪些商品有明显的销售旺季和淡季。\n" "2. 总结哪些月份是整体销售的高峰期和低谷期。\n" "3. 基于这些季节性模式，为库存规划和营销活动提供具体建议。") return self._execute_analysis_prompt(system_prompt, user_prompt, seasonal_data)

配置函数计算

进入函数计算控制台https://fcnext.console.aliyun.com/ ，创建函数

函数类型选择“任务函数”。

部署代码

创建完成后，部署代码，入口代码如下，通过解析简单的QueryString获取参数，通过不同的访问类别调用不同的SQL与大模型分析，典型的访问请求例如：

https://自定义函数地址.cn-beijing.fcapp.run?analysis_type=product_affinity&ai_result=true&generate_image=true

import json import logging # 配置日志 logger = logging.getLogger() logger.setLevel(logging.INFO) from database_wrapper import WideWorldImportersDB from ai_enhance import create_ai_analyzer db = WideWorldImportersDB() ai_service = create_ai_analyzer() def handler(event, context): """ 函数入口点 (事件驱动) Args: event (dict): 期望包含: - analysis_type: 分析类型 (必需) - ai_result: 是否需要AI分析 (可选，值为 'true' 时启用) - generate_image: 是否生成图片 (可选，值为 'true' 时启用，仅支持 product_affinity) Returns: dict: 包含原始数据或AI分析结果的字典 """ # 解析事件数据 if isinstance(event, bytes): event = json.loads(event.decode('utf-8')) elif isinstance(event, str): event = json.loads(event) # 获取分析类型参数 analysis_type = event.get('analysis_type') if not analysis_type: query_params = event.get('queryParameters', {}) analysis_type = query_params.get('analysis_type') if not analysis_type: return {"status": "error", "message": "请求中缺少 'analysis_type' 参数"} # 获取其他可选参数 ai_result = event.get('ai_result') or event.get('queryParameters', {}).get('ai_result') generate_image = event.get('generate_image') or event.get('queryParameters', {}).get('generate_image') logging.info(f"处理请求: analysis_type={analysis_type}, ai_result={ai_result}, generate_image={generate_image}") try: # 获取原始数据 raw_data = get_raw_data(analysis_type) if not raw_data: return {"status": "error", "message": f"未找到 {analysis_type} 类型的数据"} # 构建基础返回结果 result = { "status": "success", "analysis_type": analysis_type, "data_count": len(raw_data), "raw_data": raw_data } # 如果需要AI分析结果 if ai_result == 'true': logging.info(f"开始执行AI分析: {analysis_type}") analysis_result = get_ai_analysis(analysis_type, raw_data) result["ai_analysis"] = analysis_result logging.info("AI分析完成") # 如果需要生成图片（目前只支持商品关联分析） if generate_image == 'true': if analysis_type == 'product_affinity': logging.info("开始生成商品组合图片") image_result = ai_service.generate_product_bundle_image(raw_data) result["generated_image"] = image_result logging.info("图片生成完成") else: result["image_warning"] = f"图片生成功能目前只支持 'product_affinity' 类型，当前类型为 '{analysis_type}'" return result except Exception as e: error_msg = f"处理 {analysis_type} 请求时发生错误: {str(e)}" logging.error(error_msg) return {"status": "error", "message": error_msg} def get_raw_data(analysis_type): """ 根据分析类型获取原始数据 Args: analysis_type (str): 分析类型 Returns: list: 原始数据列表，如果类型不支持则返回 None """ if analysis_type == 'top_selling_products': return db.get_top_selling_products(10) elif analysis_type == 'product_affinity': return db.get_product_affinity_analysis(min_occurrence=5, limit=15) elif analysis_type == 'inventory_status': return db.get_inventory_status_analysis(30) elif analysis_type == 'customer_behavior': return db.get_customer_behavior_analysis(20) elif analysis_type == 'seasonal_sales': return db.get_seasonal_sales_trend(10) else: logging.warning(f"不支持的分析类型: {analysis_type}") return None def get_ai_analysis(analysis_type, data): """ 根据分析类型和数据获取AI分析结果 Args: analysis_type (str): 分析类型 data (list): 原始数据 Returns: str: AI分析结果文本 """ if analysis_type == 'top_selling_products': return ai_service.analyze_top_selling_products(data) elif analysis_type == 'product_affinity': return ai_service.analyze_product_affinity(data) elif analysis_type == 'inventory_status': return ai_service.analyze_inventory(data) elif analysis_type == 'customer_behavior': return ai_service.analyze_customer_behavior(data) elif analysis_type == 'seasonal_sales': return ai_service.analyze_seasonal_trends(data) else: return f"不支持的分析类型: {analysis_type}"

值得注意的是，函数计算中需要依赖的类库需要手动安装，依赖的类库如下：

requirements.txt

pymssql==2.3.7 requests==2.31.0 dashscope==1.14.1 alibabacloud-tea-openapi==0.3.7

在打包上传代码之前，可以使用下面命令部署依赖包：

pip install -r requirements.txt -t .

配置环境变量&设置端口

函数中，我们部署环境变量，存入RDS SQL Server的地址、用户名与密码，以及百炼的API Key，这里当然也可以使用KMS，但为了DEMO简单起见，使用环境变量方式

通过创建触发器，允许函数通过HTTP请求触发，如下图：

为了DEMO简单，选择无需认证，但在生产环境中不能这么做。

验证函数

函数创建完成后，调用函数查看结果，看起来一切正常：

前端

前端使用Flask + 静态页面的方式进行。较为简单，核心代码如下：

from flask import Flask, render_template, jsonify, request import requests import logging import os # Initialize Flask app with explicit template folder app = Flask(__name__, template_folder=os.path.join(os.path.dirname(os.path.abspath(__file__)), 'templates')) app.secret_key = 'your-secret-key-here' # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # API基础URL API_BASE_URL = "https://sales-insights-zfdsymcpgw.cn-beijing.fcapp.run" # 分析类型配置 ANALYSIS_TYPES = { 'top_selling_products': { 'name': '热销商品分析', 'description': '展示销售额最高的商品及其详细信息，包括收入、客户数量、平均价格等', 'icon': '🏆', 'color': 'from-blue-400 to-blue-600', 'supports_image': False }, 'product_affinity': { 'name': '商品关联分析', 'description': '分析哪些商品经常一起被购买，用于推荐系统和交叉销售', 'icon': '🔗', 'color': 'from-green-400 to-green-600', 'supports_image': True }, 'inventory_status': { 'name': '库存状态分析', 'description': '分析当前库存状态和商品需求，识别库存风险', 'icon': '📦', 'color': 'from-yellow-400 to-yellow-600', 'supports_image': False }, 'customer_behavior': { 'name': '客户行为分析', 'description': '分析客户购买模式和价值，用于客户分层和营销策略', 'icon': '👥', 'color': 'from-purple-400 to-purple-600', 'supports_image': False }, 'seasonal_sales': { 'name': '季节性销售趋势', 'description': '分析商品在不同月份的销售表现，识别季节性模式', 'icon': '📊', 'color': 'from-red-400 to-red-600', 'supports_image': False } } @app.route('/') def index(): """主页面 - 展示所有分析类型""" return render_template('index.html', analysis_types=ANALYSIS_TYPES) @app.route('/analysis/<analysis_type>') def analysis_detail(analysis_type): """分析详情页面""" if analysis_type not in ANALYSIS_TYPES: return "分析类型不存在", 404 config = ANALYSIS_TYPES[analysis_type] return render_template('analysis.html', analysis_type=analysis_type, config=config) @app.route('/api/get_data/<analysis_type>') def get_raw_data(analysis_type): """获取原始数据的API端点""" try: url = f"{API_BASE_URL}?analysis_type={analysis_type}" response = requests.get(url, timeout=30) response.raise_for_status() return jsonify(response.json()) except requests.exceptions.RequestException as e: logger.error(f"获取原始数据失败: {str(e)}") return jsonify({"status": "error", "message": f"获取数据失败: {str(e)}"}), 500 @app.route('/api/get_ai_analysis/<analysis_type>') def get_ai_analysis(analysis_type): """获取AI分析结果的API端点""" try: url = f"{API_BASE_URL}?analysis_type={analysis_type}&ai_result=true" response = requests.get(url, timeout=60) response.raise_for_status() return jsonify(response.json()) except requests.exceptions.RequestException as e: logger.error(f"获取AI分析失败: {str(e)}") return jsonify({"status": "error", "message": f"获取AI分析失败: {str(e)}"}), 500 # 替换现有的 generate_image 路由 @app.route('/api/generate_image/<analysis_type>') def generate_image(analysis_type): """生成图片 - 调用包含AI分析和图片生成的完整API""" if analysis_type != 'product_affinity': return jsonify({"status": "error", "message": "图片生成功能只支持商品关联分析"}), 400 try: # 调用包含AI分析和图片生成的完整API url = f"{API_BASE_URL}?analysis_type={analysis_type}&ai_result=true&generate_image=true" logger.info(f"调用完整API（含图片生成）: {url}") response = requests.get(url, timeout=120) # 增加超时时间，因为图片生成需要更长时间 response.raise_for_status() result = response.json() # 只返回图片相关的数据 if result.get('status') == 'success' and result.get('generated_image'): return jsonify({ "status": "success", "generated_image": result['generated_image'] }) else: return jsonify({ "status": "error", "message": result.get('message', '图片生成失败') }), 500 except requests.exceptions.Timeout: logger.error("图片生成超时") return jsonify({"status": "error", "message": "图片生成超时，请稍后重试"}), 500 except requests.exceptions.RequestException as e: logger.error(f"生成图片失败: {str(e)}") return jsonify({"status": "error", "message": f"生成图片失败: {str(e)}"}), 500 @app.errorhandler(404) def not_found(error): return render_template('404.html'), 404 @app.errorhandler(500) def internal_error(error): return render_template('500.html'), 500 if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)

整体效果

前端启动后，下面是最终效果。

主页

热销商品分析

季节销售趋势分析

商品关联分析并生成关联商品的营销图

小结

本文展示了如何利用云原生技术栈构建现代化的智能销售分析平台，实现了传统数据库、无服务器计算和人工智能的深度融合。通过阿里云RDS for SQL Server、函数计算和百炼大模型的协同工作，构建了一个具备数据洞察和内容生成能力的端到端解决方案。

核心技术亮点：

• 无服务器架构：采用函数计算作为核心引擎，实现按需付费和自动弹性伸缩，显著降低运维成本
• AI原生设计：集成通义千问和通义万像，将数据分析结果转化为商业洞察和营销素材
• 完整分析链路：从购物篮分析、商品关联挖掘到AI营销素材生成的全流程自动化

这种架构模式特别适合中小企业快速构建数据分析能力，也为大型企业的数字化转型提供了轻量级的试点方案。后续可以扩展支持更多数据源、增加实时分析能力，或集成更多AI模型来满足不同业务场景的需求。

通过该Demo项目，现代云服务和AI技术的组合能够以较低的成本和复杂度，为传统业务带来智能化提升。