将数据从向量数据库迁移到 AlloyDB

本教程介绍了如何使用 LangChain VectorStore 将数据从第三方向量数据库迁移到 AlloyDB for PostgreSQL。本教程假设第三方向量数据库中的数据是使用 LangChain VectorStore 集成创建的。如果您在不使用 LangChain 的情况下将信息放入以下某个数据库中，可能需要修改下面提供的脚本，以匹配您数据的架构。支持以下向量数据库：

本教程假定您熟悉 Google Cloud、AlloyDB 和异步 Python 编程。

目标

本教程介绍了如何执行以下操作：

从现有向量数据库中提取数据。
连接到 AlloyDB。
初始化 AlloyDB 表。
初始化向量存储区对象。
运行迁移脚本以插入数据。

费用

在本文档中，您将使用 Google Cloud的以下收费组件：

AlloyDB for PostgreSQL

You might be eligible for a free trial cluster. For more information, see AlloyDB free trial clusters overview.

您可使用价格计算器根据您的预计使用情况来估算费用。

新 Google Cloud 用户可能有资格申请免费试用。

完成本文档中描述的任务后，您可以通过删除所创建的资源来避免继续计费。如需了解详情，请参阅清理。

准备工作

确保您拥有以下任一 LangChain 第三方数据库向量存储区：

启用结算功能及所需 API

在 Google Cloud 控制台的项目选择器页面上，选择或创建Google Cloud 项目。

注意：如果您不打算保留在此过程中创建的资源，请创建新的项目，而不要选择现有的项目。完成本教程介绍的步骤后，您可以删除所创建的项目，并移除与该项目关联的所有资源。
转到“项目选择器”
确保您的 Google Cloud 项目已启用结算功能。
启用创建和连接到 AlloyDB for PostgreSQL 所需的 Cloud API。

启用 API
1. 在确认项目步骤中，点击下一步以确认您要更改的项目的名称。
2. 在启用 API 步骤中，点击启用以启用以下内容：
  - AlloyDB API
  - Compute Engine API
  - Service Networking API

所需的角色

如要获得完成本教程中的任务所需的权限，您需要拥有以下 Identity and Access Management (IAM) 角色，以便创建表和插入数据：

Owner (roles/owner) 或 Editor (roles/editor)
如果用户不是所有者或编辑者，则需要拥有以下 IAM 角色及相应的 PostgreSQL 权限：
- AlloyDB Instance Client (roles/alloydb.client)
- Cloud AlloyDB Admin (roles/alloydb.admin)
- Compute Network User (roles/compute.networkUser)

如果您想使用 IAM 身份验证而非本教程中的内置身份验证来对数据库进行身份验证，请使用介绍如何使用 AlloyDB for PostgreSQL 通过 AlloyDBVectorStore 类存储向量嵌入的笔记本。

创建 AlloyDB 集群和用户

创建 AlloyDB 集群和实例。
- 启用公共 IP 以便随时随地运行本教程。如果您使用的是专用 IP，则必须在 VPC 中运行本教程。
创建或选择 AlloyDB 数据库用户。
- 创建实例时，系统会创建一个带有密码的 postgres 用户。此用户拥有超级用户权限。
- 本教程使用内置身份验证来提高身份验证流程的流畅度。可以使用 AlloyDBEngine 进行 IAM 身份验证。

检索代码示例

通过克隆代码库从 GitHub 复制代码示例：

git clone https://github.com/googleapis/langchain-google-alloydb-pg-python.git

导航到 migrations 目录：

cd langchain-google-alloydb-pg-python/samples/migrations

从现有向量数据库中提取数据

创建客户端。

Pinecone

from pinecone import Pinecone  # type: ignore

pinecone_client = Pinecone(api_key=pinecone_api_key)
pinecone_index = pinecone_client.Index(pinecone_index_name)

Weaviate

import weaviate

