阿里云 MySQL 集成

阿里云 MySQL 是一种完全托管的关系型数据库服务，提供高可用性、可扩展性和安全性。

阿里云 MySQL 为企业级向量数据处理提供深度集成支持。它原生支持存储和计算最高 16,383 维的向量数据，集成了主流向量运算函数，并使用高度优化的分层可导航小世界（HNSW）算法实现高效的近似最近邻搜索。该功能还支持在全维度向量列上创建索引。

本指南提供了 alibabacloud-mysql 向量存储的快速入门概览。如需了解 alibabacloud-mysql 向量存储的所有功能、参数和配置的详细说明，请访问 langchain-alibabacloud-mysql。

配置

要访问 alibabacloud-mysql 向量存储，您需要创建小版本为 8.0.36 或更高版本的阿里云 RDS MySQL 实例，开启向量功能，配置访问权限，并安装 langchain-alibabacloud-mysql 集成包。

凭证

要连接阿里云 RDS MySQL 实例，您需要设置以下环境变量：

ALIBABACLOUD_MYSQL_HOST：RDS MySQL 主机地址
ALIBABACLOUD_MYSQL_PORT：MySQL 端口（默认：3306）
ALIBABACLOUD_MYSQL_USER：MySQL 用户名
ALIBABACLOUD_MYSQL_PASSWORD：MySQL 密码
ALIBABACLOUD_MYSQL_DATABASE：数据库名称

安装

LangChain alibabacloud-mysql 集成位于 langchain-alibabacloud-mysql 包中：

pip install -U langchain-alibabacloud-mysql

实例化

现在，我们可以使用您的 RDS MySQL 连接信息实例化向量存储：

Initialize vector store

import os
from langchain_alibabacloud_mysql import AlibabaCloudMySQL
from langchain_community.embeddings import DashScopeEmbeddings

# Initialize DashScope embeddings (Alibaba Cloud's embedding service)
embeddings = DashScopeEmbeddings(
    model="text-embedding-v4",
    dashscope_api_key=os.environ.get("DASHSCOPE_API_KEY"),
)
# Or you can use OpenAI embeddings
# embeddings = OpenAIEmbeddings()

# Initialize vector store
vector_store = AlibabaCloudMySQL(
    host=os.environ.get("ALIBABACLOUD_MYSQL_HOST", "localhost"),
    port=int(os.environ.get("ALIBABACLOUD_MYSQL_PORT", "3306")),
    user=os.environ.get("ALIBABACLOUD_MYSQL_USER", "root"),
    password=os.environ.get("ALIBABACLOUD_MYSQL_PASSWORD", ""),
    database=os.environ.get("ALIBABACLOUD_MYSQL_DATABASE", "test"),
    embedding=embeddings,
    table_name="langchain_vectors",
    distance_strategy="cosine",  # or "euclidean"
    hnsw_m=6,  # HNSW index M parameter (3-200)
)

要实例化向量存储，您需要提供嵌入模型。您可以使用 DashScope 嵌入（推荐用于阿里云）或 LangChain 中集成的其他嵌入模型（OpenAI 等）。如果您选择使用 DashScope 模型，可以在此处获取 API Key，并在以下代码中设置。

管理向量存储

添加条目

Add documents

from langchain_core.documents import Document

document_1 = Document(page_content="Alibaba", metadata={"source": "https://example.com"})
document_2 = Document(page_content="Cloud", metadata={"source": "https://example.com"})
document_3 = Document(page_content="RDS for MySQL", metadata={"source": "https://example.com"})
documents = [document_1, document_2, document_3]

vector_store.add_documents(documents=documents, ids=["1", "2", "3"])

更新条目

Update document by ID

updated_document = Document(
    page_content="Alibaba Cloud", metadata={"source": "https://another-example.com"}
)

vector_store.update_documents(document_id="1", document=updated_document)

删除条目

Delete documents by IDs

vector_store.delete(ids=["3"])

查询向量存储

一旦您的向量存储创建完成并添加了相关文档，您在运行链或智能体时很可能需要查询它。

直接查询

执行简单的相似度搜索可按如下方式进行：

Similarity search

results = vector_store.similarity_search(
    query="mysql", k=1, filter={"source": "https://example.com"}
)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

如果您希望执行相似度搜索并获取对应的分数，可以运行：

Similarity search with scores

results = vector_store.similarity_search_with_score(
    query="mysql", k=1, filter={"source": "https://example.com"}
)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

转换为检索器

您也可以将向量存储转换为检索器，以便在链中更方便地使用。

Create retriever

retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 1})
retriever.invoke("alibaba")

功能特性

阿里云 MySQL 向量存储支持大多数标准向量存储功能：

功能	是否支持
按 ID 删除	✅
过滤	✅
向量搜索	✅
带分数搜索	✅
异步操作	✅
通过标准测试	✅
多租户	❌
添加文档时指定 ID	✅

元数据过滤

您可以使用字典形式的过滤条件对搜索结果按元数据进行过滤：

Filter by metadata

# Search with metadata filter
results = vector_store.similarity_search(
    query="technology",
    k=5,
    filter={"category": "tech", "year": {"$gte": 2023}}
)

支持的过滤运算符：

$eq：等于
$ne：不等于
$gt：大于
$gte：大于或等于
$lt：小于
$lte：小于或等于
$in：在列表中
$nin：不在列表中
$like：LIKE 模式匹配

最大边际相关性（MMR）搜索

MMR 搜索通过平衡相关性和多样性来提供多元化的结果：

MMR search

