这是一个非常实际且重要的问题。将传统的关系型数据库 MySQL 用于向量存储和检索，是很多团队在技术选型初期，希望在不动现有架构的基础上，快速验证和落地 AI 应用的常见方案。

这篇文档会从深入思考的角度，为您详细剖析其原理、挑战，并提供一份包含代码的详细实战说明文档。

---

使用 MySQL 作为向量数据库：深度解析与实战指南

深入思考：MySQL 作为向量数据库的核心挑战与权衡

在我们直接进入操作之前，必须理解为什么 MySQL 不是 一个原生的向量数据库，以及这样做会面临哪些挑战。

1. 存储效率：向量是高维的浮点数数组。MySQL 的原生数据类型（如 TEXT, JSON, BLOB）并非为高效存储这种结构而设计。我们需要选择一种序列化方式（如存为二进制 BLOB），但这会增加应用层的复杂性。

2. 查询性能：向量检索的核心是计算“距离”（如余弦距离、欧氏距离）并找到最近的 K 个邻居。

   • 暴力搜索 (Brute-force)：在 MySQL 中，最直观的方法是取出数据库中的每一条向量，与目标向量进行计算，然后排序。这个操作的时间复杂度是 $O(N\cdot D)$，其中 N 是数据量，D 是向量维度。当 N 增大时，查询会变得极慢，无法满足线上服务的需求。
   • 缺乏原生索引：专业的向量数据库（如 Milvus, Pinecone, Weaviate）的核心优势是实现了近似最近邻（ANN）索引算法，如 HNSW、IVF-FLAT 等。这些算法能将搜索复杂度降低到对数级别（如 $O(\log N)$），从而实现毫秒级查询。标准版 MySQL (InnoDB 引擎) 完全没有这类索引。

3. 计算能力：向量距离计算是计算密集型任务。将大量的数学运算放在数据库层面，会极大地消耗 MySQL 服务器的 CPU 资源，可能影响到其他正常的业务查询。

结论与权衡：

• 适用场景：

  • 中小型项目：当向量数据量不大时（例如，十万条以内），暴力搜索的延迟可能还在可接受的范围内。
  • 已有 MySQL 体系：团队不想引入新的数据库组件，希望利用现有的运维和数据备份体系。
  • 混合查询：业务需要同时进行传统的 SQL 过滤（如 WHERE user_id = 123）和向量相似度搜索。这是 MySQL 的一个相对优势。

• 不适用场景：

  • 大规模数据：超过百万级别的向量数据。
  • 低延迟要求：需要稳定在几十毫秒内的查询响应。
  • 高并发查询：大量的向量搜索请求会拖垮数据库。

了解了这些，我们就可以开始设计一个在可接受范围内工作的方案。

---

文档正文：实战操作指南

这个指南将带你完成从文本向量化，到存入 MySQL，再到实现相似度查询的全过程。

第 1 步：核心概念与准备

1.1 向量嵌入 (Vector Embedding)
简单来说，就是用一个数学模型（例如，来自 Hugging Face 的 sentence-transformers）将一段文本（或其他数据）转换成一个由数字组成的向量（数组）。这个向量可以被认为是文本在多维空间中的“坐标”，语义相近的文本，其向量坐标也相近。

1.2 距离度量 (Distance Metrics)
我们用数学公式来衡量两个向量的“距离”。

• 余弦相似度 (Cosine Similarity)：衡量两个向量方向的相似性。值域为 [-1, 1]，值越大越相似。
  $$\text{similarity}=\cos (\theta )=\frac{\mathbf{A}\cdot \mathbf{B}}{\Vert \mathbf{A}\Vert \Vert \mathbf{B}\Vert }=\frac{{\sum }_{i=1}^n{A}_i{B}_i}{\sqrt{{\sum }_{i=1}^n{A}_i^2}\sqrt{{\sum }_{i=1}^n{B}_i^2}}$$
• 余弦距离 (Cosine Distance)：通常用 1 - Cosine Similarity 表示，值域为 [0, 2]，值越小越相似。我们在查询时通常按它升序排序。
• 欧氏距离 (Euclidean Distance / L2 Distance)：衡量两个向量在空间中的直线距离，值越小越相似。
  $$d(\mathbf{A},\mathbf{B})=\sqrt{\sum _{i=1}^n(A_i-B_i{)}^2}$$

第 2 步：设计 MySQL 数据表

存储向量的最佳选择是 BLOB (Binary Large Object) 类型。因为它存储的是原始二进制数据，没有字符集转换开销，空间效率最高。

我们创建一个表来存储知识片段：