# For a locally running weaviate instance, use `weaviate.connect_to_local()`
weaviate_client = weaviate.connect_to_weaviate_cloud(
    cluster_url=weaviate_cluster_url,
    auth_credentials=weaviate.auth.AuthApiKey(weaviate_api_key),
)

Chroma

from langchain_chroma import Chroma

chromadb_client = Chroma(
    collection_name=chromadb_collection_name,
    embedding_function=embeddings_service,
    persist_directory=chromadb_path,
)

Qdrant

from qdrant_client import QdrantClient

qdrant_client = QdrantClient(path=qdrant_path)

Milvus

milvus_client = MilvusClient(uri=milvus_uri)

从数据库中获取所有数据。

Pinecone

从 Pinecone 索引中检索向量 ID：

results = pinecone_index.list_paginated(
    prefix="", namespace=pinecone_namespace, limit=pinecone_batch_size
)
ids = [v.id for v in results.vectors]
if ids:  # Prevents yielding an empty list.
    yield ids

# Check BOTH pagination and pagination.next
while results.pagination is not None and results.pagination.get("next") is not None:
    pagination_token = results.pagination.get("next")
    results = pinecone_index.list_paginated(
        prefix="",
        pagination_token=pagination_token,
        namespace=pinecone_namespace,
        limit=pinecone_batch_size,
    )

    # Extract and yield the next batch of IDs
    ids = [v.id for v in results.vectors]
    if ids:  # Prevents yielding an empty list.
        yield ids

然后，按 ID 从 Pinecone 索引中提取记录：

import uuid

# Iterate through the IDs and download their contents
for ids in id_iterator:
    all_data = pinecone_index.fetch(ids=ids, namespace=pinecone_namespace)
    ids = []
    embeddings = []
    contents = []
    metadatas = []

    # Process each vector in the current batch
    for doc in all_data["vectors"].values():
        # You might need to update this data translation logic according to one or more of your field names
        if pinecone_id_column_name in doc:
            # pinecone_id_column_name stores the unqiue identifier for the content
            ids.append(doc[pinecone_id_column_name])
        else:
            # Generate a uuid if pinecone_id_column_name is missing in source
            ids.append(str(uuid.uuid4()))
        # values is the vector embedding of the content
        embeddings.append(doc["values"])
        # Check if pinecone_content_column_name exists in metadata before accessing
        if pinecone_content_column_name in doc.metadata:
            # pinecone_content_column_name stores the content which was encoded
            contents.append(str(doc.metadata[pinecone_content_column_name]))
            # Remove pinecone_content_column_name after processing
            del doc.metadata[pinecone_content_column_name]
        else:
            # Handle the missing pinecone_content_column_name field appropriately
            contents.append("")
        # metadata is the additional context
        metadatas.append(doc["metadata"])

    # Yield the current batch of results
    yield ids, contents, embeddings, metadatas

Weaviate

# Iterate through the IDs and download their contents
weaviate_collection = weaviate_client.collections.get(weaviate_collection_name)
ids: list[str] = []
content: list[Any] = []
embeddings: list[list[float]] = []
metadatas: list[Any] = []

for item in weaviate_collection.iterator(include_vector=True):
    # You might need to update this data translation logic according to one or more of your field names
    # uuid is the unqiue identifier for the content
    ids.append(str(item.uuid))
    # weaviate_text_key is the content which was encoded
    content.append(item.properties[weaviate_text_key])
    # vector is the vector embedding of the content
    embeddings.append(item.vector["default"])  # type: ignore
    del item.properties[weaviate_text_key]  # type: ignore
    # properties is the additional context
    metadatas.append(item.properties)

    if len(ids) >= weaviate_batch_size:
        # Yield the current batch of results
        yield ids, content, embeddings, metadatas
        # Reset lists to start a new batch
        ids = []
        content = []
        embeddings = []
        metadatas = []

Chroma