results = vector_store.max_marginal_relevance_search(
    query="artificial intelligence",
    k=4,
    fetch_k=20,  # Number of candidates to consider
    lambda_mult=0.5,  # 0 = max diversity, 1 = max relevance
)

批量操作

高效地批量添加多个文档：

Batch add documents

texts = ["Document 1", "Document 2", "Document 3"]
metadatas = [
    {"source": "doc1.pdf"},
    {"source": "doc2.pdf"},
    {"source": "doc3.pdf"},
]
ids = vector_store.add_texts(texts, metadatas=metadatas)

按 ID 获取文档

通过 ID 检索特定文档：

Get by IDs

documents = vector_store.get_by_ids(["id1", "id2", "id3"])
for doc in documents:
    print(f"{doc.page_content} - {doc.metadata}")

计数与清空

获取向量总数或清空所有数据：

Count and clear

# Count total vectors
count = vector_store.count()
print(f"Total vectors: {count}")

# Clear all vectors
vector_store.clear()

异步操作

AlibabaCloud MySQL 向量存储支持所有主要方法的异步操作：

aadd_texts() - 异步添加文本
aadd_documents() - 异步添加文档
asimilarity_search() - 异步相似度搜索
asimilarity_search_with_score() - 异步带分数相似度搜索
amax_marginal_relevance_search() - 异步 MMR 搜索
adelete() - 异步删除向量
aget_by_ids() - 异步按 ID 获取文档
aclear() - 异步清空所有向量
acount() - 异步统计向量数量
aclose() - 异步关闭连接池

用于检索增强生成（RAG）

检索增强生成（RAG）将向量搜索与语言模型生成相结合，基于您的文档提供具有上下文意识的精准回答。

基础 RAG 工作流

以下是使用阿里云 MySQL 构建 RAG 应用的完整示例：

RAG example

import os
from langchain_alibabacloud_mysql import AlibabaCloudMySQL
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_classic.chains import create_retrieval_chain
from langchain_classic.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# Step 1: Initialize embeddings and vector store
embeddings = DashScopeEmbeddings(
    model="text-embedding-v4",
    dashscope_api_key=os.environ.get("DASHSCOPE_API_KEY"),
)

vector_store = AlibabaCloudMySQL(
    host=os.environ.get("ALIBABACLOUD_MYSQL_HOST", "localhost"),
    port=int(os.environ.get("ALIBABACLOUD_MYSQL_PORT", "3306")),
    user=os.environ.get("ALIBABACLOUD_MYSQL_USER", "root"),
    password=os.environ.get("ALIBABACLOUD_MYSQL_PASSWORD", ""),
    database=os.environ.get("ALIBABACLOUD_MYSQL_DATABASE", "test"),
    embedding=embeddings,
    table_name="langchain_vectors_rag",
)

# Step 2: Load and split documents
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
splits = text_splitter.split_documents(docs)

# Step 3: Add documents to vector store
vector_store.add_documents(documents=splits)

# Step 4: Create retriever
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# Step 5: Create RAG chain
llm = ChatTongyi()

prompt = ChatPromptTemplate.from_template(
    """Answer the following question based only on the provided context:

Context: {context}

Question: {input}"""
)

document_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)

# Step 6: Query
response = rag_chain.invoke({"input": "What is task decomposition?"})
print(response["answer"])

在智能体中使用检索器

您也可以将向量存储作为智能体中的检索工具：

RAG agent

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def retrieve_context(query: str) -> str:
    """Retrieve information to help answer a query."""
    retrieved_docs = vector_store.similarity_search(query, k=2)
    return "\n\n".join(
        f"Source: {doc.metadata}\nContent: {doc.page_content}"
        for doc in retrieved_docs
    )

tools = [retrieve_context]
llm = ChatTongyi()
agent = create_agent(
    llm,
    tools,
    system_prompt="You have access to a tool that retrieves context. Use it to help answer user queries.",
)

response = agent.invoke({"messages": [{"role": "user", "content": "What is task decomposition?"}]})

更多 RAG 指南和模式，请参阅：

更多阿里云 MySQL 的详细 RAG 演示和示例，请参阅：

API 参考

我们将很快更新 API 参考文档，更多详情请参阅 langchain-alibabacloud-mysql。

Edit this page on GitHub or file an issue.

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Popular Providers

Integrations by component

配置

凭证

安装

实例化

管理向量存储

添加条目

更新条目

删除条目

查询向量存储

直接查询

转换为检索器

功能特性

元数据过滤

最大边际相关性（MMR）搜索

批量操作

按 ID 获取文档

计数与清空

异步操作

用于检索增强生成（RAG）

基础 RAG 工作流

在智能体中使用检索器

API 参考

Popular Providers

Integrations by component

​配置

​凭证

​安装

​实例化

​管理向量存储

​添加条目

​更新条目

​删除条目

​查询向量存储

​直接查询

​转换为检索器

​功能特性

​元数据过滤

​最大边际相关性（MMR）搜索

​批量操作

​按 ID 获取文档

​计数与清空

​异步操作

​用于检索增强生成（RAG）

​基础 RAG 工作流

​在智能体中使用检索器

​API 参考

配置

凭证

安装

实例化

管理向量存储

添加条目

更新条目

删除条目

查询向量存储

直接查询

转换为检索器

功能特性

元数据过滤

最大边际相关性（MMR）搜索

批量操作

按 ID 获取文档

计数与清空

异步操作

用于检索增强生成（RAG）

基础 RAG 工作流

在智能体中使用检索器

API 参考