Memigrasikan data dari database vektor ke AlloyDB

Buat cluster dan pengguna AlloyDB

1. Buat cluster dan instance AlloyDB.
   • Aktifkan IP Publik untuk menjalankan tutorial ini dari mana saja. Jika Anda menggunakan IP Pribadi, Anda harus menjalankan tutorial ini dari dalam VPC.
2. Buat atau pilih pengguna database AlloyDB.
   • Saat Anda membuat instance, pengguna postgres akan dibuat dengan sandi. Pengguna ini memiliki izin superuser.
   • Tutorial ini menggunakan autentikasi bawaan untuk mengurangi hambatan autentikasi. Autentikasi IAM dapat dilakukan menggunakan AlloyDBEngine.

Mengambil contoh kode

1. Salin contoh kode dari GitHub dengan meng-clone repositori:

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

2. Buka direktori migrations:

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

Mengekstrak data dari database vektor yang ada

1. Buat klien.

   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)

2. Dapatkan semua data dari database.

   Pinecone

   Mengambil ID vektor dari indeks Pinecone:

   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

   Kemudian, ambil rekaman menurut ID dari indeks 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

Lakukan inisialisasi tabel AlloyDB

1. Tentukan layanan penyematan.

   Antarmuka VectorStore memerlukan layanan penyematan. Alur kerja ini tidak membuat embedding baru, sehingga class FakeEmbeddings digunakan untuk menghindari biaya apa pun.

   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)

2. Siapkan tabel AlloyDB.

   1. Menghubungkan ke AlloyDB menggunakan koneksi IP publik. Untuk mengetahui informasi selengkapnya, lihat Menentukan Jenis Alamat 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,
      )

   2. Buat tabel untuk menyalin data, jika belum ada.

      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
      )

Lakukan inisialisasi objek penyimpanan vektor

Kode ini menambahkan metadata penyematan vektor tambahan ke kolom langchain_metadata dalam format JSON. Untuk membuat pemfilteran lebih efisien, atur metadata ini ke dalam kolom terpisah. Untuk mengetahui informasi selengkapnya, lihat Membuat Penyimpanan Vektor kustom.

1. Untuk menginisialisasi objek penyimpanan vektor, jalankan perintah berikut:

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

2. Sisipkan data ke dalam tabel 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)

Jalankan skrip migrasi

1. Siapkan lingkungan Python.

2. Instal dependensi contoh:

   pip install -r requirements.txt

3. Jalankan migrasi contoh.

   Pinecone

   python migrate_pinecone_vectorstore_to_alloydb.py

   Lakukan penggantian berikut sebelum Anda menjalankan contoh:

   • PINECONE_API_KEY: kunci API Pinecone.
   • PINECONE_NAMESPACE: namespace Pinecone.
   • PINECONE_INDEX_NAME: nama indeks Pinecone.
   • PROJECT_ID: project ID.
   • REGION: region tempat cluster AlloyDB di-deploy.
   • CLUSTER: nama cluster.
   • INSTANCE: nama instance.
   • DB_NAME: nama database.
   • DB_USER: nama pengguna database.
   • DB_PWD: sandi rahasia database.

   Weaviate

   python migrate_weaviate_vectorstore_to_alloydb.py

   Lakukan penggantian berikut sebelum Anda menjalankan contoh:

   • WEAVIATE_API_KEY: kunci API Weaviate.
   • WEAVIATE_CLUSTER_URL: URL cluster Weaviate.
   • WEAVIATE_COLLECTION_NAME: nama koleksi Weaviate.
   • PROJECT_ID: project ID.
   • REGION: region tempat cluster AlloyDB di-deploy.
   • CLUSTER: nama cluster.
   • INSTANCE: nama instance.
   • DB_NAME: nama database.
   • DB_USER: nama pengguna database.
   • DB_PWD: sandi rahasia database.

   Chroma

   python migrate_chromadb_vectorstore_to_alloydb.py

   Lakukan penggantian berikut sebelum Anda menjalankan contoh:

   • CHROMADB_PATH: jalur database Chroma.
   • CHROMADB_COLLECTION_NAME: nama koleksi database Chroma.
   • PROJECT_ID: project ID.
   • REGION: region tempat cluster AlloyDB di-deploy.
   • CLUSTER: nama cluster.
   • INSTANCE: nama instance.
   • DB_NAME: nama database.
   • DB_USER: nama pengguna database.
   • DB_PWD: sandi rahasia database.

   Qdrant

   python migrate_qdrant_vectorstore_to_alloydb.py

   Lakukan penggantian berikut sebelum Anda menjalankan contoh:

   • QDRANT_PATH: jalur database Qdrant.
   • QDRANT_COLLECTION_NAME: nama koleksi Qdrant.
   • PROJECT_ID: project ID.
   • REGION: region tempat cluster AlloyDB di-deploy.
   • CLUSTER: nama cluster.
   • INSTANCE: nama instance.
   • DB_NAME: nama database.
   • DB_USER: nama pengguna database.
   • DB_PWD: sandi rahasia database.

   Milvus

   python migrate_milvus_vectorstore_to_alloydb.py

   Lakukan penggantian berikut sebelum Anda menjalankan contoh:

   • MILVUS_URI: URI Milvus.
   • MILVUS_COLLECTION_NAME: nama koleksi Milvus.
   • PROJECT_ID: project ID.
   • REGION: region tempat cluster AlloyDB di-deploy.
   • CLUSTER: nama cluster.
   • INSTANCE: nama instance.
   • DB_NAME: nama database.
   • DB_USER: nama pengguna database.
   • DB_PWD: sandi rahasia database.

   Migrasi yang berhasil akan mencetak log yang mirip dengan berikut tanpa error:
   Migration completed, inserted all the batches of data to AlloyDB

4. Buka AlloyDB Studio untuk melihat data yang dimigrasikan. Untuk mengetahui informasi selengkapnya, lihat Mengelola data Anda menggunakan AlloyDB Studio.