阿里云MySQL集成 - 以编程方式使用文档

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

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

本指南提供了快速入门 AlibabaCloudMySQL 向量存储的快速概览。如需获取所有alibabacloud-mysql向量存储功能、参数和配置的详细列表，请前往 langchain-alibabacloud-mysql.

设置

要访问alibabacloud-mysql向量存储，您需要创建一个阿里云RDS MySQL实例，次版本8.0.36或更高，开启向量功能, 使其可访问，并安装 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 package:

    pip install -U langchain-alibabacloud-mysql

    uv add langchain-alibabacloud-mysql

实例化

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

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)
)

管理向量存储

添加项目

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"])

更新项目

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

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

删除项目

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

查询向量存储

一旦您的向量存储被创建并且添加了相关文档，您很可能希望在使用链或代理运行时对其进行查询。

直接查询

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

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}]")

如果您想执行相似性搜索并接收相应的分数，可以运行：

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}]")

转换为检索器

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

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

功能

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

功能	支持
按ID删除	✅
过滤	✅
向量搜索	✅
带分数搜索	✅
异步	✅
通过标准测试	✅
多租户	❌
添加文档中的ID	✅

元数据过滤

您可以使用字典式过滤器按元数据过滤搜索结果：

# 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搜索通过平衡相关性和多样性提供多样化的结果：

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
)

批量操作

高效地一次添加多个文档：

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检索特定文档：

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

计数和清除

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

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

# Clear all vectors
vector_store.clear()

异步操作

阿里云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工作流程

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

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"])

与代理一起使用检索器

您还可以将向量存储用作代理中的检索工具：

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 了解更多详情。

功能	支持
按ID删除	✅
过滤	✅
向量搜索	✅
带分数搜索	✅
异步	✅
通过标准测试	✅
多租户	❌
添加文档中的ID	✅