1. 先说清楚：大模型不会“直接操作”数据库

很多人以为把数据库连接信息丢给大模型，它就能自己插数据、改表结构。实际上大模型只是一个文本生成器，它不会真的连接数据库，也不会执行代码。它能做的是：

• 把一句自然语言翻译成一条 SQL 语句；
• 根据返回的查询结果，用自然语言做总结和回答；
• 在工具调用（Function Calling）模式下，告诉我们“该执行哪个函数、传什么参数”。

真正连接 MySQL、执行 SQL、做权限校验的，仍然是你自己写的程序。所以本篇文章的核心是：如何在自己的后台服务里，安全地让大模型生成 SQL，再由你的服务去执行并回传结果。

2. 整体架构

一个可落地的“大模型 + MySQL”系统，通常长这样：

用户提问
   |
   v
你的服务（Python）
   |
   +--> 读取表结构（SHOW CREATE TABLE）
   |
   +--> 组装 Prompt 调用大模型，让它生成 SQL
   |
   +--> 安全校验：只允许 SELECT，禁止增删改、DROP 等
   |
   +--> 用只读账号执行 SQL
   |
   +--> 把查询结果返回给大模型，生成自然语言回答
   |
   v
返回给用户

这套流程通常叫 Text-to-SQL。下面我们就按照这个架构，一步步写出完整可运行的代码。

3. 环境准备与示例数据

3.1 安装依赖

pip install pymysql openai

说明：本示例用 pymysql 连接 MySQL，用 openai 包调用模型。如果你用的是国产模型，比如 DeepSeek、Moonshot、通义千问等，它们大多兼容 OpenAI 接口，只需要改 base_url 和 model 两个参数即可，代码逻辑完全一样。

3.2 准备一个示例库

先建一个电商示例库，方便后续演示。建议在本地 MySQL 中执行下面的初始化脚本。

CREATE DATABASE IF NOT EXISTS shop DEFAULT CHARACTER SET utf8mb4;
USE shop;
CREATE TABLE products (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL COMMENT '商品名称',
category VARCHAR(50) NOT NULL COMMENT '分类',
price DECIMAL(10,2) NOT NULL COMMENT '价格',
stock INT NOT NULL COMMENT '库存'
) COMMENT '商品表';
CREATE TABLE orders (
id INT PRIMARY KEY AUTO_INCREMENT,
product_id INT NOT NULL COMMENT '商品ID',
quantity INT NOT NULL COMMENT '数量',
order_time DATETIME NOT NULL COMMENT '下单时间',
FOREIGN KEY (product_id) REFERENCES products(id)
) COMMENT '订单表';
INSERT INTO products (name, category, price, stock) VALUES
('机械键盘', '数码', 399.00, 120),
('无线鼠标', '数码', 199.00, 200),
('保温杯', '生活', 89.00, 500),
('笔记本支架', '办公', 149.00, 80);
INSERT INTO orders (product_id, quantity, order_time) VALUES
(1, 2, '2026-08-20 10:00:00'),
(2, 1, '2026-08-21 15:30:00'),
(1, 1, '2026-08-22 09:12:00'),
(3, 3, '2026-08-23 18:45:00');

3.3 创建一个只读账号

这是安全的第一道防线：即使模型生成了危险的 SQL，执行它的账号本身也没有权限。

-- 在 MySQL 中创建只读账号
CREATE USER 'readonly_user'@'%' IDENTIFIED BY 'ReadOnly123!';
GRANT SELECT ON shop.* TO 'readonly_user'@'%';
FLUSH PRIVILEGES;

这个账号只能对 shop 库执行 SELECT，无法执行 INSERT、UPDATE、DELETE 或 DROP。等会儿我们的后台服务就用这个账号连接数据库。

4. 第一步：让程序读到表结构

大模型不知道你的数据库里有哪些表、哪些字段。所以第一步，我们要把表结构取出来，拼进 Prompt。

import pymysql
conn = pymysql.connect(
host="127.0.0.1",
port=3306,
user="readonly_user",
password="ReadOnly123!",
database="shop",
charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
)
def get_schema():
"""读取当前库所有表的建表语句，作为上下文喂给大模型。"""
with conn.cursor() as cursor:
cursor.execute("SHOW TABLES")
rows = cursor.fetchall()
    table_names = []
    for row in rows:
        # DictCursor 下，SHOW TABLES 的列名是固定的 key
        table_names.append(list(row.values())[0])
schema_parts = []
for table in table_names:
    cursor.execute(f"SHOW CREATE TABLE `{table}`")
    create_row = cursor.fetchone()
    schema_parts.append(list(create_row.values())[1] + ";")
