简介这份PDF资源面向具备一定Python与数据库基础的数据分析人员、后端开发者及数据产品经理聚焦如何借助Dify平台与DeepSeek大模型搭建智能数据分析助手把自然语言查询自动转换为SQL并完成执行、统计与可视化报表输出从而降低非技术人员的取数门槛。资源包共1个PDF文件约255KB内容涵盖完整Python代码示例、Docker容器化部署方案、数据库初始化脚本、SQL安全校验与缓存优化策略并展示销售、客户、地域等多场景应用效果。已有244人学习。读者可据此理解Dify工作流节点设计逻辑与DeepSeek模型集成方式掌握从自然语言到图表推荐、Excel与PDF报告导出的全流程实现思路适合希望构建企业级低代码智能数据平台、缩短决策周期的团队参考。1. 从一句人话到一张报表这套 Dify DeepSeek 方案到底能省掉多少手工活业务同事在群里甩来一句“帮我看下上个月华东区退货率最高的五个品类”你打开数据库客户端写 SQL、跑数、导 CSV、再切到 Excel 画图一套下来二十分钟没了。这套基于 Dify 与 DeepSeek 的智能数据分析助手干的就是把这条链路压成一次对话自然语言进SQL 生成、安全校验、执行取数、图表推荐、Excel 导出全自动跑完。它适合有 Python 和数据库基础、想给团队搭一套低代码数据分析入口的工程师也适合数据产品经理拿来验证“对话式 BI”到底能不能落地。核心不是模型多强而是 Dify 工作流把每一步都变成了可编排、可校验、可回滚的节点模型只负责它最擅长的那一段——把话翻译成 SQL。2. 环境搭建与 Dify 数据源接入别让连接字符串成为第一道坎2.1 依赖安装与版本选择这套方案跑起来需要三类东西Dify 客户端、数据库驱动、数据处理与报表库。原文给的依赖清单是准的但实际装的时候有几个细节值得说清楚。# 核心依赖建议在虚拟环境里装 pip install dify-client deepseek-openai sqlalchemy pandas numpy # 报表与可视化 pip install openpyxl reportlab matplotlib seaborn # MySQL 驱动原文没写但必须装否则 create_engine 直接报错 pip install pymysql # PostgreSQL 场景换成 psycopg2-binary逻辑说明dify-client负责调 Dify 的工作流 APIsqlalchemy是数据库引擎的统一入口pandas承接查询结果做统计openpyxl写 Excelmatplotlib和seaborn出图。参数上pymysql是 MySQL 场景的必装项原文的依赖列表里漏了它很多人第一次跑create_engine(mysqlpymysql://...)就卡在这里。Python 版本建议 3.10 以上dify-client在 3.8 上偶发依赖解析冲突。环境变量用export写进 shell 只是临时生效生产环境建议落到.env文件配合python-dotenv读取避免每次开新终端都要重新导出。export DIFY_API_KEYsk-your-dify-analytics-key export DEEPSEEK_API_KEYsk-your-deepseek-key export DB_TYPEmysql export DB_HOSTlocalhost export DB_PORT3306 export DB_NAMEbusiness_data export DB_USERanalytics_user export DB_PASSWORDyour-db-password这里DB_TYPE是个分支开关代码里靠它决定拼 MySQL 还是 PostgreSQL 的连接串。DB_PASSWORD千万别硬编码进脚本这是血泪经验——一旦提交到仓库改密码的代价比省的那点事大得多。2.2 Dify 控制台里的数据源与模型配置Dify 侧要做两件事加数据源、加模型提供商。数据源配置路径是「数据源」→「添加数据源」选 MySQL/PostgreSQL/SQL Server填连接名称、主机、端口、库名、用户名、密码然后点测试连接。这里有个容易翻车的点Dify 容器如果和数据库不在同一网络localhost指的是容器自己得填宿主机真实 IP 或 Docker 网络里的服务名。测试连接失败时先看 Dify 容器能不能 ping 通数据库主机再看数据库的bind-address和用户授权范围。模型提供商配置走「模型提供商」→「添加提供商」→ 选「自定义」填提供商名称 DeepSeek、API 密钥、基础 URLhttps://api.deepseek.com/v1模型列表填deepseek-chat和deepseek-coder。deepseek-chat用于自然语言理解和报告生成deepseek-coder在 SQL 生成这种偏代码的任务上更稳。基础 URL 末尾的/v1不能少少了会 404这是最常见的配置错误之一。提示模型提供商保存时如果报凭据校验失败先确认 API 密钥有没有多余空格再确认基础 URL 是否可达。Dify 的校验请求走的是服务端网络本地能访问不代表 Dify 容器能访问。3. 自然语言转 SQL 工作流五个节点串起从人话到可执行语句3.1 工作流节点设计与连接关系原文的nl_to_sql_workflow.py把流程拆成五个节点schema_extractor抽库表结构、deepseek_sql_generator生成 SQL、sql_validator做安全校验、sql_executor执行查询、result_processor处理结果。连接关系是一条直线前一个节点的输出喂给后一个节点。from dify_client import WorkflowClient import sqlalchemy as sa from sqlalchemy import create_engine, text import pandas as pd import json import os class SmartSQLWorkflow: def __init__(self): self.client WorkflowClient(api_keyos.getenv(DIFY_API_KEY)) self.db_engine self.create_db_engine() def create_db_engine(self): 根据环境变量拼数据库连接串 db_type os.getenv(DB_TYPE, mysql) db_host os.getenv(DB_HOST, localhost) db_port os.getenv(DB_PORT, 3306) db_name os.getenv(DB_NAME, business_data) db_user os.getenv(DB_USER, analytics_user) db_password os.getenv(DB_PASSWORD, ) if db_type mysql: connection_string fmysqlpymysql://{db_user}:{db_password}{db_host}:{db_port}/{db_name} elif db_type postgresql: connection_string fpostgresql://{db_user}:{db_password}{db_host}:{db_port}/{db_name} else: raise ValueError(f不支持的数据库类型: {db_type}) return create_engine(connection_string)逻辑说明create_db_engine是整个系统的地基所有查询都走这个引擎。参数上DB_TYPE决定驱动前缀MySQL 用mysqlpymysqlPostgreSQL 用postgresql。create_engine默认带连接池高并发场景可以加pool_size和max_overflow参数控制。注意密码里如果有或:这类特殊字符直接拼进连接串会解析错乱得用urllib.parse.quote_plus转义。3.2 Schema 提取与 SQL 生成提示词schema_extractor节点用 SQLAlchemy 的inspect拿库表结构把表名、列名、类型、是否可空、主键都抽出来作为上下文喂给模型。这一步决定了模型能不能“看懂”你的库。def get_database_schema(engine): 提取库表结构供模型理解 inspector sa.inspect(engine) schema_info {} for table_name in inspector.get_table_names(): columns [] for column in inspector.get_columns(table_name): columns.append({ name: column[name], type: str(column[type]), nullable: column[nullable] }) schema_info[table_name] { columns: columns, primary_key: [key[name] for key in inspector.get_pk_constraint(table_name).get(constrained_columns, [])] } return schema_info逻辑说明get_table_names列出所有表get_columns拿每张表的列信息get_pk_constraint拿主键。参数上如果库表特别多比如上百张全量塞进提示词会撑爆上下文常见做法是只抽和查询相关的表或者先做一轮表名匹配再抽结构。这也是 Dify 工作流上下文超长的典型场景后面避坑章节会细说。SQL 生成节点的提示词是整套方案的核心资产原文给的版本已经比较完整限定只生成 SELECT、要求带 WHERE、JOIN、GROUP BY、ORDER BY、聚合函数输出 JSON 格式带sql_query、explanation、confidence。temperature设 0.1 是为了让输出稳定SQL 生成这种任务不需要创造力。output_json: True让 Dify 强制模型输出结构化 JSON省去自己解析的麻烦。3.3 SQL 安全校验与执行模型生成的 SQL 不能直接扔给数据库跑sql_validator节点用正则做黑名单拦截。import re def validate_sql(sql_query): 拦截危险 SQL 操作 forbidden_patterns [ r(?i)drop\stable, r(?i)delete\sfrom, r(?i)update\s.\sset, r(?i)insert\sinto, r(?i)alter\stable, r(?i)truncate\stable, r(?i);\s*--, r(?i)union\sselect, r(?i)exec\s*\( ] for pattern in forbidden_patterns: if re.search(pattern, sql_query): return False, 包含危险SQL操作 if not sql_query.strip().lower().startswith(select): return False, 只支持SELECT查询 return True, SQL验证通过逻辑说明黑名单覆盖了 DDL、DML 和注释注入、UNION 注入、存储过程调用。参数上(?i)是忽略大小写\s匹配任意空白。这套正则不是万能的比如SELECT ... INTO OUTFILE这种写文件的操作就没拦住生产环境建议再加一层数据库账号权限限制——给分析账号只读权限比任何正则都可靠。执行节点用engine.connect()开连接connection.execute(text(sql_query))跑查询把结果转成字典列表返回同时带上row_count和columns。异常捕获返回success: False和错误信息方便上游判断。3.4 结果处理与统计result_processor节点把查询结果转成 DataFrame算行数、列数、列名、数据类型数值列再跑一遍describe()拿统计量。def process_query_result(query_result): 把查询结果转成带统计信息的结构 if not query_result.get(success): return query_result df pd.DataFrame(query_result[data]) stats { row_count: len(df), column_count: len(df.columns), column_names: list(df.columns), data_types: {col: str(dtype) for col, dtype in df.dtypes.items()} } numeric_cols df.select_dtypes(include[number]).columns if len(numeric_cols) 0: stats[numeric_stats] df[numeric_cols].describe().to_dict() return { success: True, data: query_result[data], stats: stats, dataframe: df.to_dict(records) }逻辑说明select_dtypes(include[number])筛出数值列describe()给出 count、mean、std、min、max 和分位数。参数上to_dict(records)把 DataFrame 转成字典列表方便后续节点消费。这一步的统计信息会传给报表生成工作流作为图表类型推荐的依据。4. 报表生成工作流从查询结果到可交付的 Excel 和图表4.1 图表类型推荐与可视化生成报表工作流的第一个节点是chart_type_detector用 DeepSeek 根据查询结果的元数据行数、列数、列名、数据类型和用户需求推荐图表类型。提示词里列了柱状图、折线图、饼图、散点图、表格、热力图六种选项输出 JSON 带chart_type、reasoning、chart_title。import matplotlib.pyplot as plt import seaborn as sns import base64 from io import BytesIO import pandas as pd def generate_visualization(df, chart_type, title): 根据图表类型生成图片返回 base64 plt.style.use(default) sns.set_palette(viridis) fig, ax plt.subplots(figsize(10, 6)) if chart_type bar and len(df.columns) 2: x_col, y_col df.columns[0], df.columns[1] df.plot(kindbar, xx_col, yy_col, axax) ax.set_title(title) ax.tick_params(axisx, rotation45) elif chart_type line and len(df.columns) 2: x_col, y_col df.columns[0], df.columns[1] df.plot(kindline, xx_col, yy_col, axax, markero) ax.set_title(title) elif chart_type pie and len(df.columns) 2: labels_col, values_col df.columns[0], df.columns[1] ax.pie(df[values_col], labelsdf[labels_col], autopct%1.1f%%) ax.set_title(title) else: ax.axis(off) table ax.table(cellTextdf.values, colLabelsdf.columns, loccenter, cellLoccenter) table.auto_set_font_size(False) table.set_fontsize(10) table.scale(1, 1.5) ax.set_title(title) plt.tight_layout() buf BytesIO() plt.savefig(buf, formatpng, dpi300, bbox_inchestight) buf.seek(0) img_base64 base64.b64encode(buf.read()).decode(utf-8) plt.close() return fdata:image/png;base64,{img_base64}逻辑说明函数按chart_type分支走不同绘图逻辑最后统一转 base64 返回。参数上dpi300保证导出清晰度bbox_inchestight防止标签被裁。plt.close()必须调否则批量生成图表时内存会持续涨。饼图在类别超过 8 个时可读性急剧下降实际用的时候建议在提示词里加一条“类别超过 8 个时优先推荐柱状图”。4.2 分析报告生成与 Excel 导出report_generator节点用 DeepSeek 生成包含执行摘要、关键发现、详细分析、建议措施、可视化建议五部分的报告。temperature设 0.7比 SQL 生成高因为报告需要一定的表达灵活性。def export_to_excel(df, report_text, statsNone): 把数据、报告、统计信息写进一个 Excel 文件 output BytesIO() with pd.ExcelWriter(output, engineopenpyxl) as writer: df.to_excel(writer, sheet_name数据, indexFalse) report_df pd.DataFrame({分析报告: [report_text]}) report_df.to_excel(writer, sheet_name分析报告, indexFalse) if stats: stats_df pd.DataFrame.from_dict(stats, orientindex) stats_df.to_excel(writer, sheet_name统计信息) output.seek(0) return base64.b64encode(output.read()).decode(utf-8)逻辑说明ExcelWriter配合openpyxl引擎支持多 sheet 写入。参数上indexFalse避免把 DataFrame 索引写进 Excel 多出一列。stats参数做了可选处理没有统计信息时只写数据和报告两个 sheet。返回 base64 是为了在 Dify 工作流里传递二进制内容前端拿到后可以直接触发下载。注意pd.ExcelWriter在 pandas 2.x 里对engine参数的处理有变化如果报openpyxl相关错误先确认 pandas 和 openpyxl 版本兼容常见做法是pip install --upgrade openpyxl。5. 避坑与排查那些让工作流跑不通的细节5.1 模型返回的 SQL 带 Markdown 代码块标记现象执行节点报语法错误打印出来发现 SQL 被sql和包着。原因模型即使被要求输出 JSON有时仍会在sql_query字段里带 Markdown 标记。解决在 SQL 生成节点后加一个清洗步骤用正则去掉代码块标记或者在提示词里明确“sql_query 字段只放纯 SQL 文本不要任何 Markdown 标记”。5.2 Dify 工作流上下文超长导致节点失败现象库表多的时候schema_extractor输出的结构信息把提示词撑爆模型节点报上下文超限。原因全量 schema 塞进提示词token 数超过模型上限。解决只抽查询相关的表或者把 schema 做一层摘要——只保留表名和列名去掉类型和可空信息。常见做法是先让模型根据用户问题选表再抽选中表的结构。5.3 数据库连接在 Dify 容器里不通现象本地脚本能连数据库Dify 工作流里的 Python 节点连不上。原因Dify 跑在容器里localhost指向容器自身。解决填宿主机在 Docker 网络里的 IP或者用host.docker.internalMac/Windows或宿主机网桥 IPLinux。数据库的bind-address要允许来自容器网段的连接。5.4 数值列被当成字符串导致统计为空现象describe()返回空或者图表画出来是乱的。原因数据库里数值列被模型生成的 SQL 用CAST转成了字符串或者 pandas 读进来时类型推断错了。解决在结果处理节点加一步pd.to_numeric强制转换或者在 SQL 生成提示词里要求“数值列保持原始类型不要做不必要的类型转换”。5.5 Excel 导出中文乱码现象导出的 Excel 里中文显示成方块或乱码。原因openpyxl默认字体不支持中文或者写入时编码不对。解决ExcelWriter本身处理 Unicode 没问题乱码通常出在后续读取环节。如果是在前端展示确认响应头Content-Type带charsetutf-8。如果是 PDF 导出reportlab需要注册中文字体否则中文直接丢失。6. 进阶技巧把工作流跑稳的三个习惯第一个习惯是给 SQL 执行加超时和行数上限。分析型查询动辄扫全表一条没加 LIMIT 的 SQL 能把数据库拖垮。常见做法是在执行节点包一层connection.execute(text(sql_query).execution_options(timeout30))同时在 SQL 生成提示词里要求“默认加 LIMIT 1000”。我一般还会在结果处理节点判断row_count超过阈值就截断并标记避免下游图表节点被大数据集卡死。第二个习惯是把高频查询结果缓存起来。同一句“上个月销售额”可能被不同人问很多遍每次都跑一遍模型加数据库不划算。可以在 Dify 工作流前面加一层缓存判断用查询文本的哈希做 key命中就直接返回上次的结果。缓存有效期按业务节奏设日报类查询缓存到当天结束实时性要求高的就不缓存。这一步能把响应时间从十几秒压到一秒内。第三个习惯是给模型输出加置信度阈值。SQL 生成节点返回的confidence字段不是摆设低于 0.7 的查询建议走人工确认再执行而不是直接跑。我一般会在工作流里加一个条件分支confidence 0.7走自动执行否则返回“需要确认”并附上生成的 SQL 让用户看一眼。这个习惯拦住过好几次模型把“退货率”理解成“退货数量”的翻车。验证整套流程是否跑通最直接的办法是拿三个典型查询各跑一遍一个单表聚合“各品类销售总额”、一个多表 JOIN“北京地区销售额前 10 的客户”、一个时间序列“每月销售趋势”。三个都返回正确 SQL、正确行数、正确图表基本可以认为工作流稳了。从那以后我每次改提示词或者换模型版本都强制走一遍这三个用例确认没有回归才上线。希望帮到你。本文还有配套的精品资源点击获取