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?用戶認證方式
登錄?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 sentencepiecetext2sql.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);
節點4: 輸出結果
-
類型：輸出節點
-
配置：顯示格式化后的結果
