Migrar dados de um banco de dados de vetores para o AlloyDB

Criar um cluster e um usuário do AlloyDB

1. Crie um cluster e uma instância do AlloyDB.
   • Ative o IP público para executar este tutorial de qualquer lugar. Se você estiver usando o IP privado, execute este tutorial na sua VPC.
2. Crie ou selecione um usuário do banco de dados do AlloyDB.
   • Quando você cria a instância, um usuário postgres é criado com uma senha. Esse usuário tem permissões de superusuário.
   • Este tutorial usa a autenticação integrada para reduzir qualquer atrito de autenticação. A autenticação do IAM é possível usando o AlloyDBEngine.

Recuperar a amostra de código

1. Copie o exemplo de código do GitHub clonando o repositório:

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

2. Navegue até o diretório migrations:

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

Extrair dados de um banco de dados de vetores

1. Criar um cliente.

   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. Receba todos os dados do banco de dados.

   Pinecone

   Recupere os IDs de vetores do índice do 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

   Em seguida, extraia os registros por ID do índice do 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

Inicializar a tabela do AlloyDB

1. Defina o serviço de embedding.

   A interface VectorStore exige um serviço de incorporação. Esse fluxo de trabalho não gera novos embeddings, então a classe FakeEmbeddings é usada para evitar custos.

   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. Prepare a tabela do AlloyDB.

   1. Conecte-se ao AlloyDB usando uma conexão de IP público. Para mais informações, consulte Especificar o tipo de endereço 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. Crie uma tabela para copiar os dados, se ela ainda não existir.

      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
      )

Inicializar um objeto de repositório de vetores

Esse código adiciona mais metadados de incorporação de vetor à coluna langchain_metadata em formato JSON. Para tornar a filtragem mais eficiente, organize esses metadados em colunas separadas. Para mais informações, consulte Criar um repositório de vetores personalizado.

1. Para inicializar um objeto de armazenamento de vetores, execute o seguinte comando:

   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. Insira dados na tabela do 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)

Executar o script de migração

1. Configure o ambiente Python.

2. Instale as dependências de amostra:

   pip install -r requirements.txt

3. Execute a migração de amostra.

   Pinecone

   python migrate_pinecone_vectorstore_to_alloydb.py

   Faça as seguintes substituições antes de executar a amostra:

   • PINECONE_API_KEY: a chave de API do Pinecone.
   • PINECONE_NAMESPACE: o namespace do Pinecone.
   • PINECONE_INDEX_NAME: o nome do índice do Pinecone.
   • PROJECT_ID: o ID do projeto;
   • REGION: a região em que o cluster do AlloyDB é implantado.
   • CLUSTER: o nome do cluster.
   • INSTANCE: o nome da instância.
   • DB_NAME: o nome do banco de dados.
   • DB_USER: o nome do usuário do banco de dados.
   • DB_PWD: a senha secreta do banco de dados.

   Weaviate

   python migrate_weaviate_vectorstore_to_alloydb.py

   Faça as seguintes substituições antes de executar a amostra:

   • WEAVIATE_API_KEY: a chave de API do Weaviate.
   • WEAVIATE_CLUSTER_URL: o URL do cluster do Weaviate.
   • WEAVIATE_COLLECTION_NAME: o nome da coleção do Weaviate.
   • PROJECT_ID: o ID do projeto;
   • REGION: a região em que o cluster do AlloyDB é implantado.
   • CLUSTER: o nome do cluster.
   • INSTANCE: o nome da instância.
   • DB_NAME: o nome do banco de dados.
   • DB_USER: o nome do usuário do banco de dados.
   • DB_PWD: a senha secreta do banco de dados.

   Chroma

   python migrate_chromadb_vectorstore_to_alloydb.py

   Faça as seguintes substituições antes de executar a amostra:

   • CHROMADB_PATH: o caminho do banco de dados Chroma.
   • CHROMADB_COLLECTION_NAME: o nome da coleção do banco de dados Chroma.
   • PROJECT_ID: o ID do projeto;
   • REGION: a região em que o cluster do AlloyDB é implantado.
   • CLUSTER: o nome do cluster.
   • INSTANCE: o nome da instância.
   • DB_NAME: o nome do banco de dados.
   • DB_USER: o nome do usuário do banco de dados.
   • DB_PWD: a senha secreta do banco de dados.

   Qdrant

   python migrate_qdrant_vectorstore_to_alloydb.py

   Faça as seguintes substituições antes de executar a amostra:

   • QDRANT_PATH: o caminho do banco de dados Qdrant.
   • QDRANT_COLLECTION_NAME: o nome da coleção do Qdrant.
   • PROJECT_ID: o ID do projeto;
   • REGION: a região em que o cluster do AlloyDB é implantado.
   • CLUSTER: o nome do cluster.
   • INSTANCE: o nome da instância.
   • DB_NAME: o nome do banco de dados.
   • DB_USER: o nome do usuário do banco de dados.
   • DB_PWD: a senha secreta do banco de dados.

   Milvus

   python migrate_milvus_vectorstore_to_alloydb.py

   Faça as seguintes substituições antes de executar a amostra:

   • MILVUS_URI: o URI do Milvus.
   • MILVUS_COLLECTION_NAME: o nome da coleção do Milvus.
   • PROJECT_ID: o ID do projeto;
   • REGION: a região em que o cluster do AlloyDB é implantado.
   • CLUSTER: o nome do cluster.
   • INSTANCE: o nome da instância.
   • DB_NAME: o nome do banco de dados.
   • DB_USER: o nome do usuário do banco de dados.
   • DB_PWD: a senha secreta do banco de dados.

   Uma migração bem-sucedida imprime registros semelhantes aos seguintes sem erros:
   Migration completed, inserted all the batches of data to AlloyDB

4. Abra o AlloyDB Studio para conferir os dados migrados. Para mais informações, consulte Gerenciar dados com o AlloyDB Studio.