Qwen3模型MySQL集成实战：对话式数据库查询与报表生成

你有没有遇到过这种情况？业务同事跑过来问：“上个月华东区A产品的销售额是多少？跟去年同期比增长了多少？” 你心里一紧，得先理解他的需求，然后打开数据库客户端，回忆表结构，手写SQL，执行，再把结果整理成他能看懂的表格或图表。一来二去，半小时过去了，同事可能已经等得不耐烦了。

或者，你自己想快速看看某个业务指标的趋势，却卡在了“这个字段在哪个表里？”、“这个关联条件怎么写？”这样的问题上。对于不常写SQL的业务人员来说，直接从数据库获取洞察更是一道难以逾越的鸿沟。

今天，我们就来动手解决这个问题。我将带你一步步构建一个“智能数据库查询助手”。它的核心能力是：你直接用大白话（自然语言）提问，它帮你生成正确的SQL语句，查询MySQL数据库，并把结果清晰、美观地展示给你，甚至能生成简单的图表。整个过程，你不需要懂复杂的SQL语法，就像跟一个懂数据的同事聊天一样简单。

我们将使用强大的Qwen3大语言模型作为这个助手的“大脑”。下面，就来看看如何从零开始，把这个想法变成现实。

1. 场景与痛点：为什么需要对话式查询？

在深入技术细节之前，我们得先搞清楚，这个东西到底能用在哪儿，解决了什么实实在在的麻烦。

想象一下市场部的同事小张。他需要每周做销售报告，每次都要找技术部的你帮忙拉数据。他的需求可能是：“帮我看看最近一周每个销售人员的成单情况，要包含客户名称和金额，按金额从高到低排。” 对你来说，写这个SQL可能不难，但架不住每天都有好几个这样的请求。你的时间被重复性的取数工作占据，而小张也总要等待。

这个智能助手瞄准的，正是“数据需求”与“技术实现”之间的这道沟壑。它具体解决了以下几个痛点：

• 降低数据获取门槛：业务、运营、产品等非技术同学，可以直接用自然语言描述需求，快速获得数据，无需学习SQL或等待技术人员排期。
• 提升数据人员效率：将技术同学从大量简单、重复的取数需求中解放出来，可以更专注于复杂的数据分析、模型构建等更有价值的工作。
• 减少沟通成本与误差：“我想要上周的销量”这种模糊需求，在来回沟通中可能被明确为“我需要2024年5月20日至5月26日，所有SKU的每日销售数量”。助手能通过多轮对话澄清需求，生成精准查询。
• 快速生成数据简报：查询结果不仅能以表格形式呈现，还能进一步被总结、分析，甚至生成直观的“黑板报”式图表，让数据洞察一目了然。

这个助手非常适合用在需要频繁进行数据查询和轻度分析的场景，比如销售报表查看、运营指标监控、用户行为分析、财务数据核对等。

2. 搭建环境：让Qwen3和MySQL“握手”

要让想法落地，首先得准备好战场。我们需要两样核心“武器”：一个能运行Qwen3模型的环境，和一个有数据的MySQL数据库。

2.1 基础环境准备

我假设你已经在本地或服务器上有一个基本的Python开发环境（Python 3.8+）。接下来，我们通过pip安装必要的库。

# 安装大模型交互和数据库连接的核心库
pip install dashscope  # 阿里云灵积平台SDK，用于调用Qwen系列模型
pip install pymysql    # Python连接MySQL数据库的驱动
pip install pandas     # 数据处理和分析，用于格式化查询结果
# 可选：如果你希望结果展示更美观，可以安装tabulate
pip install tabulate

这里我们使用阿里云提供的 dashscope 库来调用Qwen3模型，这是目前比较方便和稳定的方式。当然，如果你已经通过其他方式（如ollama, vLLM等）部署了Qwen3，也可以使用对应的客户端库。

2.2 准备一个示例MySQL数据库

为了演示，我们需要一个包含业务数据的数据库。你可以使用自己已有的数据库，或者快速创建一个示例库。这里我创建一个简单的电商销售数据库。

首先，确保你的MySQL服务已经启动并可以连接。然后，执行下面的SQL语句来创建数据库和表，并插入一些示例数据。

