企业官网建设流程全解析

场景三：数据报表自动生成 Agent

第三个商业场景：数据报表。业务人员说"帮我看看上个月销售情况"，Agent 自动转 SQL、执行查询、生成图表、发送日报——不用再求数据团队排期。

本文导航

• 需求分析与 NL2SQL 挑战
• 自然语言转 SQL
• SQL 注入防御与数据脱敏
• pandas 分析与图表生成
• 日报自动发送
• 交付复盘
• 小结

需求分析与 NL2SQL 挑战

客户需求

某电商公司，运营团队 10 人，每天要看各种数据报表：

• 昨日销售总额、订单量、客单价
• 各品类销售排名
• 新用户 vs 老用户占比
• 退款率、退货原因分布

现状：

• 数据分析师 2 人，每天被需求淹没
• 运营人员不会 SQL，只能等排期
• 临时需求响应慢，影响决策

NL2SQL 的挑战

自然语言转 SQL（NL2SQL）是业界难题，主要挑战：

自然语言转 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_path=db_pathdefload_schema(self,tables:list[str]|None=None)->list[TableSchema]:"""加载表结构"""conn=sqlite3.connect(self.db_path)cursor=conn.cursor()# 获取所有表iftablesisNone:cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")tables=[row[0]forrowincursor.fetchall()]schemas=[]fortableintables:# 获取列信息cursor.execute(f"PRAGMA table_info({table})")columns=[{"name":row[1],"type":row[2],"nullable":notrow[3]}forrowincursor.fetchall()]# 获取示例数据（前 3 行）cursor.execute(f"SELECT * FROM{table}LIMIT 3")sample_data=[dict(zip([col["name"]forcolincolumns],row))forrowincursor.fetchall()]schemas.append(TableSchema(name=table,columns=columns,sample_data=sample_data,))conn.close()returnschemas

NL2SQL 生成

# 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_desc=""fortableinschemas:schema_desc+=f"\n表{table.name}:\n"forcolintable.columns:schema_desc+=f" -{col['name']}({col['type']})\n"schema_desc+=f" 示例数据:{table.sample_data[0]iftable.sample_dataelse'无'}\n"prompt=NL2SQL_PROMPT.format(schema=schema_desc,question=question,)response=client.chat.completions.create(model="deepseek-flash",messages=[{"role":"user","content":prompt}],temperature=0.1,# 低温度，更确定性)sql=(response.choices[0].message.contentor"").strip()# 清理可能的 markdown 标记sql=sql.replace("```sql","").replace("```","").strip()returnsql

SQL 注入防御与数据脱敏

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_upper=sql.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_row=row.copy()forfieldinsensitive_fields:iffieldinnew_row:value=str(new_row[field])# 手机号：保留前3后4iflen(value)==11andvalue.isdigit():new_row[field]=value[:3]+"****"+value[-4:]# 邮箱：保留首字母和域名elif"@"invalue:parts=value.split("@")new_row[field]=parts[0][0]+"***@"+parts[1]# 其他：保留首尾else:new_row[field]=value[0]+"***"+value[-1]iflen(value)>2else"***"masked.append(new_row)returnmasked

pandas 分析与图表生成

执行查询与分析

