LLM大模型如何访问MySQL业务数据库

目录

一、LLM大模型如何访问MySQL业务数据库

1.1 为什么需要SQL Agent?

1.2 什么是 create_sql_agent？

1.3 什么是SQLDatabaseToolkit?

二、SQL Agent智能体操作MySQL数据库

三、本地启动服务 验证效果

四、怎么提高SQL Agent智能体的回复准确性？

文章首先讲解了create_sql_agent和SQLDatabaseToolkit的核心作用，展示了如何通过LangChain框架连接数据库并创建具备SQL解析能力的智能代理。随后结合实际应用场景，构建了一个基于FastAPI的文件查询系统，演示了如何将自然语言问题转化为结构化数据响应，并确保数据安全与准确性。最后，探讨了提升SQL Agent回复准确性的关键因素，如提示词工程、表结构描述优化、示例引导（Few-Shot Learning）等方法

一、LLM大模型如何访问MySQL业务数据库

1.1 为什么需要SQL Agent?

用户更倾向于用自然语言提问（如“销售额最高的产品是什么？”），而非编写复杂 SQL。

非技术人员（如产品经理、业务人员）无需学习 SQL 即可查询数据库。

包括开发对应的业务智能体，实现数据库的查询和操作

1.2 什么是 create_sql_agent？

创建能通过自然语言与SQL数据库交互的AI智能体，自动生成/执行SQL查询并解析结果

核心能力:

将用户问题（如“统计每个地区的销量”）转化为 SQL 查询语句。

连接数据库执行 SQL，默认只读模式防止数据误修改。

将数据库返回的原始数据（如 [1500, 2000]）转换为用户友好的回答（如“总销售额为 $3500”）。

自动修正 SQL 语法错误或逻辑问题（如字段名拼写错误）。

1.3 什么是SQLDatabaseToolkit?

LangChain 中专门用于 连接 SQL 数据库并集成相关操作工具 的模块包

#使用 SQLDatabase.from_uri 连接数据库，自动读取表结构。
db = SQLDatabase.from_uri(
  database_uri="数据库连接信息",
  include_tables=['a***ount_file', 'storage'],
  custom_table_info={"a***ount": "查询账号相关的表"}) 

#SQLDatabaseToolkit 封装了查询执行、表结构查看等底层操作。
toolkit = SQLDatabaseToolkit(db=db, llm=ChatOpenAI())

agent = create_sql_agent(
    llm=ChatOpenAI(temperature=0, model="gpt-5"),  # 必需：大模型
    toolkit=toolkit,    # 必需：数据库工具包
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,  # Agent类型
    verbose=True,          # 显示详细执行过程
    prefix="""你是一个专业的MySQL专家...""",  # 自定义提示前缀
    suffix="""请始终检查你的查询结果..."""     # 自定义提示后缀
)

快速的看一个demo 这会更好理解流程:

import os
from langchain_***munity.agent_toolkits.sql.base import create_sql_agent
from langchain_***munity.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain_***munity.utilities import SQLDatabase
from langchain_openai import ChatOpenAI
from langchain.agents.agent_types import AgentType

# 设置OpenAI API密钥
llm = ChatOpenAI(
    model_name="qwen-plus",
    base_url="https://dashscope.aliyuncs.***/***patible-mode/v1",
    api_key="sk-xxxxxxxxxxxx",
    temperature=0,
)

# 1. 连接数据库
db = SQLDatabase.from_uri(
    f"mysql+pymysql://root:xxxxxxx@93.179.111.1111:3306/mysql",
    include_tables=["a***ount_file"],
    custom_table_info={"a***ount_file": "查询文件夹和文件内容相关的表"},  # 自定义表描述
)

# 初始化工具包
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
print(f"包含的工具个数:{len(toolkit.get_tools())}")
print("工具列表", [tool.name for tool in toolkit.get_tools()])

# 创建SQLAgent
agent = create_sql_agent(
    llm=llm,
    toolkit=toolkit,
    verbose=True,
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    handle_parsing_errors=True,
    max_iterations=10,
    prefix="You are a helpful assistant that answers questions about the database.",
    suffix="",
)


def ask_question(question: str) -> str:
    print(f"问题:{question}")
    result = agent.invoke({"input": question})
    return result["output"]


questions = ["有多少个文件"]

for question in questions:
    answer = ask_question(question)
    print(f"答案:{answer}")
    print("-" * 50)

用户请求->创建代理->执行查询->返回结果

二、SQL Agent智能体操作MySQL数据库

需求:通过自然语言查看文件列表、查看文件详细信息、查看文件夹内容

流程说明:

用户请求: 用户发送自然语言查询、获取用户ID

参数处理：处理查询参数、注入用户ID

创建代理：初始化SQL代理、准备查询环境

执行查询：执行数据库查询、获取查询结果