# Iterate through the IDs and download their contents
offset = 0
while True:
    # You might need to update this data translation logic according to one or more of your field names
    # documents is the content which was encoded
    # embeddings is the vector embedding of the content
    # metadatas is the additional context
    docs = chromadb_client.get(
        include=["metadatas", "documents", "embeddings"],
        limit=chromadb_batch_size,
        offset=offset,
    )

    if len(docs["documents"]) == 0:
        break

    # ids is the unqiue identifier for the content
    yield docs["ids"], docs["documents"], docs["embeddings"].tolist(), docs[
        "metadatas"
    ]

    offset += chromadb_batch_size

Qdrant

# Iterate through the IDs and download their contents
offset = None
while True:
    docs, offset = qdrant_client.scroll(
        collection_name=qdrant_collection_name,
        with_vectors=True,
        limit=qdrant_batch_size,
        offset=offset,
        with_payload=True,
    )

    ids: List[str] = []
    contents: List[Any] = []
    embeddings: List[List[float]] = []
    metadatas: List[Any] = []

    for doc in docs:
        if doc.payload and doc.vector:
            # You might need to update this data translation logic according to one or more of your field names
            # id is the unqiue identifier for the content
            ids.append(str(doc.id))
            # page_content is the content which was encoded
            contents.append(doc.payload["page_content"])
            # vector is the vector embedding of the content
            embeddings.append(doc.vector)  # type: ignore
            # metatdata is the additional context
            metadatas.append(doc.payload["metadata"])

    yield ids, contents, embeddings, metadatas

    if not offset:
        break

Milvus

# Iterate through the IDs and download their contents
iterator = milvus_client.query_iterator(
    collection_name=milvus_collection_name,
    filter='pk >= "0"',
    output_fields=["pk", "text", "vector", "idv"],
    batch_size=milvus_batch_size,
)

while True:
    ids = []
    content = []
    embeddings = []
    metadatas = []
    page = iterator.next()
    if len(page) == 0:
        iterator.close()
        break
    for i in range(len(page)):
        # You might need to update this data translation logic according to one or more of your field names
        doc = page[i]
        # pk is the unqiue identifier for the content
        ids.append(doc["pk"])
        # text is the content which was encoded
        content.append(doc["text"])
        # vector is the vector embedding of the content
        embeddings.append(doc["vector"])
        del doc["pk"]
        del doc["text"]
        del doc["vector"]
        # doc is the additional context
        metadatas.append(doc)
    yield ids, content, embeddings, metadatas

初始化 AlloyDB 表

定义嵌入服务。

VectorStore 接口需要使用嵌入服务。此工作流不会生成新的嵌入，因此使用 FakeEmbeddings 类可避免产生任何费用。

Pinecone

# The VectorStore interface requires an embedding service. This workflow does not
# generate new embeddings, therefore FakeEmbeddings class is used to avoid any costs.
from langchain_core.embeddings import FakeEmbeddings

embeddings_service = FakeEmbeddings(size=vector_size)

Weaviate

# The VectorStore interface requires an embedding service. This workflow does not
# generate new embeddings, therefore FakeEmbeddings class is used to avoid any costs.
from langchain_core.embeddings import FakeEmbeddings

embeddings_service = FakeEmbeddings(size=vector_size)

Chroma

# The VectorStore interface requires an embedding service. This workflow does not
# generate new embeddings, therefore FakeEmbeddings class is used to avoid any costs.
from langchain_core.embeddings import FakeEmbeddings

embeddings_service = FakeEmbeddings(size=vector_size)

Qdrant

# The VectorStore interface requires an embedding service. This workflow does not
# generate new embeddings, therefore FakeEmbeddings class is used to avoid any costs.
from langchain_core.embeddings import FakeEmbeddings

embeddings_service = FakeEmbeddings(size=vector_size)

Milvus

# The VectorStore interface requires an embedding service. This workflow does not
# generate new embeddings, therefore FakeEmbeddings class is used to avoid any costs.
from langchain_core.embeddings import FakeEmbeddings

embeddings_service = FakeEmbeddings(size=vector_size)

准备 AlloyDB 表。