-- 创建数据库
CREATE DATABASE IF NOT EXISTS sales_demo;
USE sales_demo;

-- 创建销售人员表
CREATE TABLE salesperson (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50) NOT NULL,
    region VARCHAR(50)
);

-- 创建产品表
CREATE TABLE product (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL,
    category VARCHAR(50)
);

-- 创建销售订单表
CREATE TABLE sales_order (
    id INT PRIMARY KEY AUTO_INCREMENT,
    order_date DATE NOT NULL,
    salesperson_id INT,
    product_id INT,
    quantity INT,
    amount DECIMAL(10, 2),
    FOREIGN KEY (salesperson_id) REFERENCES salesperson(id),
    FOREIGN KEY (product_id) REFERENCES product(id)
);

-- 插入示例数据
INSERT INTO salesperson (name, region) VALUES
('张三', '华东'),
('李四', '华北'),
('王五', '华南');

INSERT INTO product (name, category) VALUES
('智能手机X', '电子产品'),
('蓝牙耳机', '电子产品'),
('办公椅', '家具'),
('咖啡机', '家电');

INSERT INTO sales_order (order_date, salesperson_id, product_id, quantity, amount) VALUES
('2024-05-20', 1, 1, 5, 25000.00),
('2024-05-21', 2, 2, 10, 3000.00),
('2024-05-21', 1, 3, 2, 1600.00),
('2024-05-22', 3, 4, 3, 2400.00),
('2024-05-23', 2, 1, 3, 15000.00),
('2024-05-24', 1, 2, 5, 1500.00);

现在，我们的数据库里就有了销售人员、产品和订单数据，可以模拟真实的查询场景了。

2.3 配置模型访问密钥

要调用Qwen3模型，你需要一个阿里云灵积平台的API Key。

1. 访问阿里云灵积平台。
2. 注册/登录后，在控制台找到“API-KEY管理”。
3. 创建一个新的API Key并复制保存好。

我们将在一个Python配置文件中使用它，切记不要将密钥硬编码在代码中或上传到公开仓库。一个简单的做法是使用环境变量。

# 在终端中设置环境变量（Linux/Mac）
export DASHSCOPE_API_KEY='你的API-KEY'

# 或者在代码中临时设置（仅用于测试，不推荐生产环境）
import os
os.environ['DASHSCOPE_API_KEY'] = '你的API-KEY'

3. 核心实战：构建智能查询助手

环境搭好了，数据也有了，现在开始编写我们助手的核心逻辑。整个过程可以分解为三个关键步骤：理解问题、生成SQL、执行并返回结果。

3.1 第一步：连接数据库

我们先写一个简单的函数来建立和关闭数据库连接。这里使用 pymysql 库。

import pymysql
from pymysql.cursors import DictCursor

def get_db_connection():
    """创建并返回一个MySQL数据库连接"""
    connection = pymysql.connect(
        host='localhost',      # 你的数据库主机
        user='root',           # 你的数据库用户名
        password='your_password', # 你的数据库密码
        database='sales_demo', # 我们刚才创建的数据库
        charset='utf8mb4',
        cursorclass=DictCursor # 返回字典格式的结果，更方便处理
    )
    return connection

# 测试连接
try:
    conn = get_db_connection()
    print("数据库连接成功！")
    conn.close()
except Exception as e:
    print(f"连接数据库失败: {e}")

3.2 第二步：让Qwen3理解表结构

要让大模型生成正确的SQL，它必须知道数据库里有什么表，每个表有哪些字段，以及表之间的关系。我们需要把这些“知识”告诉它。

我们可以写一个函数来自动获取数据库的元数据（模式信息），并格式化成一段清晰的文本描述，作为给模型的“上下文”。

def get_database_schema(connection):
    """获取当前数据库的表结构描述"""
    schema_info = []
    with connection.cursor() as cursor:
        # 获取所有表名
        cursor.execute("SHOW TABLES")
        tables = cursor.fetchall()
        
        for table in tables:
            table_name = table[f'Tables_in_{connection.db.decode()}']
            schema_info.append(f"表名: {table_name}")
            
            # 获取表的字段信息
            cursor.execute(f"DESCRIBE {table_name}")
            columns = cursor.fetchall()
            
            for col in columns:
                # col 是一个字典，包含 Field, Type, Null, Key, Default, Extra
                field_name = col['Field']
                field_type = col['Type']
                schema_info.append(f"  - {field_name} ({field_type})")
            
            schema_info.append("") # 空行分隔不同表
    
    return "\n".join(schema_info)

