2024年最新：在CentOS 7上搭建MySQL 8.0+LangChain的AI开发环境

引言

随着AI技术的快速发展，将传统数据库与AI框架结合已成为开发者的新需求。本文将手把手教你如何在CentOS 7系统上搭建MySQL 8.0数据库，并集成LangChain这一强大的AI开发框架，构建一个完整的AI开发环境。

准备工作

在开始之前，请确保：
– 已安装CentOS 7系统（建议最小化安装）
– 拥有sudo权限的账户
– 至少2GB内存（LangChain运行需要一定资源）
– 稳定的网络连接（需要下载软件包）

第一部分：安装MySQL 8.0

1.1 添加MySQL官方Yum仓库

代码片段

# 下载MySQL Yum仓库安装包
sudo yum install -y https://dev.mysql.com/get/mysql80-community-release-el7-7.noarch.rpm

# 验证仓库是否添加成功
sudo yum repolist enabled | grep "mysql.*-community.*"

原理说明：我们使用MySQL官方提供的Yum仓库来确保获取最新稳定版的MySQL 8.0。

1.2 安装MySQL服务器

代码片段

# 安装MySQL服务器
sudo yum install -y mysql-community-server

# 启动MySQL服务并设置开机自启
sudo systemctl start mysqld
sudo systemctl enable mysqld

1.3 配置MySQL安全设置

首次安装后，MySQL会生成一个临时密码：

代码片段

# 查看临时密码
sudo grep 'temporary password' /var/log/mysqld.log

使用临时密码登录并运行安全脚本：

代码片段

mysql -uroot -p

输入临时密码后，执行以下SQL命令修改密码（请替换’YourNewPassword123!’为你自己的强密码）：

代码片段

ALTER USER 'root'@'localhost' IDENTIFIED BY 'YourNewPassword123!';

注意事项：
– MySQL8.0默认启用密码强度验证，需使用包含大小写字母、数字和特殊字符的强密码
– CentOS7上的SELinux可能会影响MySQL操作，如遇问题可暂时设置为permissive模式：

代码片段

sudo setenforce 0

1.4 （可选）创建专用数据库和用户

为LangChain项目创建专用数据库和用户：

代码片段

CREATE DATABASE langchain_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER 'langchain_user'@'%' IDENTIFIED BY 'LangChain_Pass123!';
GRANT ALL PRIVILEGES ON langchain_db.* TO 'langchain_user'@'%';
FLUSH PRIVILEGES;

第二部分：安装Python和必要组件

2.1 安装Python3.8+

CentOS7默认Python版本较旧，我们需要安装较新版本：

代码片段

# 安装EPEL仓库和SCL(Software Collections)
sudo yum install -y epel-release centos-release-scl

# 安装Python3.8和相关工具
sudo yum install -y rh-python38 rh-python38-python-devel gcc make zlib-devel openssl-devel libffi-devel

#启用Python3.8环境
scl enable rh-python38 bash

#验证Python版本
python3 --version && pip3 --version

实践经验：使用SCL可以在不干扰系统默认Python的情况下使用新版Python。

2.2 （推荐）创建Python虚拟环境

代码片段

python3 -m venv ~/langchain_env
source ~/langchain_env/bin/activate

第三部分：安装LangChain及相关组件

3.1 基础安装

代码片段

pip install langchain openai mysql-connector-python python-dotenv tiktoken transformers torch torchvision torchaudio sentence-transformers faiss-cpu numpy pandas scipy scikit-learn matplotlib seaborn jupyterlab ipywidgets

组件说明：
– mysql-connector-python: MySQL Python连接器
– tiktoken: OpenAI的分词器
– transformers: Hugging Face的Transformer模型库
– faiss-cpu: Facebook的相似性搜索库

3.2 （可选）GPU加速支持

如果你有NVIDIA GPU：

代码片段

pip uninstall torch torchvision torchaudio -y && pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 --no-cache-dir --upgrade --force-reinstall && pip install faiss-gpu sentence-transformers[gpu]

第四部分：配置LangChain连接MySQL数据库

4.1 MySQL表结构准备

首先在MySQL中创建一个用于测试的表：

代码片段

USE langchain_db;