return "\n\n".join(schema_parts)
if name == "main":
print(get_schema())

运行后，你会得到类似下面的输出，这段文本会作为“数据库说明书”送给模型：

CREATE TABLE `orders` (
  `id` int NOT NULL AUTO_INCREMENT,
  `product_id` int NOT NULL COMMENT '商品ID',
  `quantity` int NOT NULL COMMENT '数量',
  `order_time` datetime NOT NULL COMMENT '下单时间',
  PRIMARY KEY (`id`),
  KEY `product_id` (`product_id`),
  CONSTRAINT `orders_ibfk_1` FOREIGN KEY (`product_id`) REFERENCES `products` (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
CREATE TABLE products (
id int NOT NULL AUTO_INCREMENT,
name varchar(100) NOT NULL COMMENT '商品名称',
category varchar(50) NOT NULL COMMENT '分类',
price decimal(10,2) NOT NULL COMMENT '价格',
stock int NOT NULL COMMENT '库存',
PRIMARY KEY (id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

5. 第二步：基础版 Text-to-SQL

接下来我们写一个最简单的“自然语言转 SQL”函数。核心要点有两个：

• 把表结构放进 Prompt，并明确告诉模型“只输出 SQL，不要解释”；
• 把 temperature 设为 0，让输出尽量稳定，避免生成胡编的字段名。

from openai import OpenAI
client = OpenAI(
api_key="sk-你的密钥",
base_url="https://api.deepseek.com",  # 示例：换成你实际使用的服务地址
)
SYSTEM_PROMPT = """你是一名资深 MySQL 工程师。
请根据给定的表结构，把用户的问题翻译成一条可执行的 SELECT 语句。
要求：
只输出 SQL 语句本身，不要任何解释、注释或代码块标记；
只能使用表结构中存在的字段；
如果问题无法回答，输出一行：无法回答。"""
def text_to_sql(question: str, schema: str) -> str:
user_prompt = f"""数据库表结构如下：
{schema}
用户问题：
{question}
"""
resp = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_prompt},
],
temperature=0,
)
return resp.choices[0].message.content.strip()
if name == "main":
schema = get_schema()
sql = text_to_sql("每种分类各有多少件商品？", schema)
print(sql)

模型很可能返回类似这样的 SQL：

SELECT category, SUM(stock) AS total_stock
FROM products
GROUP BY category;

6. 第三步：安全校验 + 只读执行

千万不能把模型生成的 SQL 直接拿去执行。因为大模型可能“想太多”，比如用户只是问“库存情况”，模型却生成了一条 DELETE 或 DROP TABLE。所以执行前必须做白名单校验。

def is_safe_sql(sql: str) -> bool:
    """只允许执行单条 SELECT 查询。"""
    if not sql or not sql.strip():
        return False
upper_sql = sql.strip().upper().rstrip(";")
必须从 SELECT 开始
if not upper_sql.startswith("SELECT"):
return False
禁止危险关键字
forbidden_keywords = [
"INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE",
"CREATE", "GRANT", "REVOKE", "LOAD_FILE", "INTO OUTFILE",
"INTO DUMPFILE", "SLEEP", "BENCHMARK",
]
for keyword in forbidden_keywords:
if keyword in upper_sql:
return False
禁止一次执行多条语句
if ";" in sql.rstrip(";"):
return False
return True
def query_by_question(question: str):
schema = get_schema()
sql = text_to_sql(question, schema)
if not is_safe_sql(sql):
return {"ok": False, "sql": sql, "data": [], "message": "生成的 SQL 被安全策略拦截"}
try:
with conn.cursor() as cursor:
cursor.execute(sql)
rows = cursor.fetchall()
return {"ok": True, "sql": sql, "data": rows, "message": "查询成功"}
except Exception as exc:
return {"ok": False, "sql": sql, "data": [], "message": f"执行失败：{exc}"}
if name == "main":
result = query_by_question("库存最少的三个商品是哪些？")
print(result["sql"])
for row in result["data"]:
print(row)

这里有三重保护：

1. 账号只读：即使校验被绕过，数据库层面也拒绝增删改；
2. 关键字白名单：只放行 SELECT，拦截危险操作；
3. 单语句限制：防止模型用分号拼接第二条恶意 SQL。

7. 进阶：用 Function Calling 让模型自主查询

上面的方案已经能跑通，但有一个不足：模型只生成一条 SQL，无法处理“先查 A、再根据结果查 B”的多轮查询。更强的做法是使用工具调用（Function Calling）。我们把“执行 MySQL 查询”注册成一个工具，让模型自己决定什么时候调用、传什么 SQL。

import json
TOOLS = [
{
"type": "function",
"function": {
"name": "run_mysql_query",
"description": "在只读 MySQL 数据库 shop 上执行 SELECT 查询",
"parameters": {
"type": "object",
"properties": {
"sql": {
"type": "string",
"description": "要执行的 SELECT 语句，只能是单条查询",
}
},
"required": ["sql"],
},
},
}
]
def run_mysql_query(sql: str) -> str:
"""工具实现：执行 SQL 并返回序列化结果。"""
if not is_safe_sql(sql):
return "SQL 被安全策略拦截，请改写为合法的 SELECT 查询"
try:
    with conn.cursor() as cursor:
        cursor.execute(sql)
        rows = cursor.fetchall()
    # 限制返回行数，避免结果过长
    return json.dumps(rows[:100], ensure_ascii=False, default=str)
except Exception as exc:
    return f"查询执行失败：{exc}"
def chat_with_db(question: str) -> str:
schema = get_schema()
messages = [
{
"role": "system",
"content": (
"你是数据分析助手。下面是数据库表结构，回答问题时请先调用 "
"run_mysql_query 工具查询数据，再根据结果用中文回答。\n\n"
f"表结构：\n{schema}"
),
},
{"role": "user", "content": question},
]
# 第一轮：模型决定是否调用工具
resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=TOOLS,
    temperature=0,
)
message = resp.choices[0].message
如果模型调用了工具，执行并把结果追加回消息
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
sql = arguments.get("sql", "")
    result = run_mysql_query(sql)