# 测试获取模式
conn = get_db_connection()
schema_text = get_database_schema(conn)
print("数据库模式信息：")
print(schema_text)
conn.close()

运行后，你会得到类似下面的文本，这描述了我们的数据库结构：

表名: salesperson
  - id (int)
  - name (varchar(50))
  - region (varchar(50))

表名: product
  - id (int)
  - name (varchar(100))
  - category (varchar(50))

表名: sales_order
  - id (int)
  - order_date (date)
  - salesperson_id (int)
  - product_id (int)
  - quantity (int)
  - amount (decimal(10,2))

3.3 第三步：构建提示词，让模型生成SQL

这是最核心的一步。我们需要设计一个“提示词”（Prompt），将用户的自然语言问题、数据库模式信息以及我们的要求组合起来，引导Qwen3生成我们想要的SQL。

from dashscope import Generation

def generate_sql_from_question(user_question, db_schema):
    """使用Qwen3模型，根据用户问题和数据库模式生成SQL查询语句"""
    
    # 构建系统提示词，设定助手的角色和能力
    system_prompt = """你是一个专业的SQL专家。你的任务是根据用户提供的数据库表结构信息，将用户的自然语言问题转换成准确、高效、安全的MySQL查询语句。
请只输出SQL语句，不要包含任何解释性文字、Markdown代码块标记或额外的说明。如果问题无法通过提供的表结构回答，请输出：`无法生成有效的SQL查询。`
"""
    
    # 构建用户消息，包含模式信息和具体问题
    user_message = f"""数据库表结构如下：
{db_schema}

请根据以上表结构，将下面的问题转换成MySQL查询语句：
问题：{user_question}
"""
    
    try:
        response = Generation.call(
            model='qwen-max', # 可以使用 qwen-plus, qwen-max 等不同规格模型
            system=system_prompt,
            messages=[{'role': 'user', 'content': user_message}],
            result_format='message' # 获取完整的消息格式
        )
        
        if response.status_code == 200:
            # 提取模型返回的纯文本内容，即SQL语句
            sql_query = response.output.choices[0].message['content'].strip()
            # 清理可能出现的markdown代码块标记
            sql_query = sql_query.replace('```sql', '').replace('```', '').strip()
            return sql_query
        else:
            print(f"模型调用失败: {response.code} - {response.message}")
            return None
    except Exception as e:
        print(f"生成SQL时发生错误: {e}")
        return None

# 测试：让模型生成一个简单的SQL
conn = get_db_connection()
schema = get_database_schema(conn)
question = “查询所有销售人员的姓名和所属区域”
generated_sql = generate_sql_from_question(question, schema)
conn.close()

print(f"用户问题：{question}")
print(f"生成的SQL：\n{generated_sql}")

如果一切顺利，模型应该会返回类似 SELECT name, region FROM salesperson; 的SQL语句。你可以尝试问更复杂的问题，比如“计算每个区域的总销售额”，看看模型生成的SQL是否正确（可能需要JOIN和GROUP BY）。

3.4 第四步：执行SQL并格式化结果

生成SQL后，我们需要安全地执行它，并把结果以友好的方式呈现出来。这里我们用 pandas 来美化表格输出。

import pandas as pd

def execute_sql_and_format(connection, sql_query, max_rows=50):
    """执行SQL查询，并将结果格式化为美观的表格字符串"""
    if not sql_query or "无法生成" in sql_query:
        return "未能生成有效的SQL查询语句。"
    
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql_query)
            results = cursor.fetchall()
        
        if results:
            # 将结果转换为pandas DataFrame
            df = pd.DataFrame(results)
            # 限制显示行数，避免输出过长
            formatted_result = df.head(max_rows).to_string(index=False)
            row_count = len(results)
            info_line = f"\n\n查询成功，共返回 {row_count} 行记录。"
            if row_count > max_rows:
                info_line += f" (此处显示前 {max_rows} 行)"
            return formatted_result + info_line
        else:
            return "查询执行成功，但未返回任何数据。"
    except Exception as e:
        return f"执行SQL时出错: {e}\n生成的SQL为: {sql_query}"