CREATE TABLE documents (
    id INT AUTO_INCREMENT PRIMARY KEY,
    content TEXT NOT NULL,
    embedding JSON,
    metadata JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    FULLTEXT INDEX (content)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

Python示例代码：连接与操作数据库

创建一个demo.py文件：

代码片段

import os
from dotenv import load_dotenv
from langchain.document_loaders import TextLoader, DirectoryLoader, UnstructuredFileLoader, CSVLoader, PyPDFLoader, Docx2txtLoader, UnstructuredHTMLLoader, UnstructuredMarkdownLoader, UnstructuredWordDocumentLoader, UnstructuredPowerPointLoader, UnstructuredEmailLoader, UnstructuredEPubLoader, UnstructuredImageLoader, UnstructuredRTFLoader, NotebookLoader, SeleniumURLLoader, WebBaseLoader, GitLoader, NotionDirectoryLoader, ReadTheDocsLoader, SlackDirectoryLoader, TelegramChatFileLoader, TwitterTweetArchiveLoader, WhatsAppChatFileLoader, YouTubeAudioLoader 
from langchain.text_splitter import RecursiveCharacterTextSplitter, CharacterTextSplitter 
from langchain.docstore.document import Document 
from langchain.prompts import PromptTemplate 
from langchain.chains.question_answering import load_qa_chain 
from langchain.chains.summarize import load_summarize_chain 
from langchain.chains.conversation.memory import ConversationBufferMemory 
from langchain.chains.conversational_retrieval.prompts import CONDENSE_QUESTION_PROMPT 
from langchain.chains.conversational_retrieval.base import ConversationalRetrievalChain 
from langchain.chains.conversational_retrieval.prompts import QA_PROMPT 

import mysql.connector 

def setup_mysql_connection():
    """建立MySQL数据库连接"""
    try:
        conn = mysql.connector.connect(
            host="localhost",
            user="langchain_user",
            password="LangChain_Pass123!",
            database="langchain_db"
        )
        print("✅ MySQL连接成功")
        return conn

    except mysql.connector.Error as err:
        print(f"❌ MySQL连接失败: {err}")
        return None

def create_documents_table(conn):
    """创建文档表"""
    cursor = conn.cursor()

    create_table_query = """
    CREATE TABLE IF NOT EXISTS documents (
        id INT AUTO_INCREMENT PRIMARY KEY,
        content TEXT NOT NULL,
        embedding JSON,
        metadata JSON,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
        FULLTEXT INDEX (content)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
    """

    try:
        cursor.execute(create_table_query)
        conn.commit()
        print("✅ Documents表创建成功")

    except mysql.connector.Error as err:
        print(f"❌ Documents表创建失败: {err}")

def insert_document(conn, content):
    """插入文档到数据库"""
    cursor = conn.cursor()

    insert_query = "INSERT INTO documents (content) VALUES (%s)"

    try:
        cursor.execute(insert_query, (content,))
        conn.commit()
        print(f"✅ Document插入成功 (ID: {cursor.lastrowid})")

    except mysql.connector.Error as err:
        print(f"❌ Document插入失败: {err}")

def main():
    # MySQL连接配置示例（生产环境应使用环境变量）
    conn = setup_mysql_connection()

    if conn:
        # Create documents table if not exists (演示用)
        create_documents_table(conn)

        # Insert sample document (演示用)
        sample_text = "这是通过LangChain存储到MySQL的一个示例文档。"
        insert_document(conn, sample_text)

if __name__ == "__main__":
    main()

运行测试：

代码片段

python demo.py

LangChain与MySQL集成的高级示例

下面是一个更完整的示例，展示如何使用LangChain从文本文件加载内容，处理后存储到MySQL：

代码片段

import os  
import json  
import numpy as np  
from typing import List  
from dotenv import load_dotenv  
from mysql.connector import pooling  

class MySQLVectorStore:  

"""自定义的向量存储实现"""  

def __init__(self):  
self._setup_pool()  

def _setup_pool(self):  
"""建立连接池"""  
self.cnxpool = pooling.MySQLConnectionPool(  
pool_name="langchain_pool",  
pool_size=5,  
host="localhost",  
user="langchain_user",  
password="LangChain_Pass123!",  
database="langchain_db"  
)  

def add_texts(selftexts: List[str], embeddings: List[List[float]], metadatas: List[dict] = None):  

"""添加文本及其嵌入向量到数据库"""  

if metadatas is None:  

metadatas = [{} for _ in texts]  

cnx = self.cnxpool.get_connection()  

cursor = cnx.cursor()  

try:  

for textembeddingmetadata in zip(textsembeddingsmetadatas):  

insert_query = """INSERT INTO documents (contentembeddingmetadata) VALUES (%s%s%s)"""  

cursor.execute(insert_query(textjson.dumps(embedding)json.dumps(metadata)))  

cnx.commit()  

except Exception as e:  

print(f"Error inserting documents: {e}")  

cnx.rollback()  

finally:  

cursor.close()  

cnx.close()  


def similarity_search(selfquery_vector: List[float]k: int =5) -> List[dict]:  

"""基于余弦相似度搜索最相似的文档"""  


cnx = self.cnxpool.get_connection()  


cursor = cnx.cursor(dictionary=True)  


results = []  


try:  


cursor.execute("SELECT idcontentembeddingFROM documents")  


docs = cursor.fetchall()  


query_norm = np.linalg.norm(query_vector)  


for doc in docs:  


if doc['embedding'] is None: continue  


doc_vector = np.array(json.loads(doc['embedding']))  


doc_norm = np.linalg.norm(doc_vector)  


cosine_sim = np.dot(query_vectordoc_vector)/(query_norm*doc_norm) if query_norm*doc_norm !=0 else0  


results.append({**doc'score': cosine_sim})  


results.sort(key=lambda x:x['score']reverse=True)  


return results[:k]  


except Exception as e:  


print(f"Error searching documents:{e}") 


return [] 


finally:


cursor.close()


cnx.close()

if __name__ == "__main__":
load_dotenv()

vector_store= MySQLVectorStore()

sample_texts=[
"机器学习是人工智能的一个分支",
"深度学习使用神经网络进行特征学习",
"Transformer模型在NLP领域表现出色"
]

sample_embeddings=[
[0..10..20..3],
[0..11..21..31],
[0..12..22..32]
]

vector_store.add_texts(sample_textssample_embeddings)

query_vec=[0..11..21..31]
similar_docs=vector_store.similarity_search(query_vec)

print("相似文档搜索结果:")
for doc in similar_docs:
print(f"\nID:{doc['id']}\nScore:{doc['score']:.4f}\nContent:{doc['content']}")

常见问题解决

Q1：无法连接到MySQL服务器？
A1：
检查服务状态：

代码片段

sudo systemctl status mysqld`

检查防火墙设置：

代码片段

sudo firewall-cmd --list-all`

如需开放端口：

代码片段

sudo firewall-cmd --zone=public --add-port=3306/tcp --permanent`

Q2：Python连接报错”Authentication plugin ‘caching_sha2_password'”？

A2：
这是因为MySQL8默认使用了新的认证插件。两种解决方案：

方案一：修改用户认证方式（安全性较低）：

代码片段

ALTER USER 'langchain_user'@'%' IDENTIFIED WITH mysql_native_password BY 'password';
FLUSH PRIVILEGES;

方案二（推荐）：在客户端指定正确的认证插件：

代码片段

pip install mysqlclient cryptography`

然后在连接字符串中添加参数：

代码片段

auth_plugin='mysql_native_password'

Q3：如何优化大文本处理性能？

A3：
对于大文本处理建议：
1．合理设置max_allowed_packet参数（在my.cnf中）
2．考虑分块处理大文本后再存储或分析。
3．对频繁查询的字段建立适当索引。

Q4：如何升级到更高版本的Python？

A4：
对于需要更新版本的情况可以使用pyenv工具管理多版本Python而不影响系统稳定性。

总结与下一步建议通过本文我们完成了以下工作：

✅ CentOS7环境下安装了最新版MySQL8．0数据库系统。
✅配置了适合AI开发的Python环境。
✅集成了LangChain框架及其相关组件。
✅实现了LangChain与MySQL数据库的连接与交互。
✅提供了完整的代码示例和常见问题解决方案。