# report_agent/analyzer.pyimportsqlite3importpandasaspdimportmatplotlib.pyplotaspltfrompathlibimportPathclassDataAnalyzer:"""数据分析器"""def__init__(self,db_path:str,output_dir:str="reports"):self.db_path=db_path self.output_dir=Path(output_dir)self.output_dir.mkdir(exist_ok=True)defexecute_query(self,sql:str)->pd.DataFrame:"""执行 SQL 查询"""conn=sqlite3.connect(self.db_path)df=pd.read_sql_query(sql,conn)conn.close()returndfdefanalyze(self,question:str,df:pd.DataFrame)->str:"""用 DeepSeek 分析数据"""prompt=f"""根据以下数据回答问题。 问题：{question}数据：{df.to_string(index=False)}请给出简洁的分析结论，包括关键数字和趋势。 """response=client.chat.completions.create(model="deepseek-flash",messages=[{"role":"user","content":prompt}],)returnresponse.choices[0].message.contentor""defgenerate_chart(self,df:pd.DataFrame,chart_type:str="bar")->str:"""生成图表"""plt.rcParams['font.sans-serif']=['SimHei']# 中文字体plt.rcParams['axes.unicode_minus']=Falsefig,ax=plt.subplots(figsize=(10,6))ifchart_type=="bar"andlen(df.columns)>=2:x_col,y_col=df.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_type=="pie"andlen(df.columns)>=2:labels_col,values_col=df.columns[0],df.columns[1]ax.pie(df[values_col],labels=df[labels_col],autopct='%1.1f%%')else:ax.table(cellText=df.values,colLabels=df.columns)ax.axis('off')plt.tight_layout()chart_path=self.output_dir/f"chart_{pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')}.png"plt.savefig(chart_path,dpi=150)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:str="reports"):self.db_path=db_path self.analyzer=DataAnalyzer(db_path,output_dir)self.schema_loader=SchemaLoader(db_path)self.schemas=self.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. 生成 SQLsql=generate_sql(question,self.schemas)safe,reason=check_sql_safety(sql)ifnotsafe:report_lines.append(f"⚠️ SQL 安全检查未通过:{reason}")report_lines.append("")continue# 2. 执行查询try:df=self.analyzer.execute_query(sql)exceptExceptionase:report_lines.append(f"⚠️ 查询执行失败:{e}")report_lines.append("")continue# 3. 脱敏sensitive_fields=["phone","email","id_card"]df_dict=df.to_dict("records")masked=mask_sensitive_data(df_dict,sensitive_fields)df=pd.DataFrame(masked)# 4. 分析analysis=self.analyzer.analyze(question,df)report_lines.append(analysis)report_lines.append("")# 5. 图表（如果有数值列）iflen(df.columns)>=2anddf.iloc[:,1].dtypein["int64","float64"]:chart_path=self.analyzer.generate_chart(df)report_lines.append(f"![图表]({chart_path})")report_lines.append("")# 保存报告report_path=self.analyzer.output_dir/f"daily_report_{datetime.now().strftime('%Y%m%d')}.md"report_path.write_text("\n".join(report_lines),encoding="utf-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_host=smtp_host self.smtp_port=smtp_port self.username=username self.password=passworddefsend_report(self,to_emails:list[str],report_path:str,subject:str="数据日报"):"""发送日报邮件"""msg=MIMEMultipart("related")msg["Subject"]=subject msg["From"]=self.username msg["To"]=", ".join(to_emails)# 读取报告内容report_content=Path(report_path).read_text(encoding="utf-8")# 简单 Markdown 转 HTML（实际可用 markdown 库）html_content=f"<pre>{report_content}</pre>"msg.attach(MIMEText(html_content,"html","utf-8"))# 附加图表forimg_pathinPath(report_path).parent.glob("*.png"):withopen(img_path,"rb")asf:img=MIMEImage(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 个月后：

报价

开发费用： - NL2SQL 核心：3 周 × 7000 元/周 = 21000 元 - 图表生成：1 周 × 7000 元/周 = 7000 元 - 邮件发送：0.5 周 × 7000 元/周 = 3500 元 - 安全加固：1 周 × 7000 元/周 = 7000 元 - 合计：38500 元 运维费用（年）： - API 调用费：约 2000 元/年 - 服务器：500 元/月 × 12 = 6000 元/年 - 维护支持：6000 元/年 - 合计：约 14000 元/年 客户最终成交：开发 38500 元 + 年运维 14000 元

关键经验

1. Schema 注入是 NL2SQL 的关键：表结构描述越清晰，生成的 SQL 越准确。示例数据能帮助模型理解字段含义。

2. 安全检查不能省：SQL 注入、敏感数据泄露都是 P0 事故。白名单 + 黑名单 + 脱敏三管齐下。

3. 图表要简单实用：柱状图、饼图覆盖 80% 场景，不要追求花哨。

小结

1. NL2SQL：Schema 注入 + 低温度 + 清晰提示词，准确率可达 80%+。
2. SQL 安全：黑名单（DROP/DELETE）+ 白名单（允许的表）+ 脱敏（手机号/邮箱）。
3. pandas 分析：执行查询后让 DeepSeek 分析数据，给出结论。
4. 图表生成：matplotlib 柱状图/饼图，简单实用。
5. 日报自动发送：Markdown 报告 + 图表附件 + 邮件发送。
6. 效果显著：日报生成时间 -96%，数据分析师工作量 -60%。
7. 报价合理：开发 38500 + 年运维 14000，ROI 明显。

下节预告

第四个商业场景：自动化测试生成器——读源码、推断行为、生成 pytest 用例、跑通验证、覆盖率报告。还有一个"生成-执行-修复"自愈循环，让测试用例自己修自己。

如果觉得本文对你有帮助，欢迎点赞、收藏、关注三连！
本系列持续更新中，关注不迷路~