返回结果：返回JSON格式响应

 先思考几个问题:

问题一：大模型操作Mysql数据库如何做到个人数据隔离，不会查询到别人的数据？

问题二：大模型响应给调用方的内容格式如何限制？

问题三：大模型如何更精确的执行用户的查询需求？

解决方案:【！！！！提示词工程！！！】 

 定义返回结果的实体pan_schemas.py文件 

from pydantic import BaseModel
from typing import Optional, Dict, Any, List
from datetime import datetime

class PanQueryRequest(BaseModel):
    """文件查询请求模型"""
    a***ount_id: Optional[int] = None  # 从token中获取，请求中可选
    query: str

class FileInfo(BaseModel):
    """文件信息模型"""
    id: int  # a***ount_file表的ID
    file_id: int  # 实际存储的文件ID
    file_name: str
    file_type: str
    file_suffix: str
    file_size: int
    gmt_create: datetime
    gmt_modified: datetime

class PanQueryResponse(BaseModel):
    """文件查询响应模型"""
    type: str
    data: Dict[str, Any] 

创建路由文件file.py

from fastapi import APIRouter, Depends
from models.pan_schemas import PanQueryRequest
from agents.pan_agent import process_pan_query
from core.auth import get_current_user
import logging


logger = logging.getLogger(__name__)
# 创建路由
router = APIRouter(prefix="/api/pan", tags=["文件查询"])


@router.post("/query")
async def pan_query(request: PanQueryRequest):

    request.a***ount_id = 1
    logger.info(f"用户{request.a***ount_id}开始查询文件")
    # 调用智能体进行查询
    return await process_pan_query(request)

file_agent.py

process_pan_query 这个异步方法主要是创建agent ->构建输入->执行获取输出->解析结果返回

async def process_pan_query(request: PanQueryRequest) -> JsonData:
    agent = create_pan_agent()

    # 构建查询输入，可以做更多事情，包括检查过滤用户的输入
    query_input = f"用户ID为{request.a***ount_id}的{request.query}"

    # 获取代理的输出
    response = await agent.ainvoke({"input": query_input})

    if "output" not in response:
        return JsonData.error("查询失败,请换种方式再重试下")

    output = response["output"]

    # 解析数据
    try:
        data = json.loads(output) if isinstance(output, str) else output
        logger.info(f"解析数据成功: {data}")
        return JsonData.su***ess({"type": data.get("type"), "data": data.get("data")})

    except json.JSONDecodeError as e:
        logger.error(f"解析数据失败: {str(e)}")
        return JsonData.su***ess(data={"content": str(output)})

