场景三数据报表自动生成 Agent第三个商业场景数据报表。业务人员说帮我看看上个月销售情况Agent 自动转 SQL、执行查询、生成图表、发送日报——不用再求数据团队排期。本文导航需求分析与 NL2SQL 挑战自然语言转 SQLSQL 注入防御与数据脱敏pandas 分析与图表生成日报自动发送交付复盘小结需求分析与 NL2SQL 挑战客户需求某电商公司运营团队 10 人每天要看各种数据报表昨日销售总额、订单量、客单价各品类销售排名新用户 vs 老用户占比退款率、退货原因分布现状数据分析师 2 人每天被需求淹没运营人员不会 SQL只能等排期临时需求响应慢影响决策NL2SQL 的挑战自然语言转 SQLNL2SQL是业界难题主要挑战挑战说明应对策略表结构复杂几十张表、上百字段提供 schema 摘要歧义表达上个月是自然月还是30天明确约定 确认机制复杂查询JOIN、子查询、窗口函数分步生成 验证数据安全敏感字段不能随便查权限控制 脱敏自然语言转 SQL表 Schema 注入# report_agent/schema_loader.pyimportsqlite3frompydanticimportBaseModelclassTableSchema(BaseModel):表结构name:strcolumns:list[dict]# [{name: id, type: INTEGER, comment: 主键}]sample_data:list[dict]# 示例数据帮助理解classSchemaLoader:Schema 加载器def__init__(self,db_path:str):self.db_pathdb_pathdefload_schema(self,tables:list[str]|NoneNone)-list[TableSchema]:加载表结构connsqlite3.connect(self.db_path)cursorconn.cursor()# 获取所有表iftablesisNone:cursor.execute(SELECT name FROM sqlite_master WHERE typetable)tables[row[0]forrowincursor.fetchall()]schemas[]fortableintables:# 获取列信息cursor.execute(fPRAGMA table_info({table}))columns[{name:row[1],type:row[2],nullable:notrow[3]}forrowincursor.fetchall()]# 获取示例数据前 3 行cursor.execute(fSELECT * FROM{table}LIMIT 3)sample_data[dict(zip([col[name]forcolincolumns],row))forrowincursor.fetchall()]schemas.append(TableSchema(nametable,columnscolumns,sample_datasample_data,))conn.close()returnschemasNL2SQL 生成# report_agent/nl2sql.pyfromdeep_pilot.clientimportclientfromreport_agent.schema_loaderimportSchemaLoader NL2SQL_PROMPT你是一个 SQL 专家。根据用户的自然语言问题和数据库表结构生成正确的 SQL 查询语句。 数据库表结构 {schema} 用户问题{question} 要求 1. 只生成 SQL 查询语句不要解释 2. 使用标准 SQL 语法 3. 如果问题模糊生成最合理的解释对应的 SQL 4. 日期字段使用 ISO 格式YYYY-MM-DD 5. 金额字段保留 2 位小数 SQL: defgenerate_sql(question:str,schemas:list[TableSchema])-str:自然语言转 SQL# 构建 schema 描述schema_descfortableinschemas:schema_descf\n表{table.name}:\nforcolintable.columns:schema_descf -{col[name]}({col[type]})\nschema_descf 示例数据:{table.sample_data[0]iftable.sample_dataelse无}\npromptNL2SQL_PROMPT.format(schemaschema_desc,questionquestion,)responseclient.chat.completions.create(modeldeepseek-flash,messages[{role:user,content:prompt}],temperature0.1,# 低温度更确定性)sql(response.choices[0].message.contentor).strip()# 清理可能的 markdown 标记sqlsql.replace(sql,).replace(,).strip()returnsqlSQL 注入防御与数据脱敏SQL 安全检查# report_agent/sql_guard.pyimportre# 危险关键词黑名单DANGEROUS_KEYWORDS[r\bDROP\b,r\bDELETE\b,r\bTRUNCATE\b,r\bINSERT\b,r\bUPDATE\b,r\bALTER\b,r\bCREATE\b,r\bEXEC\b,r\bUNION\b,]# 允许的表白名单ALLOWED_TABLES[orders,users,products,categories]defcheck_sql_safety(sql:str)-tuple[bool,str]:检查 SQL 是否安全sql_uppersql.upper()# 检查危险关键词forpatterninDANGEROUS_KEYWORDS:ifre.search(pattern,sql_upper):returnFalse,f包含危险关键词:{pattern}# 检查表名白名单# 简单检查SQL 中出现的表名必须在白名单中# 实际应该用 SQL 解析库fortableinALLOWED_TABLES:iftableinsql_lower:sql.lower():continue# 检查是否有多条语句if;insql.rstrip(;):returnFalse,不允许多条语句returnTrue,defmask_sensitive_data(data:list[dict],sensitive_fields:list[str])-list[dict]:脱敏敏感字段masked[]forrowindata:new_rowrow.copy()forfieldinsensitive_fields:iffieldinnew_row:valuestr(new_row[field])# 手机号保留前3后4iflen(value)11andvalue.isdigit():new_row[field]value[:3]****value[-4:]# 邮箱保留首字母和域名elifinvalue:partsvalue.split()new_row[field]parts[0][0]***parts[1]# 其他保留首尾else:new_row[field]value[0]***value[-1]iflen(value)2else***masked.append(new_row)returnmaskedpandas 分析与图表生成执行查询与分析# report_agent/analyzer.pyimportsqlite3importpandasaspdimportmatplotlib.pyplotaspltfrompathlibimportPathclassDataAnalyzer:数据分析器def__init__(self,db_path:str,output_dir:strreports):self.db_pathdb_path self.output_dirPath(output_dir)self.output_dir.mkdir(exist_okTrue)defexecute_query(self,sql:str)-pd.DataFrame:执行 SQL 查询connsqlite3.connect(self.db_path)dfpd.read_sql_query(sql,conn)conn.close()returndfdefanalyze(self,question:str,df:pd.DataFrame)-str:用 DeepSeek 分析数据promptf根据以下数据回答问题。 问题{question}数据{df.to_string(indexFalse)}请给出简洁的分析结论包括关键数字和趋势。 responseclient.chat.completions.create(modeldeepseek-flash,messages[{role:user,content:prompt}],)returnresponse.choices[0].message.contentordefgenerate_chart(self,df:pd.DataFrame,chart_type:strbar)-str:生成图表plt.rcParams[font.sans-serif][SimHei]# 中文字体plt.rcParams[axes.unicode_minus]Falsefig,axplt.subplots(figsize(10,6))ifchart_typebarandlen(df.columns)2:x_col,y_coldf.columns[0],df.columns[1]ax.bar(df[x_col].astype(str),df[y_col])ax.set_xlabel(x_col)ax.set_ylabel(y_col)elifchart_typepieandlen(df.columns)2:labels_col,values_coldf.columns[0],df.columns[1]ax.pie(df[values_col],labelsdf[labels_col],autopct%1.1f%%)else:ax.table(cellTextdf.values,colLabelsdf.columns)ax.axis(off)plt.tight_layout()chart_pathself.output_dir/fchart_{pd.Timestamp.now().strftime(%Y%m%d_%H%M%S)}.pngplt.savefig(chart_path,dpi150)plt.close()returnstr(chart_path)日报自动发送生成日报# report_agent/daily_report.pyfromdatetimeimportdatetime,timedeltafromreport_agent.nl2sqlimportgenerate_sqlfromreport_agent.sql_guardimportcheck_sql_safety,mask_sensitive_datafromreport_agent.analyzerimportDataAnalyzerfromreport_agent.schema_loaderimportSchemaLoaderclassDailyReportGenerator:日报生成器def__init__(self,db_path:str,output_dir:strreports):self.db_pathdb_path self.analyzerDataAnalyzer(db_path,output_dir)self.schema_loaderSchemaLoader(db_path)self.schemasself.schema_loader.load_schema()defgenerate(self,questions:list[str])-str:生成日报report_lines[f# 数据日报 -{datetime.now().strftime(%Y-%m-%d)},,]forquestioninquestions:report_lines.append(f##{question})report_lines.append()# 1. 生成 SQLsqlgenerate_sql(question,self.schemas)safe,reasoncheck_sql_safety(sql)ifnotsafe:report_lines.append(f⚠️ SQL 安全检查未通过:{reason})report_lines.append()continue# 2. 执行查询try:dfself.analyzer.execute_query(sql)exceptExceptionase:report_lines.append(f⚠️ 查询执行失败:{e})report_lines.append()continue# 3. 脱敏sensitive_fields[phone,email,id_card]df_dictdf.to_dict(records)maskedmask_sensitive_data(df_dict,sensitive_fields)dfpd.DataFrame(masked)# 4. 分析analysisself.analyzer.analyze(question,df)report_lines.append(analysis)report_lines.append()# 5. 图表如果有数值列iflen(df.columns)2anddf.iloc[:,1].dtypein[int64,float64]:chart_pathself.analyzer.generate_chart(df)report_lines.append(f)report_lines.append()# 保存报告report_pathself.analyzer.output_dir/fdaily_report_{datetime.now().strftime(%Y%m%d)}.mdreport_path.write_text(\n.join(report_lines),encodingutf-8)returnstr(report_path)邮件发送# report_agent/mailer.pyimportsmtplibfromemail.mime.textimportMIMETextfromemail.mime.multipartimportMIMEMultipartfromemail.mime.imageimportMIMEImagefrompathlibimportPathclassReportMailer:报告邮件发送器def__init__(self,smtp_host:str,smtp_port:int,username:str,password:str):self.smtp_hostsmtp_host self.smtp_portsmtp_port self.usernameusername self.passwordpassworddefsend_report(self,to_emails:list[str],report_path:str,subject:str数据日报):发送日报邮件msgMIMEMultipart(related)msg[Subject]subject msg[From]self.username msg[To], .join(to_emails)# 读取报告内容report_contentPath(report_path).read_text(encodingutf-8)# 简单 Markdown 转 HTML实际可用 markdown 库html_contentfpre{report_content}/premsg.attach(MIMEText(html_content,html,utf-8))# 附加图表forimg_pathinPath(report_path).parent.glob(*.png):withopen(img_path,rb)asf:imgMIMEImage(f.read())img.add_header(Content-ID,f{img_path.name})msg.attach(img)# 发送withsmtplib.SMTP_SSL(self.smtp_host,self.smtp_port)asserver:server.login(self.username,self.password)server.sendmail(self.username,to_emails,msg.as_string())交付复盘实际效果上线 1 个月后指标上线前上线后变化日报生成时间2 小时人工5 分钟-96%临时需求响应1-2 天排期即时-数据分析师工作量100%40%-60%运营满意度3.5/54.5/529%报价开发费用 - NL2SQL 核心3 周 × 7000 元/周 21000 元 - 图表生成1 周 × 7000 元/周 7000 元 - 邮件发送0.5 周 × 7000 元/周 3500 元 - 安全加固1 周 × 7000 元/周 7000 元 - 合计38500 元 运维费用年 - API 调用费约 2000 元/年 - 服务器500 元/月 × 12 6000 元/年 - 维护支持6000 元/年 - 合计约 14000 元/年 客户最终成交开发 38500 元 年运维 14000 元关键经验Schema 注入是 NL2SQL 的关键表结构描述越清晰生成的 SQL 越准确。示例数据能帮助模型理解字段含义。安全检查不能省SQL 注入、敏感数据泄露都是 P0 事故。白名单 黑名单 脱敏三管齐下。图表要简单实用柱状图、饼图覆盖 80% 场景不要追求花哨。小结NL2SQLSchema 注入 低温度 清晰提示词准确率可达 80%。SQL 安全黑名单DROP/DELETE 白名单允许的表 脱敏手机号/邮箱。pandas 分析执行查询后让 DeepSeek 分析数据给出结论。图表生成matplotlib 柱状图/饼图简单实用。日报自动发送Markdown 报告 图表附件 邮件发送。效果显著日报生成时间 -96%数据分析师工作量 -60%。报价合理开发 38500 年运维 14000ROI 明显。下节预告第四个商业场景自动化测试生成器——读源码、推断行为、生成 pytest 用例、跑通验证、覆盖率报告。还有一个生成-执行-修复自愈循环让测试用例自己修自己。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~