大模型时代下的MySQL：从关系型数据库到智能数据基座

-- 创建支持向量检索的数据库
CREATE DATABASE ai_shop CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
USE ai_shop;

-- 创建产品表，包含向量字段（1536维适配OpenAI text-embedding-3-small）
CREATE TABLE products (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) NOT NULL COMMENT '产品名称',
    description TEXT COMMENT '产品描述',
    description_vector VECTOR(1536) NOT NULL COMMENT '描述文本嵌入向量',
    category VARCHAR(64) COMMENT '产品分类',
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    -- 为向量字段创建HNSW索引（MySQL 8.4.0+）
    INDEX description_vector_idx (description_vector) USING VECTOR
) COMMENT='产品表（带向量语义检索）';

-- 创建余弦相似度计算函数（简化查询）
CREATE FUNCTION cosine_similarity(a VECTOR, b VECTOR) 
RETURNS FLOAT DETERMINISTIC 
RETURN COSINE_SIMILARITY(a, b);

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j</artifactId>
    <version>0.29.0</version>
</dependency>
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-open-ai</artifactId>
    <version>0.29.0</version>
</dependency>

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/ai_shop?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    username: ${DB_USERNAME}
    password: ${DB_PASSWORD}
openai:
  api-key: ${OPENAI_API_KEY}
  embedding-model: text-embedding-3-small

// Product.java
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;

@Table("products")
public class Product {
    @Id
    private Long id;
    private String name;
    private String description;
    private Vector descriptionVector;  // 自定义Vector类型
    private String category;
    private LocalDateTime createdAt;
    // getters/setters...
}

// ProductRepository.java
@Repository
public interface ProductRepository extends CrudRepository<Product, Long> {
    // 向量相似度查询 - 查找与目标向量最相似的N个产品
    @Query("""
        SELECT p.*, 
               COSINE_SIMILARITY(p.description_vector, :targetVector) AS similarity
        FROM products p
        WHERE p.category = :category  -- 结构化过滤 + 向量检索复合查询
        ORDER BY COSINE_SIMILARITY(p.description_vector, :targetVector) DESC
        LIMIT :limit
    """)
    List<ProductWithSimilarity> findSimilarByCategory(
        @Param("targetVector") Vector targetVector,
        @Param("category") String category,
        @Param("limit") int limit
    );
}

@Service
public class SemanticSearchService {
    
    @Autowired
    private ProductRepository productRepository;
    
    private final EmbeddingModel embeddingModel;
    
    public SemanticSearchService() {
        this.embeddingModel = new OpenAiEmbeddingModel(
            OpenAiEmbeddingModelName.TEXT_EMBEDDING_3_SMALL
        );
    }
    
    // 将文本转换为向量
    public Vector embedText(String text) {
        List<Float> embedding = embeddingModel.embed(text).content();
        return new Vector(embedding);
    }
    
    // 语义搜索：用户输入自然语言，返回语义匹配的产品
    public List<ProductWithSimilarity> semanticSearch(String query, String category, int limit) {
        Vector queryVector = embedText(query);
        return productRepository.findSimilarByCategory(queryVector, category, limit);
    }
    
    // 批量入库：将产品描述向量化后存入MySQL
    public void indexProducts(List<Product> products) {
        products.forEach(p -> {
            Vector vec = embedText(p.getDescription());
            p.setDescriptionVector(vec);
            productRepository.save(p);
        });
    }
}

-- 查询"2024年发布的、与'MySQL向量检索'语义相似的技术文档"
SELECT d.id, d.title, d.publish_year,
       COSINE_SIMILARITY(d.content_vector, VECTOR('[0.12, -0.34, ...]')) AS similarity
FROM documents d
WHERE d.publish_year = 2024
  AND d.doc_type = 'technical'
  AND COSINE_SIMILARITY(d.content_vector, VECTOR('[0.12, -0.34, ...]')) > 0.75
ORDER BY similarity DESC
LIMIT 10;

# 全局安装SQL-MCP
npm i -g @polarisxb/sql-mcp

# Stdio模式启动（适合Cursor等本地工具集成）
sql-mcp --type mysql \
  --host 127.0.0.1 --port 3306 \
  --user root --password your_password --database your_db \
  --transport stdio