def create_pan_agent() -> Any:
    """创建网盘查询agent"""
    # 创建数据库连接，只读模式
    db = SQLDatabase.from_uri(
        f"mysql+pymysql://{settings.MYSQL_USER}:{settings.MYSQL_PASSWORD}@{settings.MYSQL_HOST}:{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}",
        include_tables=["a***ount_file"],
    )

    # 创建大模型
    llm = get_default_llm()

    # 创建数据库工具包
    tookit = SQLDatabaseToolkit(db=db, llm=llm)

    # 创建提示词
    # 创建提示模板
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                """你是一个智能网盘助手，专门用于查询用户的网盘文件信息。你只能执行查询操作，不能执行任何修改数据的操作。

        重要警告：
        1. 你绝对不能生成或编造任何数据
        2. 你只能返回实际查询到的数据
        3. 如果查询没有结果，必须返回空结果
        4. 任何生成或编造数据的行为都是严重错误
        5. 你只能使用数据库中的实际数据
        6. 不能对查询结果进行任何修改或补充
        7. 不能生成示例数据或占位数据
        8. 不能假设或推测数据
        9. 不能使用模板或示例数据
        10. 不能对数据进行任何形式的加工或美化

        数据库表结构说明：
        - a***ount_file: 用户文件表
          - id: 文件ID（a***ount_file表的主键）
          - a***ount_id: 用户ID
          - is_dir: 是否为文件夹（0不是，1是）
          - parent_id: 上层文件夹ID（顶层为0）
          - file_id: 实际存储的文件ID
          - file_name: 文件名称
          - file_type: 文件类型（***mon/***press/excel/word/pdf/txt/img/audio/video/ppt/code/csv）
          - file_suffix: 文件后缀名
          - file_size: 文件大小（字节）
          - del: 是否删除（0未删除，1已删除）
          - del_time: 删除时间
          - gmt_create: 创建时间
          - gmt_modified: 更新时间
        
        你可以处理以下类型的查询请求：
        1. 文件查询
           - 查看我的文件列表
           - 搜索特定文件
           - 查看文件详细信息
           - 查看文件夹内容
           - 查看最近修改的文件
              

        重要限制：
        1. 你只能执行 SELECT 查询，不能执行任何修改数据的操作
        2. 所有查询必须包含 a***ount_id 条件，确保数据安全
        3. 不能执行以下操作：
           - 删除文件
           - 修改文件
           - 创建文件
           - 移动文件
           - 重命名文件
           - 清空回收站
           - 修改存储空间
        4. 如果用户请求执行任何修改操作，请礼貌地拒绝并说明原因
        5. 如果查询没有结果，必须返回空结果，不能生成示例数据
        6. 绝对不能生成或编造任何数据
        7. 只能返回实际查询到的数据
        8. 不能对数据进行任何形式的加工或美化

        处理请求时请注意：
        1. 必须使用 a***ount_id 过滤用户数据，确保数据安全
        2. 对于文件夹查询，使用 is_dir=1 和 parent_id
        3. 对于文件类型查询，使用 file_type 字段
        4. 对于模糊查询，使用 LIKE 和通配符
        5. 对于时间相关的查询，使用 gmt_create 和 gmt_modified
        7. 结果要简洁明了，突出重点
        8. 所有查询必须包含 a***ount_id 条件
        9. 查询文件信息时，必须返回 a***ount_file 表的 id 和 file_id
        10. 所有响应必须返回 JSON 格式的数据，包含完整的文件信息
        11. 如果查询没有结果，返回空数组或空对象，不要生成示例数据
        12. 绝对不能生成或编造任何数据

        响应格式必须符合以下模型结构：
        1. 文件列表响应：
           {{
               "type": "file_list",
               "data": List[FileInfo]  # FileInfo包含id, file_id, file_name, file_type, file_suffix, file_size, gmt_create, gmt_modified
           }}
            
        请根据用户的问题，使用 SQL 查询来获取信息，并返回符合上述格式的 JSON 数据。
        重要警告：你绝对不能生成或编造任何数据，只能返回实际查询到的数据。任何生成或编造数据的行为都是严重错误。""",
            ),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
  # 创建SQLAgent
    agent = create_sql_agent(
        llm=llm,
        toolkit=tookit,
        agent_type="openai-tools",
        verbose=True,
        return_intermediate_steps=True,
        max_iterations=15,
        handle_parsing_errors=True,
        prompt=prompt,
    )

    return agent

def get_default_llm():
    return ChatOpenAI(
       model = settings.LLM_MODEL_NAME,
       base_url = settings.LLM_BASE_URL,
       api_key = settings.LLM_API_KEY,
       temperature = settings.LLM_TEMPERATURE,
       streaming = settings.LLM_STREAMING
    )

最主要的就是上面的这些提示词,用来给大模型做各种限制,给大模型提供样例

三、本地启动服务 验证效果

这些是数据库中存在的数据:

使用ApiFox 请求api 

{
    "code": 0,
    "data": {
        "type": "file_list",
        "data": [
            {
                "id": 2,
                "file_id": 101,
                "file_name": "Resume.pdf",
                "file_type": "pdf"
            }
        ]
    },
    "msg": "",
    "type": "text"
}

成功返回~ 

四、怎么提高SQL Agent智能体的回复准确性？

 SQL Agent智能体的回复准确性，和大模型参数、温度、能力、提示词工程强相关，也包括用户的提问内容

比如更详细的表结构Schema说明，提问改写，提供FewShot样例等，都是可以提高准确度

增强 Schema 理解

 # 在提示词中添加增强元数据描述
prefix = """
你连接的数据库包含以下关键表：
[Customers] 客户表（重要字段：CustomerId, FirstName, LastName, Country）
[Invoices] 发票表（与 Customers 通过 CustomerId 关联）
优先使用 JOIN 代替子查询，注意 Country 字段存储的是国家全称
"""

Few-Shot Learning 通过示例引导生成模式

 examples = [
    {
        "input": "法国客户数量是多少？",
        "query": "SELECT COUNT(*) FROM Customers WHERE Country = 'France'"
    },
    {
        "input": "显示最新的5张发票", 
        "query": "SELECT * FROM Invoices ORDER BY InvoiceDate DESC LIMIT 5"
    }
] 

总结:

本文围绕如何构建一个能够通过自然语言与MySQL数据库交互的SQL Agent智能体展开，涵盖了从环境搭建到本地验证的完整流程。首先介绍了使用create_sql_agent和SQLDatabaseToolkit来实现自然语言转SQL的核心机制，并给出了代码示例，帮助开发者快速上手。接着，通过构建一个网盘文件查询服务，展示了如何将智能体集成到实际业务场景中，包括定义数据模型、设计API接口、编写异步处理逻辑等内容。特别强调了提示词工程在控制模型行为、保证数据真实性和避免编造信息方面的重要性。最后，文章分析了影响SQL Agent性能的关键因素，并提出了多种优化策略，如增强Schema理解、提供示例引导等