这个Demo的缘起挺偶然的。我帮朋友公司做运维咨询的时候看到他们每个周一上午都在“手工造周报”——把CRM导出的Excel插进模板再人工写上“本周销量下滑3%建议关注华东区”。整个过程两三个小时起步结论还经常对不上数据。我就想能不能把“查数据”和“写结论”这两步全自动化于是就有了这套基于阿里云RDS SQL Server、函数计算和通义AI的智能销售分析平台Demo。整个Demo的技术链路不算复杂RDS SQL Server负责存数据、算聚合函数计算扛起定时调度和HTTP请求通义AI负责把数字翻译成人话。最终输出是一份每天自动生成、可直接打开的HTML销售洞察日报。适合正在做BI报表、数据中台或者想尝试AI数据落地的朋友参考尤其适合被周报月报折磨的团队。1. 先用一个场景想清楚这套Demo到底解决什么问题1.1 周报背后的真实痛点先说朋友公司的情况。他们有CRM系统也有ERP导出数据但每周做销售分析时运营要手动干三件事把Excel数据复制到固定模板里用透视表算环比和排行榜再根据数字“编”一段结论。这里的“编”不是乱写而是靠人的经验去解读数字——比如某区域销量掉了可能是因为上个季度有个大促基数高了。这个流程有三个明显的毛病。第一耗时一个人小半天就耗进去了第二口径不统一有人算销售额含退款有人不含第三结论滞后周一看到的还是上周六的数据再一汇总行动已经慢了半拍。所以我想做一个Demo验证一下数据到结论这条链路能不能用云上托管服务和AI完全接起来让日报在每天早上自动生成并且结论部分由AI基于同样的数据输出口径一致。这套Demo的目标也很具体每天9点自动跑一次先把RDS SQL Server里的销售订单聚合好再调用通义AI生成趋势分析、异常提醒和建议动作最后输出一份可以直接在浏览器打开的HTML日报。整个过程不需要人参与延迟控制在分钟级。1.2 为什么是这三件套而不是别的方案很多人看到“SQL Server”会先想到自建实例看到“AI分析”会想到自己微调模型。但在这个场景下三件套的组合是性价比最高的。方案优点缺点适合场景自建SQL Server 定时任务可控、许可费用固定备份、高可用、监控全要自己搞私有化、长期固定环境ECS自装全套 脚本调度灵活、可以随便装东西运维成本高弹性差特殊硬件或老系统兼容RDS SQL Server 函数计算 通义AI托管、按量付费、免运维有一定学习成本云上费用持续产生云上快速落地、事件驱动任务选择RDS SQL Server而不是自建核心原因是备份和可用性不用自己操心。SQL Server的备份策略、日志管理本来就很讲究云上RDS直接帮你把自动化备份开了还能一键恢复到任意时间点。函数计算的逻辑也类似——这个任务是典型的“定时触发”一周可能只跑几次用一台ECS常驻等cron太浪费函数计算的按量付费模式几乎是为此设计的。至于通义AI中文语义理解能力强API调用简单和阿里云生态无缝集成做数据解读这种场景很合适。2. 环境准备把三个云服务搭起来2.1 RDS SQL Server实例创建与关键参数创建RDS SQL Server实例的过程在控制台几步就能完成但有几个参数值得提前想清楚。地域要选离业务最近或者和函数计算同地域的区域。跨地域调用虽然也能通但延迟和费用都不划算。实例规格我选的是SQL Server 2022 Standard2核4G通用型对Demo来说绰绰有余。如果你只是做功能验证最低配也够用但千万别在生产环境用Express版它单库大小上限是10GB而且没有SQL Agent很多自动化操作会受限。网络这块创建实例时默认会分配一个VPC网络记下内网连接地址函数计算后面要绑同一个VPC才能走内网访问。白名单一开始只加你的本机IP方便用SSMS或Navicat调试后面再放函数计算的VPC网段。数据库账号建议单独建不要直接用高权限账号跑业务。我建了一个名叫rds_sales的账号只授权了sales_demo这个库的读写权限。RDS默认会开启强密码策略初始创建账号时密码必须满足复杂度要求。网上经常有人搜“SQL Server 2022 关闭密码策略”如果你确实想在测试环境关掉密码过期检查可以在实例参数组里调整但我不建议这么干云上数据库的安全基线还是保留比较好。连接验证用命令行最直观sqlcmd -S rdsxxxx.sqlserver.rds.aliyuncs.com,1433 -U rds_sales -P yourpass -d sales_demo能进到SQLCMD交互界面就说明网络、账号、白名单都通了。2.2 函数计算服务与依赖层函数计算我选了Python 3.9运行时内存512MB超时时间60秒。创建服务时有一个关键步骤绑定VPC。只有绑定了RDS所在VPC和交换机函数计算才能通过内网地址读写RDS数据否则每次查库都走公网延迟高不说还多一道安全风险。依赖安装是很多人没踩过的坑。函数计算的运行环境里默认不会有pymssql和dashscope你得自己打包上传。我测试过pyodbc在函数计算里的方案需要额外装ODBC驱动非常折腾改用pymssql就省心很多这个库自带FreeTDS函数计算里直接能跑。打包命令如下pip install pymssql dashscope -t python/ zip -r pymssql_layer.zip python/然后在函数计算控制台的“层”这里新建一个层上传这个zip再绑定到函数上。注意函数计算跑在Linux环境pip安装时别加--platform参数直接在当前环境装好打包就行。Python的C扩展会带有当前系统GLIBC版本的依赖建议在本地用干净环境打包。环境变量统一放在函数的配置里包括DB_HOST、DB_USER、DB_PASSWORD、DB_NAME、DASHSCOPE_API_KEY。代码里不要写死任何密钥。2.3 通义AI模型接入通义AI我通过百炼平台接入。在百炼控制台创建API-KEY后本地先做一次连通性测试import dashscope from dashscope import Generation dashscope.api_key sk-xxxx resp Generation.call( modelqwen-plus, messages[{role: user, content: 你好}], ) print(resp.output.choices[0].message.content)模型选择上qwen-plus对分析类任务更稳qwen-turbo更快更便宜。Demo场景我用的qwen-plus后面如果发现函数执行经常超时再降级到qwen-turbo也行。如果你习惯OpenAI的SDK也可以走兼容模式把base_url设置为https://dashscope.aliyuncs.com/compatible-mode/v1效果一样代码更通用。3. 数据层表设计与聚合SQL3.1 销售订单表与模拟数据整个Demo的数据模型不需要很复杂三张表就够产品表、客户表、订单明细表。产品表和客户表先造一批数据订单表用递归CTE随机生成近90天的模拟记录。CREATE TABLE Products ( ProductID INT PRIMARY KEY, ProductName NVARCHAR(100), Category NVARCHAR(50), UnitPrice DECIMAL(10,2) ); CREATE TABLE Customers ( CustomerID INT PRIMARY KEY, CustomerName NVARCHAR(100), Level NVARCHAR(20) ); CREATE TABLE SalesOrders ( OrderID INT PRIMARY KEY, OrderDate DATETIME, CustomerID INT, ProductID INT, Quantity INT, Amount DECIMAL(10,2), Region NVARCHAR(50), SalesPerson NVARCHAR(50) );生成模拟数据时最容易犯的错是图省事用RAND()。在SQL Server的SELECT语句里同一行的多个RAND()调用会返回不同值导致数据分布看起来像随机实际上计划不稳定。我推荐用ABS(CHECKSUM(NEWID())) % N这种写法每次取值相对独立。WITH Orders AS ( SELECT TOP 10000 ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) AS OrderID, DATEADD(day, -ABS(CHECKSUM(NEWID())) % 90, GETDATE()) AS OrderDate, ABS(CHECKSUM(NEWID())) % 100 1 AS CustomerID, ABS(CHECKSUM(NEWID())) % 20 1 AS ProductID, ABS(CHECKSUM(NEWID())) % 5 1 AS Quantity, ABS(CHECKSUM(NEWID())) % 4 1 AS RegionID, ... FROM sys.all_columns a CROSS JOIN sys.all_columns b ) INSERT INTO SalesOrders (OrderID, OrderDate, CustomerID, ProductID, Quantity, Amount, Region, SalesPerson) SELECT OrderID, OrderDate, CustomerID, ProductID, Quantity, Quantity * (SELECT UnitPrice FROM Products WHERE ProductID Orders.ProductID), CASE RegionID WHEN 1 THEN 华东 WHEN 2 THEN 华南 WHEN 3 THEN 华北 ELSE 西南 END, ... FROM Orders;金额字段不要直接随机生成而是用Quantity * UnitPrice算出来这样数据在业务上才自洽。3.2 多维聚合SQL与环比计算核心报表是近14天按日聚合的销售额、订单数、客单价以及环比变化率。这里要用窗口函数LAG取前一天的值注意先聚合再算环比否则行数会对不上。WITH Daily AS ( SELECT CONVERT(varchar(10), OrderDate, 120) AS Day, COUNT(DISTINCT OrderID) AS OrderCnt, SUM(Amount) AS SalesAmount, SUM(Amount) / COUNT(DISTINCT OrderID) AS AvgOrderValue FROM SalesOrders WHERE OrderDate DATEADD(day, -13, GETDATE()) GROUP BY CONVERT(varchar(10), OrderDate, 120) ) SELECT Day, OrderCnt, SalesAmount, AvgOrderValue, LAG(SalesAmount, 1) OVER (ORDER BY Day) AS PrevSales, CASE WHEN LAG(SalesAmount, 1) OVER (ORDER BY Day) 0 THEN NULL ELSE (SalesAmount - LAG(SalesAmount, 1) OVER (ORDER BY Day)) * 100.0 / LAG(SalesAmount, 1) OVER (ORDER BY Day) END AS DayOverDayPct FROM Daily ORDER BY Day;再给一个品类×区域的排行查询用来做AI的“结构分析”输入SELECT TOP 10 Category, Region, SUM(Amount) AS SalesAmount, SUM(Quantity) AS TotalQty FROM SalesOrders WHERE OrderDate DATEADD(day, -7, GETDATE()) GROUP BY Category, Region ORDER BY SalesAmount DESC;这两个查询我直接封装成存储过程函数计算只调用存储过程不传原始SQL。好处有两个一是业务口径能收口改动口径时只改存储过程不用改应用代码二是避免拼接SQL带来的注入风险。4. 核心链路函数计算的代码实现4.1 触发器配置定时与HTTP两种函数计算支持两种触发器我两个都配了。定时触发器用cron表达式0 0 9 * * *表示每天早上9点执行。为什么cron里没写秒因为函数计算定时触发器的cron格式是六段前两位分别是秒和分我这里0 0 9 * * *对应“每天的9点0分0秒”。需要留意的是时区问题如果函数计算控制台显示的时区是UTC8就不用换算如果发现任务跑的时间比预期早了8小时先去触发器设置里看时区说明。这个坑我踩过一次。HTTP触发器是给“即时查询”用的。创建函数时选择HTTP触发器路径/analyze请求方式GET和POST都行。这样你随时浏览器敲一下https://xxxx.fc.aliyuncs.com/analyze就能手动触发一次分析和报告生成不用干等定时任务。4.2 代码结构查询、组装、调AI、拼HTML整个handler的流程分四步连接RDS执行聚合查询把结果格式化成结构化文本调用通义AI生成洞察最后把AI返回的JSON渲染成HTML。import json import os import pymssql import dashscope from dashscope import Generation dashscope.api_key os.environ.get(DASHSCOPE_API_KEY) def get_conn(): return pymssql.connect( serveros.environ[DB_HOST], useros.environ[DB_USER], passwordos.environ[DB_PASSWORD], databaseos.environ[DB_NAME], port1433, login_timeout15, timeout30, charsetutf8 ) def query_daily(): sql 这里放日维度聚合SQL with get_conn() as conn: with conn.cursor() as cur: cur.execute(sql) cols [d[0] for d in cur.description] rows [dict(zip(cols, r)) for r in cur.fetchall()] return rows def query_top(): sql 这里放品类×区域TOP10 SQL with get_conn() as conn: with conn.cursor() as cur: cur.execute(sql) cols [d[0] for d in cur.description] rows [dict(zip(cols, r)) for r in cur.fetchall()] return rows def build_data_text(daily, top): text 近7天每日数据\n for r in daily: text f{r[Day]} 订单{r[OrderCnt]} 销售额{r[SalesAmount]} 客单价{r[AvgOrderValue]:.2f}\n text \n最近7天品类×区域\n for r in top: text f{r[Category]} / {r[Region]} 销售额{r[SalesAmount]}\n return text def call_ai(data_text): resp Generation.call( modelqwen-plus, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: data_text} ], temperature0.2, result_formatmessage ) return resp.output.choices[0].message.content def handler(event, context): daily query_daily() top query_top() data_text build_data_text(daily, top) ai_output call_ai(data_text) insight extract_json(ai_output) html render_html(insight, daily, top) return { statusCode: 200, headers: {Content-Type: text/html; charsetutf-8}, body: html }代码看起来不长但有几个细节容易出错。第一pymssql.connect的返回对象要确保用完关闭。上面用了with get_conn() as conn的写法配合异常处理回收连接。实际使用中如果函数计算实例被复用Python的全局变量会保留但因为with结构每次都重新建立连接短连接模式在低并发下完全没问题也省心。第二result_formatmessage会返回OpenAI风格的消息结构取内容时用resp.output.choices[0].message.content。如果返回里还带finish_reason字段可以在代码里判断一下防止因为token截断导致JSON不完整。第三HTTP触发器返回的body如果是HTML一定要设置Content-Type为text/html; charsetutf-8否则浏览器会把HTML当纯文本显示。4.3 把AI输出渲染成HTML我建议让AI输出结构化JSON而不是直接输出HTML。原因很简单模型生成的HTML排版经常不可控今天带个红色的标题明天CSS位置全乱了。JSON的字段是固定的前端想怎么渲染都行后续如果要做图表、做邮件推送都很方便。def extract_json(text): start text.find({) end text.rfind(}) 1 if start 0 and end start: return json.loads(text[start:end]) raise ValueError(no json found in AI output)def render_html(insight, daily, top): summary insight.get(summary, ) trend insight.get(trend, {}) insights insight.get(insights, []) risks insight.get(risks, []) actions insight.get(actions, []) items_i .join(fli{x}/li for x in insights) items_r .join(fli{x}/li for x in risks) items_a .join(fli{x}/li for x in actions) # 用内联样式不依赖CDN离线也能看 return f html headmeta charsetutf-8title销售分析日报/title/head body stylefont-family: sans-serif; max-width: 900px; margin: 40px auto; h2销售智能分析日报/h2 pb一句话总览/b{summary}/p pb趋势/b{trend.get(direction, )} {trend.get(rate, )}%/p p{trend.get(analysis, )}/p h3核心发现/h3ul{items_i}/ul h3风险提示/h3ul{items_r}/ul h3建议动作/h3ul{items_a}/ul /body/html HTML解析时会直接插入JSON字段要给AI的Prompt加一条“禁止在JSON里使用HTML标签”防止模型输出的字段里带script之类的危险内容。5. AI分析层Prompt设计与模型调优5.1 让AI读得懂销售数据的Prompt模板模型能不能输出靠谱结论七成功夫在Prompt。我在Demo里的System Prompt是这样写的你是一名有10年经验的销售运营分析师。你会收到最近14天的销售数据按日聚合以及最近7天品类×区域的销售额。 请完成以下任务 1. 整体趋势判断最近7天相比前7天的变化幅度结合实际运营常识分析可能的原因 2. 异常信号识别找出哪些品类/区域的销售额波动超过20%给出合理的业务猜测 3. 风险点提示连续下降、客单价异常、品类过于集中等 4. 建议动作给出下个周期可执行的动作列表按优先级排序。 要求 - 使用通俗中文不得编造输入数据中没有出现的数字 - 输出必须是JSON对象字段限定为summary, trend, insights, risks, actions - insights和risks每条不超过40个字 - 禁止在JSON字段值中插入任何HTML标签。用户消息部分放的是build_data_text拼接的文本。这里有个细节数据文本的行很长时模型容易丢上下文建议在每行前加字段名比如“日期2025-06-01 订单120 销售额15800”而不是只给一堆用逗号分隔的数字。关于temperature参数我实测下来设0.2比较合适。温度太高AI会用词华丽但结论飘今天说“大幅增长”明天同样的数据说“小幅度上扬”前后口径不一致。温度在0.1到0.3之间分析类任务会比较稳定。5.2 结构化JSON输出与前端渲染qwen-plus在严格JSON输出上通常表现不错但偶尔还是会“发挥失常”——答案是正常JSON但前面加一句“这是分析结果”。extract_json函数就是为了兜底这种情况。如果JSON解析直接抛异常建议在handler里加一个降级逻辑把AI原始输出包裹成{summary: AI输出格式异常请人工查看原始数据, raw: a_output}保证页面不至于白屏。另外AI生成的JSON里有小数点、百分号之类的数字最好在后端做一次round()保留一位小数再放进HTML。模型可能输出“12.345678%”直接展示不专业也容易被老板追问。6. 常见问题与排查实录6.1 高频报错速查表搭建这套Demo时我前前后后踩了不少坑这里把典型问题和排查思路整理成一张表新来的同学直接对着找原因。现象原因与排查解决“已成功与服务器建立连接但在登录前握手失败”SSL/TLS协商失败。驱动默认要求加密连接RDS侧强制开启TLSSSMS连接时勾选“Encrypt connection”Navicat可将加密选项关闭仅限测试环境确认客户端TLS版本不低于1.0账号密码正确但登录报18456密码策略过期或账号权限不足去RDS控制台重置密码检查账号是否授权了目标数据库的读写权限pymssql报“DB-Lib error message 20002”连接超时白名单没放通或VPC不通先查RDS白名单是否包含函数计算所在VPC网段再确认函数计算服务有没有绑定RDS的VPC和交换机函数计算首次调用特别慢冷启动重新拉取环境、加载依赖在函数计算配置预留实例精简依赖包体积HTTP触发器用单实例并发减少重复初始化AI调用偶发超时模型推理耗时超过函数超时时间把函数超时调整到120秒换qwen-turbo缩小输入数据的行数上限定时任务时间不对时区设置不一致确认函数计算定时触发器的时区说明通常控制台显示UTC8若仍不对改用0 0 9 * * *并用本地测试函数打印时间SQL Server日志文件持续增大日志未截断或数据库恢复模式为FULL且无定期备份查看DBCC SQLPERF(LOGSPACE)确认增长比例RDS实例检查备份策略FULL恢复模式下日志要靠日志备份截断不要在业务高峰期频繁DBCC SHRINKFILE6.2 更稳的工程化方向如果你打算把这个Demo往生产方向推有几件事值得提前做。第一加一层缓存。同一个分析周期内多次请求AI分析和SQL查询结果其实是一样的。可以在函数计算里加个Redis缓存键是日期值是生成好的HTML或JSONTTL设成一天。省掉不必要的AI调用成本直接下来。第二把AI分析结果存表。我建议在RDS里建一张SalesAnalysisLog记录每次分析的时间、输入摘要、AI原始输出、渲染后的HTML。这样出了问题能回溯也方便做历史趋势对比。第三失败告警。函数计算的定时触发器如果执行失败控制台没有默认的主动通知。我后来加了一个钉钉机器人Webhookhandler里包一层try-except出错时把错误信息发到钉钉群才知道任务是不是真的每天跑了。第四Prompt版本管理。别把Prompt直接写在代码里。用环境变量或者专门的管理表存Prompt模板线上要调整措辞时只改配置不重新发版。AI应用的Prompt迭代频率远高于代码这个习惯越早养成越好。关于成本这套Demo按每天跑一次、每次消耗约600个token的AI分析量来算函数计算一个月的费用几乎可以忽略RDS最低配一个月几十元加上通义AI的调用费加起来在百元以内。相比人工做周报的时间成本这点投入非常划算。最后分享一个最深的体会这套链路真正难的不是连数据库也不是调API而是把“业务口径”翻译成“查询逻辑”和“Prompt约束”。同一个“销售额”在财务眼里不含退款在运营眼里含优惠券在AI眼里如果不说清楚它就会自己脑补。所以我在SQL里统一把Amount定义为“订单实付金额”在Prompt里也重复了一遍“所有结论基于Amount字段”这样AI才没有自由发挥的空间。如果你也想复现我建议按这个顺序走先在SSMS里把第3章的聚合SQL跑通再用本地Python脚本连着RDS和通义AI把完整链路调通最后才搬到函数计算上做触发器。一步到位的话排查问题时会分不清是网络不通还是Prompt写得不对容易把自己绕晕。