# 整合测试：从问题到结果
def ask_database(question):
    """主函数：接收自然语言问题，返回查询结果"""
    conn = get_db_connection()
    try:
        # 1. 获取模式
        schema = get_database_schema(conn)
        # 2. 生成SQL
        sql = generate_sql_from_question(question, schema)
        if not sql:
            return "抱歉，生成SQL时出现问题。"
        print(f"[调试] 生成的SQL: {sql}")
        # 3. 执行并格式化结果
        result = execute_sql_and_format(conn, sql)
        return result
    finally:
        conn.close()

# 来问几个问题试试
questions = [
    “列出所有产品及其类别”，
    “统计每个销售人员的总销售额”，
    “找出2024年5月21日之后的所有订单，显示订单日期、销售员名字和产品名称”
]

for q in questions:
    print(f"\n{'='*50}")
    print(f"问题：{q}")
    print(f"{'='*50}")
    answer = ask_database(q)
    print(answer)

运行这段代码，你应该能看到针对每个自然语言问题，程序自动生成了SQL，执行后并打印出了清晰的表格结果。比如对于“统计每个销售人员的总销售额”，它可能会生成包含销售人员姓名和对应销售总额的表格。

4. 功能增强：从表格到“黑板报”

基本的查询和表格展示已经实现了，但这还不够“智能”和“直观”。我们可以继续增强两个功能：自然语言总结和简单图表生成，让输出更像一份即时生成的“数据黑板报”。

4.1 为结果添加智能总结

查询结果是一堆数字和文本，我们可以让Qwen3再帮我们解读一下，用一两句话点明核心发现。

def generate_summary_for_result(question, sql_result_text):
    """根据原始问题和查询结果，生成一段自然语言总结"""
    
    prompt = f"""你是一个数据分析师。请根据用户最初的问题和对应的数据查询结果，用一两句简洁的话总结核心发现。
不要重复数据，而是给出洞察。例如，指出最大值、最小值、趋势、异常或关键结论。

原始问题：{question}

查询结果（前若干行）：
{sql_result_text[:500]}... [结果可能较长，已截断]

请给出你的总结：
"""
    
    try:
        response = Generation.call(
            model='qwen-max',
            messages=[{'role': 'user', 'content': prompt}],
            result_format='message'
        )
        if response.status_code == 200:
            summary = response.output.choices[0].message['content'].strip()
            return summary
        else:
            return “（总结生成失败）”
    except Exception as e:
        print(f"生成总结时出错: {e}")
        return “（总结生成失败）”

# 修改主函数，加入总结功能
def ask_database_with_summary(question):
    conn = get_db_connection()
    try:
        schema = get_database_schema(conn)
        sql = generate_sql_from_question(question, schema)
        if not sql:
            return “抱歉，生成SQL时出现问题。”, “”
        print(f“[调试] SQL: {sql}”)
        
        result_table = execute_sql_and_format(conn, sql)
        # 生成总结
        summary = generate_summary_for_result(question, result_table)
        return result_table, summary
    finally:
        conn.close()

# 测试增强版
question = “统计每个销售人员的总销售额”
table_result, text_summary = ask_database_with_summary(question)
print(f"\n查询结果表格：\n{table_result}")
print(f"\n【智能总结】\n{text_summary}")

现在，对于销售额查询，你不仅能看到表格，可能还会得到类似“张三的销售额最高，占总销售额的50%以上”这样的总结。

4.2 生成简易图表（控制台版）

在命令行环境下，我们可以生成一些简单的基于文本的图表，比如条形图。虽然简陋，但能直观感受趋势。