使用公共 IP 连接到 AlloyDB。如需了解详情，请参阅指定 IP 地址类型。

Pinecone

from langchain_google_alloydb_pg import AlloyDBEngine

alloydb_engine = await AlloyDBEngine.afrom_instance(
    project_id=project_id,
    region=region,
    cluster=cluster,
    instance=instance,
    database=db_name,
    user=db_user,
    password=db_pwd,
    ip_type=IPTypes.PUBLIC,  # Optionally use IPTypes.PRIVATE
)

Weaviate

from langchain_google_alloydb_pg import AlloyDBEngine

alloydb_engine = await AlloyDBEngine.afrom_instance(
    project_id=project_id,
    region=region,
    cluster=cluster,
    instance=instance,
    database=db_name,
    user=db_user,
    password=db_pwd,
    ip_type=IPTypes.PUBLIC,
)

Chroma

from langchain_google_alloydb_pg import AlloyDBEngine

alloydb_engine = await AlloyDBEngine.afrom_instance(
    project_id=project_id,
    region=region,
    cluster=cluster,
    instance=instance,
    database=db_name,
    user=db_user,
    password=db_pwd,
    ip_type=IPTypes.PUBLIC,
)

Qdrant

from langchain_google_alloydb_pg import AlloyDBEngine

alloydb_engine = await AlloyDBEngine.afrom_instance(
    project_id=project_id,
    region=region,
    cluster=cluster,
    instance=instance,
    database=db_name,
    user=db_user,
    password=db_pwd,
    ip_type=IPTypes.PUBLIC,
)

Milvus

from langchain_google_alloydb_pg import AlloyDBEngine

alloydb_engine = await AlloyDBEngine.afrom_instance(
    project_id=project_id,
    region=region,
    cluster=cluster,
    instance=instance,
    database=db_name,
    user=db_user,
    password=db_pwd,
    ip_type=IPTypes.PUBLIC,
)

创建一个要向其中复制数据的表（如果尚不存在）。

Pinecone

from langchain_google_alloydb_pg import Column

await alloydb_engine.ainit_vectorstore_table(
    table_name=alloydb_table,
    vector_size=vector_size,
    # Customize the ID column types if not using the UUID data type
    # id_column=Column("langchain_id", "TEXT"),  # Default is Column("langchain_id", "UUID")
    # overwrite_existing=True,  # Drop the old table and Create a new vector store table
)

Weaviate

await alloydb_engine.ainit_vectorstore_table(
    table_name=alloydb_table,
    vector_size=vector_size,
    # Customize the ID column types with `id_column` if not using the UUID data type
)

Chroma

await alloydb_engine.ainit_vectorstore_table(
    table_name=alloydb_table,
    vector_size=vector_size,
    # Customize the ID column types with `id_column` if not using the UUID data type
)

Qdrant

await alloydb_engine.ainit_vectorstore_table(
    table_name=alloydb_table,
    vector_size=vector_size,
    # Customize the ID column types with `id_column` if not using the UUID data type
)

Milvus

await alloydb_engine.ainit_vectorstore_table(
    table_name=alloydb_table,
    vector_size=vector_size,
    # Customize the ID column types with `id_column` if not using the UUID data type
)

初始化向量存储区对象

此代码以 JSON 格式向 langchain_metadata 列添加了额外的向量嵌入元数据。为了提高过滤效率，请将这些元数据整理到单独的列中。如需了解详情，请参阅创建自定义向量存储区。

如需初始化向量存储区对象，请运行以下命令：

Pinecone

from langchain_google_alloydb_pg import AlloyDBVectorStore

vs = await AlloyDBVectorStore.create(
    engine=alloydb_engine,
    embedding_service=embeddings_service,
    table_name=alloydb_table,
)

Weaviate

from langchain_google_alloydb_pg import AlloyDBVectorStore

vs = await AlloyDBVectorStore.create(
    engine=alloydb_engine,
    embedding_service=embeddings_service,
    table_name=alloydb_table,
)

