Windows 10企业版中实现MySQL数据库与LangChain框架的高效数据交互

引言

在AI应用开发中，将大语言模型(LLM)与数据库连接是实现智能问答、数据分析等功能的常见需求。本文将指导你在Windows 10企业版环境下，使用Python搭建MySQL数据库与LangChain框架之间的高效数据交互通道。

准备工作

环境要求

Windows 10企业版
Python 3.8或更高版本
MySQL Server 8.0+
Git Bash或PowerShell（推荐）

需要安装的软件包

代码片段

pip install langchain mysql-connector-python python-dotenv openai

第一部分：MySQL数据库配置

1. 安装MySQL Server

从MySQL官网下载Windows版安装包：
https://dev.mysql.com/downloads/installer/

安装时选择”Developer Default”配置，设置root密码并记住它。

2. 创建测试数据库

使用MySQL命令行客户端执行：

代码片段

CREATE DATABASE langchain_demo;
USE langchain_demo;

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    description TEXT,
    price DECIMAL(10,2),
    stock INT DEFAULT 0
);

INSERT INTO products (name, description, price, stock) VALUES 
('Laptop', 'High performance gaming laptop', 1299.99, 15),
('Smartphone', 'Latest model with AI camera', 899.99, 30),
('Headphones', 'Noise cancelling wireless', 199.99, 50);

第二部分：配置Python环境

1. 创建项目目录结构

代码片段

mkdir langchain-mysql-demo
cd langchain-mysql-demo

2. 设置环境变量

创建.env文件保存敏感信息：

代码片段

# .env文件内容
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password_here
MYSQL_DATABASE=langchain_demo
OPENAI_API_KEY=your_openai_key_here

注意：
– .env文件应加入.gitignore
– MySQL密码不要使用特殊字符避免连接问题

第三部分：实现数据交互

1. MySQL连接工具类

创建db_utils.py:

代码片段

import mysql.connector
from dotenv import load_dotenv
import os

load_dotenv()

class MySQLConnector:
    def __init__(self):
        self.host = os.getenv("MYSQL_HOST")
        self.user = os.getenv("MYSQL_USER")
        self.password = os.getenv("MYSQL_PASSWORD")
        self.database = os.getenv("MYSQL_DATABASE")
        self.port = os.getenv("MYSQL_PORT", "3306")

    def get_connection(self):
        return mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database,
            port=self.port
        )

    def execute_query(self, query, params=None):
        conn = self.get_connection()
        cursor = conn.cursor(dictionary=True)

        try:
            cursor.execute(query, params or ())

            if query.strip().lower().startswith('select'):
                result = cursor.fetchall()
            else:
                conn.commit()
                result = cursor.rowcount

            return result

        except Exception as e:
            conn.rollback()
            raise e

        finally:
            cursor.close()
            conn.close()

# 测试连接是否正常
if __name__ == "__main__":
    db = MySQLConnector()
    results = db.execute_query("SELECT * FROM products LIMIT 1")
    print("测试结果:", results)

2. LangChain集成实现

创建langchain_mysql.py:

代码片段

from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage
from db_utils import MySQLConnector
from dotenv import load_dotenv

load_dotenv()

class LangChainMySQLAgent:
    def __init__(self):
        self.db = MySQLConnector()
        self.llm = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0,
            openai_api_key=os.getenv("OPENAI_API_KEY")
        )

    def get_table_schema(self):
        """获取表结构作为上下文"""
        query = """
        SELECT COLUMN_NAME, DATA_TYPE 
        FROM INFORMATION_SCHEMA.COLUMNS 
        WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'products'
        """
        schema = self.db.execute_query(query, (os.getenv("MYSQL_DATABASE"),))

        schema_str = "\n".join([
            f"{col['COLUMN_NAME']} ({col['DATA_TYPE']})" 
            for col in schema
        ])

        return f"表结构:\n{schema_str}"

    def query_data(self, natural_language_query):
        """将自然语言转换为SQL查询"""

        # Step1: LLM生成SQL查询语句（带安全检查）
        system_prompt = f"""
        你是一个专业的SQL开发者。基于以下表结构：

        {self.get_table_schema()}

        请将用户的问题转换为安全的MySQL查询语句。
        注意：
        1. ONLY生成SELECT语句，禁止修改数据 
        2. NEVER包含敏感操作如DROP/DELETE/UPDATE等

        返回格式：
        理由：<解释为什么这样设计查询>
        查询：<SQL查询语句>
        参数：<参数列表>

        示例输入："最贵的3个产品是什么？"
        示例输出：
        理由：需要按价格降序排序并限制结果数量...
        查询：SELECT name, price FROM products ORDER BY price DESC LIMIT %s...
        参数：[3]

        当前问题：{natural_language_query}
        """

        messages = [
            SystemMessage(content=system_prompt),
            HumanMessage(content=natural_language_query)
        ]

        response = self.llm(messages).content

        # Step2:解析LLM响应提取SQL和参数（简单实现）
        try:
            sql_part = response.split("查询：")[1].split("\n")[0].strip()

            if "参数：" in response:
                params_part = response.split("参数：")[1].split("\n")[0].strip()
                params = eval(params_part) if params_part else []
            else:
                params = []

            # Step3:执行安全校验（生产环境需要更严格的检查）
            if not sql_part.lower().startswith('select'):
                raise ValueError("只允许执行SELECT查询")

            # Step4:执行查询并返回结果    
            results = self.db.execute_query(sql_part, params)

            return {
                "sql": sql_part,
                "params": params,
                "results": results,
                "analysis": response.split("理由：")[1].split("\n")[0] if "理由：" in response else ""
            }

        except Exception as e:
            return {"error": str(e), "raw_response": response}