# HTTP模式启动（作为独立服务）
sql-mcp --type mysql \
  --host 127.0.0.1 --port 3306 \
  --user root --password your_password --database your_db \
  --transport http --port 8080

# Docker部署TencentDB MySQL MCP
git clone https://github.com/TencentCloud/mcp-server-cdb.git
cd mcp-server-cdb
docker build -t tencentcloud-mcp-server-cdb .
docker run -d --name mysql-mcp -p 9000:9000 --env-file .env tencentcloud-mcp-server-cdb

# 验证服务
curl http://localhost:9000/health

@Configuration
public class MCPClientConfig {
    
    @Bean
    public McpClient mysqlMcpClient() {
        return McpClient.builder()
            .transport(new HttpMcpTransport("http://localhost:8080"))
            .build();
    }
}

@Service
public class IntelligentQueryService {
    
    @Autowired
    private McpClient mcpClient;
    
    // 大模型驱动的自然语言查询
    public String queryWithLLM(String naturalLanguageQuery) {
        // 1. 获取数据库Schema元数据
        String schema = mcpClient.callTool("describeDatabase", "your_db");
        
        // 2. 构建Prompt，让大模型生成SQL
        String prompt = buildPrompt(schema, naturalLanguageQuery);
        String sql = llm.generate(prompt);
        
        // 3. 通过MCP执行只读查询
        String result = mcpClient.callTool("executeQuery", sql);
        return result;
    }
    
    // 查询优化：让MCP分析并优化SQL
    public String optimizeWithMCP(String sql) {
        return mcpClient.callTool("optimizeQuery", sql);
    }
}

用户查询 → Schema Investigator（过滤无关表）
        → Query Planner（生成多个SQL候选）
        → Validator（执行反馈验证）
        → Fix Agent（根据错误修正）
        → Selection Agent（选出最优SQL）
        → 返回结果

class SQLErrorClassifier:
    ERROR_TYPES = {
        'SCHEMA_LINKING': '表/字段链接错误',
        'JOIN_INCONSISTENCY': 'JOIN条件不一致',
        'AGGREGATION_MISUSE': '聚合函数滥用',
        'LOGICAL_SEMANTIC': '语义意图不匹配'  # 语法正确但逻辑错误
    }
    
    def classify_and_fix(self, sql: str, error: str) -> str:
        error_type = self.classify(error)
        fix_prompt = self.build_fix_prompt(sql, error_type)
        return llm.generate(fix_prompt)

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import MySQLVectorStore
from langchain.llms import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader
import MySQLdb

# 1. 配置MySQL连接（含向量扩展）
connection_params = {
    'host': 'localhost',
    'user': 'root',
    'password': 'your_password',
    'database': 'rag_knowledge',
    'use_pure': True
}

# 2. 初始化向量存储（MySQL作为向量数据库）
vector_store = MySQLVectorStore(
    connection_params=connection_params,
    embedding_function=OpenAIEmbeddings(),
    table_name='doc_embeddings',
    vector_column='embedding',
    metadata_columns=['source', 'chunk_index']
)

# 3. 文档加载与分块
loader = TextLoader('knowledge_base.txt')
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
texts = text_splitter.split_documents(documents)

# 4. 向量化入库（存入MySQL）
vector_store.add_documents(texts)

# 5. 构建RAG检索链
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model='gpt-4', temperature=0),
    chain_type='stuff',
    retriever=vector_store.as_retriever(search_kwargs={'k': 3})
)

# 6. 执行查询
response = qa_chain.run("什么是检索增强生成技术？")
print(response)

# 7. 会话历史持久化（MySQL存储多轮对话上下文）
def save_conversation(session_id, user_query, assistant_response):
    conn = MySQLdb.connect(**connection_params)
    cursor = conn.cursor()
    cursor.execute("""
        INSERT INTO conversation_history (session_id, user_query, assistant_response, created_at)
        VALUES (%s, %s, %s, NOW())
    """, (session_id, user_query, assistant_response))
    conn.commit()
    cursor.close()
    conn.close()