1. 安装MySQL和PyMySQL

安装MySQL

# 在Ubuntu/Debian上安装
sudo apt update
sudo apt install mysql-server
sudo mysql_secure_installation# 启动MySQL服务
sudo systemctl start mysql
sudo systemctl enable mysql

安装PyMySQL

pip install pymysql

使用 `apt` 安装 MySQL 后，默认情况下 root 用户没有密码，但需要通过 `sudo` 权限访问。

如果希望设置密码（推荐）

使用 `mysql_secure_installation`

运行以下命令交互式设置密码：

sudo mysql_secure_installation

按照提示：

选择密码强度验证策略（通常选 0 跳过）
输入新密码并确认
后续选项建议全部选 Y（移除匿名用户、禁止远程 root 登录等）

用 sudo 登录 MySQL

sudo mysql -u root

检查 MySQL 用户认证方式

SELECT user, host, plugin FROM mysql.user WHERE user='root';

修改 root 用户认证方式为密码

假设你已经用 sudo mysql 进入了 MySQL，执行：

ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY '12345678';
FLUSH PRIVILEGES;

创建数据库和表

import pymysql# 替换为你的MySQL root密码
MYSQL_PASSWORD = 'your_root_password'connection = pymysql.connect(host='localhost',user='root',password='12345678'
)try:with connection.cursor() as cursor:# 创建数据库cursor.execute("CREATE DATABASE IF NOT EXISTS qwen_demo")cursor.execute("USE qwen_demo")# 创建产品表cursor.execute("""CREATE TABLE IF NOT EXISTS products (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(100),category VARCHAR(50),price DECIMAL(10,2),stock INT)""")# 插入示例数据cursor.execute("""INSERT INTO products (name, category, price, stock)VALUES ('笔记本电脑', '电子产品', 5999.00, 50),('智能手机', '电子产品', 3999.00, 100),('平板电脑', '电子产品', 2999.00, 30),('办公椅', '家具', 899.00, 20),('书桌', '家具', 1299.00, 15)""")connection.commit()print("数据库和表创建成功，示例数据已插入！")
finally:connection.close()

2. 部署Qwen3-0.5B模型

pip install transformers torch sentencepiece

text2sql.py

from transformers import AutoModelForCausalLM, AutoTokenizermodel_path = "Qwen/Qwen1.5-0.5B"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto")def generate_sql_from_nl(query):prompt = f"""将以下中文问题转换为SQL查询语句。只返回SQL语句，不要有其他解释或说明。数据库表结构：
表名：products
字段：id, name, category, price, stock问题：{query}
SQL:"""inputs = tokenizer(prompt, return_tensors="pt").to(model.device)outputs = model.generate(**inputs, max_new_tokens=200)sql = tokenizer.decode(outputs[0], skip_special_tokens=True)# 提取SQL部分sql = sql.split("SQL:")[-1].strip()return sql

测试代码：

from transformers import AutoModelForCausalLM, AutoTokenizermodel_path = "/root/.cache/modelscope/hub/models/Qwen/Qwen2.5-1.5B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(model_path, device_map="auto")def generate_sql_from_nl(query):prompt = f"""将以下中文问题转换为SQL查询语句。只返回SQL语句，不要有其他解释或说明。数据库表结构：
表名：products
字段：id, name, category, price, stock问题：{query}
SQL:"""inputs = tokenizer(prompt, return_tensors="pt").to(model.device)outputs = model.generate(**inputs, max_new_tokens=300)sql = tokenizer.decode(outputs[0], skip_special_tokens=True)# 提取SQL部分sql = sql.split("SQL:")[-1].strip()return sqlif __name__ == "__main__":query = "查询所有价格大于100的产品"sql = generate_sql_from_nl(query)print("问题：", query)print("SQL：", sql)

3. 使用Flask部署API

pip install flask flask-cors

创建 app.py:

from flask import Flask, request, jsonify
from flask_cors import CORS
import pymysql
from qwen_model import generate_sql_from_nl  # 假设上面的Qwen代码保存在qwen_model.pyapp = Flask(__name__)
CORS(app)# MySQL配置
db_config = {'host': 'localhost','user': 'root','password': 'your_password','database': 'qwen_demo','charset': 'utf8mb4','cursorclass': pymysql.cursors.DictCursor
}@app.route('/api/query', methods=['POST'])
def handle_query():data = request.jsonuser_query = data.get('query')if not user_query:return jsonify({'error': 'No query provided'}), 400try:# 生成SQLsql = generate_sql_from_nl(user_query)# 执行SQLconnection = pymysql.connect(**db_config)with connection.cursor() as cursor:cursor.execute(sql)result = cursor.fetchall()return jsonify({'sql': sql,'result': result})except Exception as e:return jsonify({'error': str(e)}), 500if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)

启动Flask服务：

python app.py

from flask import Flask, request, jsonify
from flask_cors import CORS
import pymysql
from qwen_model import generate_sql_from_nl  # 假设上面的Qwen代码保存在qwen_model.pyapp = Flask(__name__)
CORS(app)# MySQL配置
db_config = {'host': 'localhost','user': 'root','password': '12345678','database': 'qwen_demo','charset': 'utf8mb4','cursorclass': pymysql.cursors.DictCursor
}@app.route('/api/query', methods=['POST'])
def handle_query():data = request.jsonuser_query = data.get('query')if not user_query:return jsonify({'error': 'No query provided'}), 400try:# 生成SQLsql = generate_sql_from_nl(user_query)# 执行SQLconnection = pymysql.connect(**db_config)with connection.cursor() as cursor:cursor.execute(sql)result = cursor.fetchall()return jsonify({'sql': sql,'result': result})except Exception as e:return jsonify({'error': str(e)}), 500if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)

test_api.py

import requestsurl = "http://127.0.0.1:5000/api/query"
data = {"query": "价格大于3000的产品"  # 这里可以换成你想测试的自然语言问题
}response = requests.post(url, json=data)
print("Status Code:", response.status_code)
print("Response:", response.json())

4. 在Dify中创建工作流

登录Dify平台
创建一个新的工作流
添加以下节点：

节点1: 用户输入

类型：输入节点
配置：接收用户的中文查询

节点2: 调用Flask API

类型：HTTP请求节点
配置：
- URL: http://your-flask-server:5000/api/query
- 方法: POST
- Headers:
  - Content-Type: application/json
- Body:
```
{"query": "{{input.query}}"
}
```

节点3: 结果格式化

类型：JavaScript处理节点
代码：

function formatResult(data) {const result = data.result;if (result.length === 0) return "没有找到匹配的结果";let output = "查询结果：\\n";result.forEach(item => {output += `名称: ${item.name}, 类别: ${item.category}, 价格: ${item.price}, 库存: ${item.stock}\\n`;});return {sql: data.sql,result: output};
}return formatResult(input);