使用 LlamaIndex 的 AlibabaCloudMySQLVectorStore 构建 MySQL 向量检索应用-开发者社区

使用 LlamaIndex 的 AlibabaCloudMySQLVectorStore 构建 MySQL 向量检索应用

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

本篇技术指南聚焦于 LlamaIndex 生态中的AlibabaCloudMySQLVectorStore（阿里云 RDS MySQL 向量存储集成），系统讲解其环境前提、安装方式、核心参数、建表结构、增删查改与元数据过滤的完整用法，并结合仓库源码剖析其底层实现原理。读者学完后，能够直接用阿里云 RDS MySQL 8.0 作为 LlamaIndex 的向量数据库，完成文档向量化写入、相似度检索与异步调用等实战任务。

背景：为什么需要 Alibaba Cloud MySQL 向量存储

在 RAG（检索增强生成）应用中，向量数据库负责存储文本块的嵌入向量并执行相似度检索。阿里云 RDS MySQL 在 8.0.36+ 版本中引入了原生向量类型（VECTOR）与向量函数（如VEC_FromText、VEC_DISTANCE_COSINE），使开发者无需额外部署独立的向量数据库，即可在已有 MySQL 实例上获得向量检索能力。

AlibabaCloudMySQLVectorStore正是这一能力的 LlamaIndex 接入层。它实现了 LlamaIndex 核心库中 BasePydanticVectorStore 抽象基类定义的标准接口（add、query、delete、get_nodes、count等），并针对阿里云 MySQL 的向量语法做了专门适配。其 API 参考文档位于 alibabacloud_mysql.md，类实现位于 base.py。

环境前提：RDS MySQL 版本与向量能力检查

在开始之前，请确认你的数据库实例满足以下条件（这些约束由源码 base.py 中的_check_vector_support()严格校验）：

必须是阿里云 RDS MySQL 实例：初始化时会执行SHOW VARIABLES LIKE 'rds_release_date'检查该变量是否存在，非 RDS 实例无法读取该变量会直接报错；
版本要求 RDS MySQL 8.0.36+：只有该版本起才提供向量函数支持；
rds_release_date必须 ≥ 20251031：源码中通过int(rds_release_date) < 20251031判定，低于该发布日期的实例会抛出ValueError；
向量函数可用：初始化时会执行SELECT VEC_FromText('[1,2,3]') IS NOT NULL探测VEC_FromText函数是否存在。

从源码结构看，这套校验被设计为初始化即验证：当perform_setup=True（默认值）时，构造函数会依次执行_connect()、_check_vector_support()与_create_table_if_not_exists()，提前暴露环境不兼容问题，而不是等到写入数据时才报错。

安装

通过 pip 安装集成包（详见集成包 README.md）：

pip install llama-index-vector-stores-alibabacloud-mysql

该包的依赖声明在 pyproject.toml 中，包括：

llama-index-core>=0.13.0,<0.15
sqlalchemy>=1.4.0,<3.0.0
pymysql>=1.0.0（同步驱动）
aiomysql>=0.2.0（异步驱动）

其中pymysql与aiomysql分别支撑同步与异步两条连接路径，SQLAlchemy 负责统一的 ORM 层封装。

快速开始：创建向量存储实例

直接实例化