CREATE TABLE `knowledge_vectors` (
  `id` INT AUTO_INCREMENT PRIMARY KEY,
  `content` TEXT NOT NULL COMMENT '原始文本内容',
  `vector_data` BLOB NOT NULL COMMENT '存储向量的二进制数据',
  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

• content：存储原始文本，便于查询后返回给用户。
• vector_data：存储文本对应的向量。假设我们使用的模型生成 768 维的 float32 向量，每个浮点数占 4 字节，那么一个向量将占用 768 * 4 = 3072 字节，BLOB 类型完全足够。

第 3 步：文本向量化并存入 MySQL (Python 示例)

我们将使用 Python 来演示这个过程，因为它有非常成熟的 AI 和数据库生态。

3.1 环境安装

pip install sentence-transformers mysql-connector-python numpy

3.2 编写存储代码
这段代码将连接数据库，使用预训练模型将文本列表向量化，然后将文本和向量存入数据库。

import numpy as np
import mysql.connector
from sentence_transformers import SentenceTransformer

# --- 1. 配置 ---
DB_CONFIG = {
    'host': 'localhost',
    'user': 'your_user',
    'password': 'your_password',
    'database': 'your_database'
}
# 使用一个通用的中文 embedding 模型
MODEL_NAME = 'shibing624/text2vec-base-chinese' 

# --- 2. 初始化模型和数据库连接 ---
print("正在加载向量化模型...")
model = SentenceTransformer(MODEL_NAME)
print("模型加载完毕。")

try:
    conn = mysql.connector.connect(**DB_CONFIG)
    cursor = conn.cursor()
    print("数据库连接成功。")
except mysql.connector.Error as err:
    print(f"数据库连接失败: {err}")
    exit()

# --- 3. 准备数据并进行向量化 ---
documents = [
    "人工智能是计算机科学的一个分支。",
    "机器学习是实现人工智能的一种方法。",
    "深度学习是机器学习的一个热门领域。",
    "中国的首都是北京。",
    "今天天气怎么样？"
]

print("正在将文本数据向量化...")
# a. 文本 -> 向量 (得到一个 numpy 数组列表)
embeddings = model.encode(documents, convert_to_tensor=False)
print(f"向量化完成，向量维度为: {embeddings.shape[1]}")

# --- 4. 将数据存入 MySQL ---
def store_vector_in_db(text, vector):
    """将单个文本和其向量存入数据库"""
    # b. 向量 (numpy array) -> 二进制 (bytes)
    vector_bytes = vector.astype(np.float32).tobytes()
    
    sql = "INSERT INTO knowledge_vectors (content, vector_data) VALUES (%s, %s)"
    try:
        cursor.execute(sql, (text, vector_bytes))
        # print(f"成功插入: {text}")
    except mysql.connector.Error as err:
        print(f"插入失败: {err}")

# 遍历所有数据并存储
for doc, emb in zip(documents, embeddings):
    store_vector_in_db(doc, emb)

conn.commit()
print(f"成功将 {len(documents)} 条数据存入数据库。")

# --- 5. 关闭连接 ---
cursor.close()
conn.close()
print("数据库连接已关闭。")

代码解析:

1. model.encode(): 核心函数，将一批文本转换成 numpy.ndarray 数组。
2. vector.astype(np.float32).tobytes(): 这是关键步骤。我们将 numpy 数组统一转换为 float32 类型，然后调用 .tobytes() 方法将其序列化为二进制字节流，这样就可以存入 BLOB 字段。

第 4 步：从 MySQL 中查询相似向量

这是最核心也最体现 MySQL 局限性的部分。我们将演示在应用层进行计算的方法，因为这是在没有 UDF (用户定义函数) 的情况下最通用和实际的方案。

import numpy as np
import mysql.connector
from sentence_transformers import SentenceTransformer

# --- 复用上面的配置和模型初始化 ---
DB_CONFIG = {
    'host': 'localhost',
    'user': 'your_user',
    'password': 'your_password',
    'database': 'your_database'
}
MODEL_NAME = 'shibing624/text2vec-base-chinese' 
model = SentenceTransformer(MODEL_NAME)
conn = mysql.connector.connect(**DB_CONFIG)
cursor = conn.cursor(dictionary=True) # 使用字典游标，方便获取列名

# --- 查询函数 ---
def search_similar_documents(query_text, top_k=3):
    """
    接收一个查询文本，返回最相似的 top_k 个文档
    """
    # 1. 将查询文本向量化
    query_vector = model.encode(query_text, convert_to_tensor=False)
    
    # 2. 从数据库中取出所有向量
    cursor.execute("SELECT id, content, vector_data FROM knowledge_vectors")
    all_data = cursor.fetchall()
    
    if not all_data:
        print("数据库中没有数据。")
        return []
        
    db_vectors = []
    # 3. 将 BLOB 数据反序列化为 numpy 向量
    vector_dim = embeddings.shape[1] # 向量维度
    for row in all_data:
        # 从 BLOB 字节流恢复为 numpy 数组
        vector = np.frombuffer(row['vector_data'], dtype=np.float32)
        # 做一个简单的校验
        if vector.shape[0] == vector_dim:
            db_vectors.append(vector)
        else:
            print(f"警告: ID {row['id']} 的向量维度不匹配，已跳过。")


    db_vectors = np.array(db_vectors)
    
    # 4. 在应用层计算余弦相似度
    # a. 计算点积 (Dot Product)
    dot_products = np.dot(db_vectors, query_vector)
    
    # b. 计算各自的范数 (Norm)
    db_norms = np.linalg.norm(db_vectors, axis=1)
    query_norm = np.linalg.norm(query_vector)
    
    # c. 计算余弦相似度
    similarities = dot_products / (db_norms * query_norm)
    
    # 5. 找到最相似的 top_k 个结果的索引
    # 使用 argsort 获取排序后的索引，[-top_k:] 取最后k个，[::-1] 进行反转得到从大到小的顺序
    top_k_indices = np.argsort(similarities)[-top_k:][::-1]
    
    # 6. 准备返回结果
    results = []
    for idx in top_k_indices:
        results.append({
            'id': all_data[idx]['id'],
            'content': all_data[idx]['content'],
            'similarity': similarities[idx]
        })
        
    return results

# --- 执行查询 ---
query = "AI有什么应用？"
search_results = search_similar_documents(query)

print(f"\n查询: '{query}'")
print("最相关的结果是:")
for res in search_results:
    print(f"  - ID: {res['id']}, 相似度: {res['similarity']:.4f}, 内容: {res['content']}")

# --- 关闭连接 ---
cursor.close()
conn.close()

代码解析：

1. np.frombuffer(..., dtype=np.float32): tobytes() 的逆操作。从二进制数据中恢复 numpy 数组，必须指定正确的数据类型 dtype。
2. 计算逻辑: 我们没有在 SQL 中做任何计算。而是将所有向量加载到内存中，利用 numpy 强大的矩阵运算能力快速计算出所有向量与查询向量的余弦相似度。
3. 优点: 简单直接，不给数据库增加计算压力，可以利用 numpy 的底层优化。
4. 缺点: 需要将 所有 向量加载到应用服务器的内存中。如果数据量达到几 GB，这将是不可接受的。

第 5 步：优化策略 - 混合搜索

为了解决全量加载向量的问题，我们可以采用“先过滤，后计算”的混合搜索策略。这是在 MySQL 中实现向量搜索最有价值的技巧。

假设我们的 knowledge_vectors 表还有一个 category 字段：

ALTER TABLE knowledge_vectors ADD COLUMN category VARCHAR(50);
-- 更新一些数据
UPDATE knowledge_vectors SET category = 'AI' WHERE id IN (1,2,3);
UPDATE knowledge_vectors SET category = 'Geography' WHERE id = 4;

现在，当用户查询时，我们可以先用传统的 WHERE 子句缩小范围，然后再对这个小得多的子集进行向量计算。

修改 search_similar_documents 函数：

def hybrid_search(query_text, category_filter, top_k=3):
    # ... 向量化查询文本 ...
    query_vector = model.encode(query_text, convert_to_tensor=False)

    # 1. 使用 WHERE 子句进行预过滤
    sql = "SELECT id, content, vector_data FROM knowledge_vectors WHERE category = %s"
    cursor.execute(sql, (category_filter,))
    filtered_data = cursor.fetchall()
    
    if not filtered_data:
        print(f"在分类 '{category_filter}' 下没有找到数据。")
        return []

    # 2. 仅对过滤后的子集进行向量相似度计算
    # ... (后续的计算逻辑与之前相同，只是操作的数据源是 filtered_data) ...
    # ...
    
    return results

# --- 执行混合搜索 ---
query = "机器学习是什么"
category = "AI"
results = hybrid_search(query, category)
# ... 打印结果 ...

这种方法极大地提升了性能和可行性，因为它将昂贵的向量计算量限制在了一个很小的范围内。

总结与展望

1. 标准方案: 对于中小型项目，BLOB 存储 + 混合搜索（SQL过滤 + 应用层计算） 是在 MySQL 中实现向量检索最平衡、最实用的方案。
2. 进阶方案 (UDF): 如果你对性能有更高要求，且不想将计算放在应用层，可以研究 C++ 编写 MySQL 的用户定义函数（UDF），将距离计算的逻辑直接集成到 MySQL 中。这样就可以写出 SELECT content, COSINE_DISTANCE(vector_data, ?) as dist FROM ... ORDER BY dist ASC 这样的 SQL。但这需要较高的开发和运维成本。
3. 未来方向 (MySQL HeatWave): Oracle 已经在其商业版的 MySQL HeatWave 云服务中加入了原生的向量存储和查询支持，包括 ANN 索引。如果你的项目在云上且预算充足，这是一个可以无缝升级的选择，能提供接近专业向量数据库的性能。

通过这篇指南，你应该已经深入理解了如何“Hack” MySQL 来满足基本的向量数据库需求，并清楚地知道了其能力边界和优化方向。