if __name__ == "__main__":
    agent = LangChainMySQLAgent()

    while True:
        question = input("\n请输入关于产品的问题(输入q退出): ")

        if question.lower() == 'q':
            break

        result = agent.query_data(question)

        if "error" in result:
            print(f"错误: {result['error']}")

        print("\n==== SQL分析 ====")
        print(f"分析: {result['analysis']}")

        print("\n==== SQL执行 ====")


if __name__ == "__main__":
    agent = LangChainMySQLAgent()

    while True:
question=input("\n请输入关于产品的问题(输入q退出): ")
if question.lower() == 'q':
break

result=agent.query_data(question)
if"error"in result:
print(f"错误:{result['error']}")
print(f"原始响应:{result['raw_response']}")
continue

print("\n==== SQL分析 ====")
print(f"分析:{result['analysis']}")
print(f"生成SQL:{result['sql']}")
print(f"使用参数:{result['params']}")

print("\n====查询结果 ====")
for row in result["results"]:
print(row)

第四部分：高级优化技巧

1.SQL注入防护增强

在生产环境中，建议添加额外的安全检查层：

代码片段



def is_safe_sql(sql):
"""检查SQL是否安全"""
sql=sql.lower().strip()

#不允许的操作类型

forbidden=[
'insert','update','delete','drop',
'truncate','alter','create','grant'
]

for cmd in forbidden:
if cmd in sql and not sql.startswith(cmd+' '):
return False

return True


#在query_data方法中添加检查


if not is_safe_sql(sql_part):
raise ValueError("检测到潜在危险操作")

2.ChatGPT提示工程优化

可以改进系统提示以获得更准确的SQL生成：

代码片段



system_prompt="""
你是一个专业的{db_type}开发者。基于以下表结构：

{schema_info}

规则：
1.STRICTLY遵循这些列名和数据类型

2.NEVER假设不存在的列

3.PREFER使用参数化查询(%s)而非字符串拼接

4.AVOID子查询除非必要

5.EXPLAIN你的设计思路

当前数据库类型:{db_type}
当前问题:{question}
"""

第五部分：常见问题解决

Q1:连接MySQL时出现认证错误

A:检查以下几点：
-确认.env中的密码正确

-尝试用MySQL Workbench测试相同凭证

-如果是MySQL8+，可能需要更改认证方式：

代码片段



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

Q2:LangChain生成的SQL不正确

A:调试步骤：
1.打印完整的LLM响应(response.content)查看原始输出

2.添加更详细的表结构信息（包括示例数据）

3.降低temperature值获得更确定性的结果

Q3:Windows防火墙阻止连接

A:在管理员PowerShell中运行：

代码片段



New-NetFirewallRule-DisplayName"Allow MySQL"-Direction Inbound-LocalPort3306-ProtocolTCP-Action Allow

总结

本文实现了Windows10环境下MySQL与LangChain的集成方案，关键点包括：

1.MySQL的正确安装和配置

2.LangChain与OpenAI API的对接

3.SQL生成的安全防护机制

4.Windows环境的特殊注意事项

完整项目代码可从GitHub获取：[假想的示例仓库链接]

下一步改进方向：
-添加更多表的复杂关系处理

-实现数据修改操作的审批流程

-集成可视化界面展示查询结果