Chroma

from langchain_google_alloydb_pg import AlloyDBVectorStore

vs = await AlloyDBVectorStore.create(
    engine=alloydb_engine,
    embedding_service=embeddings_service,
    table_name=alloydb_table,
)

Qdrant

from langchain_google_alloydb_pg import AlloyDBVectorStore

vs = await AlloyDBVectorStore.create(
    engine=alloydb_engine,
    embedding_service=embeddings_service,
    table_name=alloydb_table,
)

Milvus

from langchain_google_alloydb_pg import AlloyDBVectorStore

vs = await AlloyDBVectorStore.create(
    engine=alloydb_engine,
    embedding_service=embeddings_service,
    table_name=alloydb_table,
)

将数据插入到 AlloyDB 表中：

Pinecone

pending: set[Any] = set()
for ids, contents, embeddings, metadatas in data_iterator:
    pending.add(
        asyncio.ensure_future(
            vs.aadd_embeddings(
                texts=contents,
                embeddings=embeddings,
                metadatas=metadatas,
                ids=ids,
            )
        )
    )
    if len(pending) >= max_concurrency:
        _, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
if pending:
    await asyncio.wait(pending)

Weaviate

pending: set[Any] = set()
for ids, contents, embeddings, metadatas in data_iterator:
    pending.add(
        asyncio.ensure_future(
            vs.aadd_embeddings(
                texts=contents,
                embeddings=embeddings,
                metadatas=metadatas,
                ids=ids,
            )
        )
    )
    if len(pending) >= max_concurrency:
        _, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
if pending:
    await asyncio.wait(pending)

Chroma

pending: set[Any] = set()
for ids, contents, embeddings, metadatas in data_iterator:
    pending.add(
        asyncio.ensure_future(
            vs.aadd_embeddings(
                texts=contents,
                embeddings=embeddings,
                metadatas=metadatas,
                ids=ids,
            )
        )
    )
    if len(pending) >= max_concurrency:
        _, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
if pending:
    await asyncio.wait(pending)

Qdrant

pending: set[Any] = set()
for ids, contents, embeddings, metadatas in data_iterator:
    pending.add(
        asyncio.ensure_future(
            vs.aadd_embeddings(
                texts=contents,
                embeddings=embeddings,
                metadatas=metadatas,
                ids=ids,
            )
        )
    )
    if len(pending) >= max_concurrency:
        _, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
if pending:
    await asyncio.wait(pending)

Milvus

pending: set[Any] = set()
for ids, contents, embeddings, metadatas in data_iterator:
    pending.add(
        asyncio.ensure_future(
            vs.aadd_embeddings(
                texts=contents,
                embeddings=embeddings,
                metadatas=metadatas,
                ids=ids,
            )
        )
    )
    if len(pending) >= max_concurrency:
        _, pending = await asyncio.wait(
            pending, return_when=asyncio.FIRST_COMPLETED
        )
if pending:
    await asyncio.wait(pending)

运行迁移脚本