messages.append(message)  # 模型发出的工具调用消息
messages.append(
    {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,
    }
)
第二轮：模型根据工具结果生成最终回答
resp2 = client.chat.completions.create(
model="deepseek-chat",
messages=messages,
temperature=0,
)
return resp2.choices[0].message.content
模型没有调用工具，直接返回
return message.content
if name == "main":
print(chat_with_db("帮我算一下总销售额是多少？"))

这个函数的工作流程是：

1. 先把表结构放进系统提示词；
2. 用户提问后，模型判断需要查数据，于是调用 run_mysql_query，参数里带上它自己生成的 SQL；
3. 你的程序执行 SQL，把结果作为 tool 消息回传；
4. 模型拿到真实数据后，重新组织语言，输出最终答案。

这种方式的好处是：模型可以连续调用多次工具，比如先查出销量最高的商品 ID，再查这个商品的库存，非常适合复杂问题。

8. 常见坑与最佳实践

8.1 给模型的信息要具体

如果表很多，不要把全库几百张表的结构都塞进 Prompt，既费 Token 又容易让模型选错表。可以先做一层表路由：让模型先选出相关表，再只把这些表的结构发过去。

8.2 给字段加注释

从第 3 节的建表语句可以看到，我给每个字段都写了 COMMENT。这些注释会跟着 SHOW CREATE TABLE 一起返回，能显著提升模型理解字段含义的准确率。字段名用 price 好过 p1，有注释好过没注释。

8.3 限制返回行数

大模型上下文长度有限，查询结果别一次返回几万行。像第 7 节那样，执行后只取前 100 行，并在工具描述里说明“请使用 LIMIT 限制返回量”。

8.4 永远不要用 root 账号

无论对校验多有信心，都请使用只读账号，并且让服务端对执行的 SQL 做白名单校验和超时控制。还要给执行语句加上超时，比如：

with conn.cursor() as cursor:
    cursor.execute("SET SESSION MAX_EXECUTION_TIME = 3000")  # 3 秒超时
    cursor.execute(sql)
    rows = cursor.fetchall()

8.5 记得处理“模型答不上来”的情况

在系统提示词里明确告诉模型：如果问题与数据库无关，或者无法用 SQL 表达，就返回固定的话术，比如“无法回答”。这样你的程序可以据此做降级处理，而不是把错误 SQL 硬塞进数据库。

9. 总结

让大模型操作 MySQL，本质上是一套「生成 SQL → 校验 → 执行 → 回传结果 → 生成回答」的流水线。把数据库连接器、权限控制和查询工具准备好，模型才能真正发挥“口语提问、拿数据说话”的能力。

• 简单场景：用 text_to_sql 生成一条查询即可；
• 安全要求高的场景：加上只读账号、关键字白名单、超时与行数限制；
• 复杂多轮场景：注册 run_mysql_query 工具，用 Function Calling 让模型自主决策。

你可以先把第 5 节的单条查询跑通，再逐步加上第 6 节的安全校验和第 7 节的工具调用，最终就能得到一个稳定、可落地的“大模型 MySQL 助手”。