from llama_index.vector_stores.alibabacloud_mysql import ( AlibabaCloudMySQLVectorStore, ) vector_store = AlibabaCloudMySQLVectorStore( table_name="llama_index_vectorstore", host="your-instance-endpoint.mysql.rds.aliyuncs.com", port=3306, user="llamaindex", password="password", database="vectordb", embed_dim=1536, # OpenAI 等模型的 embedding 维度 default_m=6, # 向量索引的 M 参数（HNSW 类索引的每节点连接数） distance_method="COSINE", # 距离度量：COSINE 或 EUCLIDEAN )

使用 from_params 类方法

from_params与直接构造完全等价，仅参数传递方式不同（源码 base.py）：

vector_store = AlibabaCloudMySQLVectorStore.from_params( host="your-instance-endpoint.mysql.rds.aliyuncs.com", port=3306, user="llamaindex", password="password", database="vectordb", )

构造参数详解

参数	类型	默认值	说明
`host`	str	必填	RDS MySQL 实例连接地址
`port`	int	必填	端口，通常为 3306
`user`	str	必填	数据库用户名
`password`	str	必填	数据库密码，构造连接串时经`quote_plus`转义
`database`	str	必填	数据库名
`table_name`	str	`"llama_index_table"`	向量存储表名，需符合 SQL 标识符规范（见下文参数校验）
`embed_dim`	int	`1536`	嵌入向量维度，需为正整数
`default_m`	int	`6`	向量索引的 M 值，需为正整数
`distance_method`	Literal	`"COSINE"`	距离度量，仅支持`"COSINE"`或`"EUCLIDEAN"`，非法值由 Pydantic 校验拦截
`perform_setup`	bool	`True`	是否自动执行向量能力检查与建表
`debug`	bool	`False`	是否开启 SQLAlchemy 的 SQL 回显（`echo`）日志

密码安全细节：源码在拼装连接串时使用了quote_plus(password)（base.py），自动转义密码中的特殊字符，避免密码包含@、:等字符时破坏 URL 连接串的解析。

构造参数校验

构造函数会对三个关键入参做防御性校验（base.py）：

table_name：通过正则^[a-zA-Z_][a-zA-Z0-9_]*$校验，只允许字母、数字、下划线且不能以数字开头，防止 SQL 注入与非法标识符；
embed_dim/default_m：必须是大于 0 的整数。

对应的单元测试见 test_alibabacloud_mysql.py，覆盖了合法标识符、数字开头、含连字符/空格/点号等非法场景，以及 0、负数、浮点数等非法数值场景。

连接机制：SQLAlchemy 双引擎架构

_connect()方法（base.py）同时创建两套连接：

同步引擎：使用pymysql驱动，create_engine(connection_string, echo=self.debug)；
异步引擎：将连接串中的mysql+pymysql://替换为mysql+aiomysql://后，用aiomysql驱动创建create_async_engine。

这种双引擎设计使得add、query、delete等同步方法与async_add、aquery、adelete等异步方法各自拥有独立的连接池与 session，同步/异步场景互不干扰。client属性则暴露底层的同步 SQLAlchemy 引擎（未初始化时为None）。

所有公开方法在调用时都会先执行self._initialize()确保连接与表结构就绪；perform_setup=False时跳过向量检查与建表，适用于表已由外部（如 DBA）预先创建好的场景。

自动建表：向量表 DDL 结构

当perform_setup=True时，初始化阶段会自动执行CREATE TABLE IF NOT EXISTS（base.py），生成的表结构如下：

CREATE TABLE IF NOT EXISTS `{table_name}` ( id VARCHAR(36) PRIMARY KEY, node_id VARCHAR(255) NOT NULL, text LONGTEXT, metadata JSON, embedding VECTOR({embed_dim}) NOT NULL, INDEX `node_id_index` (node_id), VECTOR INDEX (embedding) M={default_m} DISTANCE={distance_method} ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

各字段与索引的含义：

id：自增主键，由 MySQL 的UUID()函数在写入时生成；
node_id：LlamaIndex 节点的唯一 ID，建有普通索引便于按节点查询/删除；
text：LONGTEXT，保存节点原始文本内容；
metadata：JSON类型，保存节点元数据（含序列化的_node_content）；
embedding：VECTOR({embed_dim})，阿里云 MySQL 原生向量列，维度与embed_dim一致；
VECTOR INDEX ... M={default_m} DISTANCE={distance_method}：向量索引，M控制索引质量与召回精度，DISTANCE与构造参数distance_method保持一致；
字符集统一为utf8mb4 / utf8mb4_unicode_ci，兼容中文等多语言文本。

核心操作一：写入文档节点（add / async_add）

add(nodes)将 LlamaIndex 的BaseNode列表写入 MySQL（base.py）：

通过_node_to_table_row()将节点转换为行数据：node_id取节点 ID，text取MetadataMode.NONE模式下的纯文本内容，embedding取节点向量，metadata由node_to_metadata_dict(node, remove_text=True, flat_metadata=False)序列化（写入时去除冗余文本，避免 JSON 膨胀）；
执行INSERT ... VALUES (UUID(), :node_id, :text, VEC_FromText(:embedding), :metadata)，其中 embedding 以 JSON 字符串形式传给VEC_FromText解析为向量；
使用ON DUPLICATE KEY UPDATE实现幂等写入：同一node_id重复写入时自动更新text、embedding、metadata，适用于索引刷新与增量更新场景；
返回所有节点的node_id列表。

async_add(nodes)与add逻辑一致，仅切换为async_session执行（base.py）。

核心操作二：相似度检索（query / aquery）

query(VectorStoreQuery)执行向量相似度搜索（base.py）：

SELECT node_id, text, embedding, metadata, {distance_func}(embedding, VEC_FromText(:query_embedding)) AS distance FROM `{table_name}` {where_clause} ORDER BY distance LIMIT :limit

关键实现点：

距离函数选择：distance_method="COSINE"时使用VEC_DISTANCE_COSINE，否则使用VEC_DISTANCE_EUCLIDEAN，与建表时的DISTANCE参数保持一致；
相似度换算：底层返回的是"距离"（distance，越小越相似），代码通过similarity = 1 - distance换算为相似度分值（base.py），并填充进VectorStoreQueryResult；
结果还原：_db_rows_to_query_result()用metadata_dict_to_node()从元数据还原节点对象，再set_content()回填文本，保证返回的节点可直接用于下游合成（base.py）；
Top-K 限制：LIMIT :limit使用query.similarity_top_k控制返回条数。

查询模式限制：从源码可见，query与aquery仅支持VectorStoreQueryMode.DEFAULT，传入其他模式（如TEXT_SEARCH、HYBRID、MMR等，完整枚举见 types.py）会抛出NotImplementedError。对应测试见 test_alibabacloud_mysql.py。

核心操作三：元数据过滤

查询时可通过VectorStoreQuery.filters传入MetadataFilters实现结构化过滤，源码将其编译为基于 JSON 路径的 SQL 条件：

from llama_index.core.vector_stores.types import ( MetadataFilter, MetadataFilters, FilterOperator, VectorStoreQuery, ) filters = MetadataFilters( filters=[ MetadataFilter(key="category", value="技术", operator=FilterOperator.EQ), MetadataFilter(key="priority", value=1, operator=FilterOperator.GT), ], condition="and", ) query = VectorStoreQuery( query_embedding=[0.1, 0.2, ...], similarity_top_k=10, filters=filters, ) result = vector_store.query(query)

实现细节：

操作符映射：_to_mysql_operator()将 LlamaIndex 的FilterOperator枚举映射为 SQL 操作符——EQ→=、GT→>、LT→<、NE→!=、GTE→>=、LTE→<=、IN→IN、NIN→NOT IN，不支持的操作符记录 warning 后回退为=（base.py）；
JSON 字段取值：每个过滤条件编译为JSON_VALUE(metadata, '$.key') <op> :param，直接对metadataJSON 列取值比较；
参数化查询：所有过滤值均通过 SQLAlchemy 命名占位符:param_N绑定（IN/NIN 会展开为多个占位符），并用全局计数器保证参数名唯一，从机制上杜绝 SQL 注入（base.py）；
嵌套组合：MetadataFilters支持AND/OR组合，且过滤列表中可以嵌套子MetadataFilters（自动加括号），实现复杂布尔表达式（base.py）。

过滤相关的单测覆盖了操作符映射、IN 展开、AND/OR 组合及 SQL 文本断言，见 test_alibabacloud_mysql.py。

核心操作四：读取、删除与生命周期管理

方法	同步	异步	行为说明
读取节点	`get_nodes(node_ids=None, filters=None)`	—	按`node_id IN (...)`参数化查询并还原节点；不传`node_ids`时返回全表节点（base.py）
按文档删除	`delete(ref_doc_id)`	`adelete(ref_doc_id)`	通过`JSON_EXTRACT(metadata, '$.ref_doc_id') = :doc_id`删除某来源文档的全部节点（base.py）
按节点删除	`delete_nodes(node_ids)`	`adelete_nodes(node_ids)`	按`node_id IN (...)`参数化批量删除（base.py）
统计数量	`count()`	—	执行`SELECT COUNT(*)`返回表内节点总数（base.py）
清空数据	`clear()`	`aclear()`	`DELETE FROM`清空全部行（base.py）
删除表	`drop()`	—	`DROP TABLE IF EXISTS`，随后自动`close()`释放资源（base.py）
关闭连接	`close()`	`aclose()`	释放同步与异步引擎；`close()`针对运行中事件循环做了特殊处理（base.py）

上述方法的同步/异步配对与 LlamaIndexBasePydanticVectorStore的接口约定一致，异步版本是对同步逻辑的完整封装。

与 LlamaIndex 索引无缝集成

将AlibabaCloudMySQLVectorStore接入标准 RAG 流程，只需在构建索引时通过StorageContext注入向量存储：

from llama_index.core import VectorStoreIndex, StorageContext from llama_index.core.node_parser import SentenceSplitter from llama_index.core import SimpleDirectoryReader # 1. 加载并切分文档 documents = SimpleDirectoryReader("data").load_data() nodes = SentenceSplitter(chunk_size=512).get_nodes_from_documents(documents) # 2. 构建向量存储 vector_store = AlibabaCloudMySQLVectorStore( host="your-instance-endpoint.mysql.rds.aliyuncs.com", port=3306, user="llamaindex", password="password", database="vectordb", table_name="llama_index_vectorstore", embed_dim=1536, ) # 3. 创建索引并写入 storage_context = StorageContext.from_defaults(vector_store=vector_store) index = VectorStoreIndex(nodes, storage_context=storage_context) # 4. 查询 query_engine = index.as_query_engine(similarity_top_k=5) response = query_engine.query("你的问题")

该集成包遵循 LlamaIndex 的标准 Vector Store 协议（stores_text=True、is_embedding_query=True，见 base.py 与 types.py），因此VectorStoreIndex、RetrieverQueryEngine等上层组件可以直接使用，无需任何适配代码。测试文件 test_vector_stores_alibabacloud_mysql.py 亦验证了该类的 MRO 继承自BasePydanticVectorStore。

质量保障：测试与验证

集成包附带两套测试：

单元测试test_alibabacloud_mysql.py（1180 行）：通过 mock 会话隔离数据库依赖，覆盖参数校验、向量支持检查（含 4 种失败分支）、操作符映射、过滤子句构建、查询结果还原、SQL 文本断言等；
集成测试test_vector_stores_alibabacloud_mysql.py：验证类的继承关系。

README 说明集成测试需要真实可用的、支持向量检索的阿里云 MySQL 实例，环境不满足时测试会自动跳过（pytest -v即可运行）。

使用建议与注意事项

版本先行：务必确认实例满足 RDS MySQL 8.0.36+ 且rds_release_date >= 20251031，否则初始化即抛错；可在阿里云控制台或通过SHOW VARIABLES LIKE 'rds_release_date'提前核验；
向量维度一致性：embed_dim必须与所选 embedding 模型的输出维度严格一致（如 OpenAItext-embedding-3-small为 1536 维），建表后维度不可变更，需要重建表才可调整；
距离度量选择：COSINE适合绝大多数语义检索场景且对向量归一化不敏感；EUCLIDEAN适合距离含义明确的场景；需在初始化时与建表 DDL 一并确定；
过滤走 JSON 路径：元数据过滤通过JSON_VALUE提取字段，嵌套 JSON 的键路径为$.key，键名中若含特殊字符需注意转义；
查询模式受限：当前仅支持DEFAULT向量检索模式，混合检索（hybrid）、MMR 等高级模式需等待后续版本或自行扩展；
幂等写入：add依赖ON DUPLICATE KEY UPDATE，重复索引同一文档会覆盖旧数据而非报错，适合增量刷新场景；
生产环境关闭 debug：debug=True会开启 SQLAlchemy 的 SQL 回显，仅用于本地排障，生产环境应保持默认的False。