当前位置：首页 > news >正文

深度解析 mysql 作为向量数据库如何存储和查询向量数据

news 2025/10/18 15:20:38

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

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

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

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

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

存储效率：向量是高维的浮点数数组。MySQL 的原生数据类型（如 TEXT, JSON, BLOB）并非为高效存储这种结构而设计。我们需要选择一种序列化方式（如存为二进制 BLOB），但这会增加应用层的复杂性。
查询性能：向量检索的核心是计算“距离”（如余弦距离、欧氏距离）并找到最近的 K 个邻居。
- 暴力搜索 (Brute-force)：在 MySQL 中，最直观的方法是取出数据库中的每一条向量，与目标向量进行计算，然后排序。这个操作的时间复杂度是 $\cdot D)$ ，其中 N 是数据量，D 是向量维度。当 N 增大时，查询会变得极慢，无法满足线上服务的需求。
- 缺乏原生索引：专业的向量数据库（如 Milvus, Pinecone, Weaviate）的核心优势是实现了近似最近邻（ANN）索引算法，如 HNSW、IVF-FLAT 等。这些算法能将搜索复杂度降低到对数级别（如 $O(log⁡N)O(\log N)$ ），从而实现毫秒级查询。标准版 MySQL (InnoDB 引擎) 完全没有这类索引。
计算能力：向量距离计算是计算密集型任务。将大量的数学运算放在数据库层面，会极大地消耗 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]，值越大越相似。
$similarity=cos⁡(θ)=A⋅B∥A∥∥B∥=∑i=1nAiBi∑i=1nAi2∑i=1nBi2\text{similarity} = \cos(\theta) = \frac{\mathbf{A} \cdot \mathbf{B}}{\|\mathbf{A}\| \|\mathbf{B}\|} = \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(A,B)=∑i=1n(Ai−Bi)2d(\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("数据库连接已关闭。")

代码解析:

model.encode(): 核心函数，将一批文本转换成 numpy.ndarray 数组。
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()

代码解析：

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

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

总结与展望

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

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

查看全文

http://www.dtcms.com/a/496900.html