def generate_simple_barchart(dataframe, value_col, label_col):
    """生成一个简单的控制台文本条形图"""
    if dataframe.empty:
        return “数据为空，无法生成图表。”
    
    # 取数据
    labels = dataframe[label_col].astype(str).tolist()
    values = dataframe[value_col].tolist()
    
    if not values:
        return “无可用的数值数据。”
    
    # 归一化到最大长度，比如20个字符
    max_val = max(values)
    scale = 20.0 / max_val if max_val > 0 else 1
    
    chart_lines = []
    for label, value in zip(labels, values):
        bar_length = int(value * scale)
        bar = '█' * bar_length
        # 格式化输出，固定宽度便于对齐
        chart_lines.append(f"{label[:15]:<15} | {bar} {value:.2f}")
    
    header = f"{label_col[:15]:<15} | 图表 (数值)"
    separator = "-" * (len(header)+10)
    return header + "\n" + separator + "\n" + "\n".join(chart_lines)

# 修改主函数，对适合图表的数据进行展示
def ask_database_full(question):
    conn = get_db_connection()
    try:
        schema = get_database_schema(conn)
        sql = generate_sql_from_question(question, schema)
        if not sql:
            return “抱歉，生成SQL时出现问题。”, “”, “”
        
        # 执行并获取原始DataFrame
        with conn.cursor() as cursor:
            cursor.execute(sql)
            results = cursor.fetchall()
        df = pd.DataFrame(results)
        
        # 格式化表格输出
        table_output = df.head(20).to_string(index=False) if not df.empty else “无数据”
        row_count = len(df)
        table_output += f"\n\n查询成功，共返回 {row_count} 行记录。"
        
        # 生成总结
        summary = generate_summary_for_result(question, table_output)
        
        # 尝试生成图表（假设查询结果有两列，且第二列是数值）
        chart_output = “”
        if len(df.columns) >= 2 and pd.api.types.is_numeric_dtype(df.iloc[:, 1]):
            chart_output = generate_simple_barchart(df, df.columns[1], df.columns[0])
        
        return table_output, summary, chart_output
    finally:
        conn.close()

# 测试完整功能
question = “统计每个销售人员的总销售额”
table, summary, chart = ask_database_full(question)
print(f"\n📊 查询结果：\n{table}")
print(f"\n💡 数据洞察：\n{summary}")
if chart:
    print(f"\n📈 简易图表：\n{chart}")

运行后，你会在命令行看到一个用“█”字符组成的简单条形图，直观地展示了每个销售人员的销售额对比。

5. 总结与展望

跟着上面的步骤走一遍，一个具备基础能力的对话式数据库查询助手就搭建起来了。它已经能听懂“人话”，帮你访问数据库，并把结果用清晰的表格和总结呈现出来。对于很多日常的、简单的数据探查需求，这已经能节省大量时间。

实际用下来，我感觉最爽的点在于“思维的连续性”没有被中断。以前要在数据库工具、SQL语法、业务问题之间来回切换，现在只需要持续思考“我想知道什么”，然后直接问出来就行。对于业务同学来说，这种体验的提升是巨大的。

当然，这只是一个起点。如果你想把这件事做得更扎实、用到实际工作中，还有一些方向可以考虑：

• 安全性加固：这是重中之重。现在的实现直接执行模型生成的SQL，存在SQL注入风险（虽然模型通常生成SELECT）。生产环境中，必须引入严格的SQL解析和校验，例如使用sqlparse库分析语法，限制只能执行SELECT查询，甚至建立一个允许查询的“白名单”模式。
• 多轮对话与澄清：当前是单次问答。可以增加记忆功能，让助手能基于上下文追问，比如用户说“对比一下”，助手能问“是和上周对比，还是和去年同期对比？”
• 连接更多数据源：除了MySQL，还可以适配PostgreSQL、Snowflake、甚至Excel/CSV文件。
• 更丰富的可视化：将结果接入matplotlib、plotly甚至grafana，生成真正的折线图、饼图，而不是控制台字符图。
• 部署为Web服务：使用Flask或FastAPI将整个功能包装成API，再配上一个简单的前端页面，就成了一个内部的数据查询工具。

构建这个助手的过程，本身也是一个很好的学习案例，它串联了大模型应用、数据库操作和基础的数据处理。建议你先在测试环境里多玩一玩，尝试各种奇怪的问题，看看模型的边界在哪里，然后再思考如何将它融入到你的具体工作流中去。毕竟，工具的价值，最终体现在它解决了多少实际问题上。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。