设置 Python 环境。
安装示例依赖项：
```
pip install -r requirements.txt
```
运行示例迁移。
Pinecone
```
python migrate_pinecone_vectorstore_to_alloydb.py
```
在运行示例之前，请先进行以下替换：
- PINECONE_API_KEY：Pinecone API 密钥。
- PINECONE_NAMESPACE：Pinecone 命名空间。
- PINECONE_INDEX_NAME：Pinecone 索引的名称。
- PROJECT_ID：项目 ID。
- REGION：部署 AlloyDB 集群的区域。
- CLUSTER：集群的名称。
- INSTANCE：实例的名称。
- DB_NAME：数据库的名称。
- DB_USER：数据库用户的名称。
- DB_PWD：数据库 Secret 密码。
Weaviate
```
python migrate_weaviate_vectorstore_to_alloydb.py
```
在运行示例之前，请先进行以下替换：
- WEAVIATE_API_KEY：Weaviate API 密钥。
- WEAVIATE_CLUSTER_URL：Weaviate 集群网址。
- WEAVIATE_COLLECTION_NAME：Weaviate 集合名称。
- PROJECT_ID：项目 ID。
- REGION：部署 AlloyDB 集群的区域。
- CLUSTER：集群的名称。
- INSTANCE：实例的名称。
- DB_NAME：数据库的名称。
- DB_USER：数据库用户的名称。
- DB_PWD：数据库 Secret 密码。
Chroma
```
python migrate_chromadb_vectorstore_to_alloydb.py
```
在运行示例之前，请先进行以下替换：
- CHROMADB_PATH：Chroma 数据库路径。
- CHROMADB_COLLECTION_NAME：Chroma 数据库集合的名称。
- PROJECT_ID：项目 ID。
- REGION：部署 AlloyDB 集群的区域。
- CLUSTER：集群的名称。
- INSTANCE：实例的名称。
- DB_NAME：数据库的名称。
- DB_USER：数据库用户的名称。
- DB_PWD：数据库 Secret 密码。
Qdrant
```
python migrate_qdrant_vectorstore_to_alloydb.py
```
在运行示例之前，请先进行以下替换：
- QDRANT_PATH：Qdrant 数据库路径。
- QDRANT_COLLECTION_NAME：Qdrant 集合名称。
- PROJECT_ID：项目 ID。
- REGION：部署 AlloyDB 集群的区域。
- CLUSTER：集群的名称。
- INSTANCE：实例的名称。
- DB_NAME：数据库的名称。
- DB_USER：数据库用户的名称。
- DB_PWD：数据库 Secret 密码。
Milvus
```
python migrate_milvus_vectorstore_to_alloydb.py
```
在运行示例之前，请先进行以下替换：
- MILVUS_URI：Milvus URI。
- MILVUS_COLLECTION_NAME：Milvus 集合的名称。
- PROJECT_ID：项目 ID。
- REGION：部署 AlloyDB 集群的区域。
- CLUSTER：集群的名称。
- INSTANCE：实例的名称。
- DB_NAME：数据库的名称。
- DB_USER：数据库用户的名称。
- DB_PWD：数据库 Secret 密码。
成功迁移会输出类似于以下内容的日志，且不包含任何错误：
Migration completed, inserted all the batches of data to AlloyDB
打开 AlloyDB Studio 以查看迁移的数据。如需了解详情，请参阅使用 AlloyDB Studio 管理数据。

清理

为避免因本教程中使用的资源导致您的 Google Cloud 账号产生费用，请删除包含这些资源的项目，或者保留项目但删除各个资源。

在 Google Cloud 控制台中，前往集群页面。

转到集群
在资源名称列中，点击您创建的集群的名称。
点击 删除集群。
在删除集群中，输入集群的名称以确认您要删除相应集群。
点击删除。

如果您在创建集群时创建了专用连接，请按以下步骤删除该专用连接：
前往 Google Cloud 控制台的“网络”页面，然后点击删除 VPC 网络。

将数据从向量数据库迁移到 AlloyDB 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

目标

费用

准备工作

启用结算功能及所需 API

所需的角色

创建 AlloyDB 集群和用户

检索代码示例

从现有向量数据库中提取数据

Pinecone

Weaviate

Chroma

Qdrant

Milvus

Pinecone

Weaviate

Chroma

Qdrant

Milvus

初始化 AlloyDB 表

Pinecone

Weaviate

Chroma

Qdrant

Milvus

Pinecone

Weaviate

Chroma

Qdrant

Milvus

Pinecone

Weaviate

Chroma

Qdrant

Milvus

初始化向量存储区对象

Pinecone

Weaviate

Chroma

Qdrant

Milvus

Pinecone

Weaviate

Chroma

Qdrant

Milvus

运行迁移脚本

Pinecone

Weaviate

Chroma

Qdrant

Milvus

清理

后续步骤

将数据从向量数据库迁移到 AlloyDB