Azure OpenAI para big data

Pode usar o serviço Azure OpenAI para resolver muitas tarefas em linguagem natural solicitando a API de completamento. Para facilitar a ampliação dos seus fluxos de prompting de alguns exemplos para grandes conjuntos de dados, o serviço Azure OpenAI integra-se com a biblioteca distribuída de aprendizagem automática SynapseML. Ao usar esta integração, pode utilizar a framework de computação distribuída Apache Spark para processar milhões de prompts com o serviço da OpenAI. Este tutorial mostra como aplicar grandes modelos de linguagem numa escala distribuída usando o Azure OpenAI e o Microsoft Fabric.

Pré-requisitos

Os pré-requisitos chave para este quickstart incluem um recurso Azure OpenAI funcional e um cluster Apache Spark com SynapseML instalado.

• Faça uma subscrição Microsoft Fabric. Ou, inscreva-se para um teste gratuito Microsoft Fabric.

• Iniciar sessão em Microsoft Fabric.

• Altera para o Fabric usando o alternador de experiência no canto inferior esquerdo da sua página inicial.

  

• Aceda à Experiência de Ciência de Dados no Microsoft Fabric.
• Crie um novo bloco de notas.
• Um recurso Azure OpenAI - criar um recurso

Importar este guia como um bloco de notas

A próxima etapa é adicionar esse código ao cluster do Spark. Você pode criar um bloco de anotações em sua plataforma Spark e copiar o código para este bloco de anotações para executar a demonstração. Ou descarregar o caderno e importá-lo para o Synapse Analytics.

1. Faça o download desta demonstração como um bloco de anotações (selecione Raw e salve o arquivo)
2. Importe o notebook para o espaço de trabalho Synapse ou, se estiver usando o Fabric, importe para o Fabric Workspace
3. Instale o SynapseML no seu cluster. Consulte as instruções de instalação do Synapse na parte inferior do site do SynapseML. Se estiveres a usar Fabric, consulta o Guia de Instalação. Este passo requer colar uma célula extra no topo do caderno que importou.
4. Conecte o seu notebook a um cluster e siga o processo, editando e executando as células.

Preencha as informações do serviço

Em seguida, edite a célula no caderno para apontar para o seu serviço. Defina as service_namevariáveis , deployment_name, location, e key para corresponder ao seu serviço OpenAI:

import os
from pyspark.sql import SparkSession
from synapse.ml.core.platform import running_on_synapse, find_secret

# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()

if running_on_synapse():
    from notebookutils.visualization import display

# Fill in the following lines with your service information
# Learn more about selecting which embedding model to choose: https://openai.com/blog/new-and-improved-embedding-model
service_name = "synapseml-openai"
deployment_name = "gpt-4.1-mini"
deployment_name_embeddings = "text-embedding-3-small"

key = find_secret(
    "openai-api-key"
)  # please replace this line with your key as a string

assert key is not None and service_name is not None

Criar um conjunto de dados de prompts

Em seguida, crie um dataframe que consiste em uma série de linhas, com um prompt por linha.

Você também pode carregar dados diretamente do ADLS ou de outros bancos de dados. Para obter mais informações sobre como carregar e preparar dataframes do Spark, consulte o guia de carregamento de dados do Apache Spark.

df = spark.createDataFrame(
    [
        ("Hello my name is",),
        ("The best code is code that's",),
        ("SynapseML is ",),
    ]
).toDF("prompt")

Criar o cliente OpenAIPrompt Apache Spark

Para aplicar o serviço Azure OpenAI ao seu dataframe, crie um objeto OpenAIPrompt, que atua como um cliente distribuído. Defina os parâmetros do serviço com um único valor ou uma coluna de dataframe usando os ajustadores apropriados no OpenAIPrompt objeto. Neste exemplo, definido maxTokens para 200. Um token tem cerca de quatro caracteres, e este limite aplica-se à soma do prompt e do resultado. Defina o parâmetro promptCol com o nome da coluna de solicitação no dataframe.

from synapse.ml.services.openai import OpenAIPrompt

completion = (
    OpenAIPrompt()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMaxTokens(200)
    .setPromptCol("prompt")
    .setErrorCol("error")
    .setOutputCol("completions")
)

Transforme o dataframe usando o cliente OpenAIPrompt

Depois de criar o dataframe e o cliente de prompts, transforme o seu conjunto de dados de entrada e adicione uma coluna nomeada completions com toda a informação que o serviço acrescenta. Selecione apenas o texto para simplificar.

from pyspark.sql.functions import col

completed_df = completion.transform(df).cache()
display(
    completed_df.select(
        col("prompt"),
        col("error"),
        col("completions.choices.text").getItem(0).alias("text"),
    )
)

A sua saída deverá ser algo assim. O texto de conclusão é diferente do exemplo.

Mais exemplos de uso

Geração de embeddings de texto

Para além de completar texto, também pode incorporar texto para uso em algoritmos posteriores ou arquiteturas de recuperação vetorial. Ao criar embeddings, pode pesquisar e recuperar documentos de grandes coleções. Utilize esta abordagem quando a engenharia de prompts não seja suficiente para a tarefa. Para mais informações sobre a utilização de OpenAIEmbedding, consulte o guia de integração.

from synapse.ml.services.openai import OpenAIEmbedding

embedding = (
    OpenAIEmbedding()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name_embeddings)
    .setCustomServiceName(service_name)
    .setTextCol("prompt")
    .setErrorCol("error")
    .setOutputCol("embeddings")
)

display(embedding.transform(df))

Conclusão do chat

Modelos como GPT-4o e GPT-4.1 compreendem conversas em vez de prompts únicos. O OpenAIChatCompletion transformador expõe esta funcionalidade em larga escala.

from synapse.ml.services.openai import OpenAIChatCompletion
from pyspark.sql import Row
from pyspark.sql.types import *


def make_message(role, content):
    return Row(role=role, content=content, name=role)


chat_df = spark.createDataFrame(
    [
        (
            [
                make_message(
                    "system", "You are an AI chatbot with red as your favorite color"
                ),
                make_message("user", "What's your favorite color"),
            ],
        ),
        (
            [
                make_message("system", "You are very excited"),
                make_message("user", "How are you today"),
            ],
        ),
    ]
).toDF("messages")


chat_completion = (
    OpenAIChatCompletion()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMessagesCol("messages")
    .setErrorCol("error")
    .setOutputCol("chat_completions")
)

display(
    chat_completion.transform(chat_df).select(
        "messages", "chat_completions.choices.message.content"
    )
)

Melhore a taxa de transferência com o processamento em lote de solicitações

O exemplo faz várias solicitações ao serviço, uma para cada prompt. Para concluir vários prompts em uma única solicitação, use o modo de lote. Primeiro, no objeto OpenAIPrompt, em vez de definir a coluna Prompt para "Prompt", defina "batchPrompt" para a coluna BatchPrompt. Para fazer isso, crie um dataframe com uma lista de prompts por linha.

batch_df = spark.createDataFrame(
    [
        (["The time has come", "Pleased to", "Today stocks", "Here's to"],),
        (["The only thing", "Ask not what", "Every litter", "I am"],),
    ]
).toDF("batchPrompt")

De seguida, cria o OpenAIPrompt objeto. Em vez de configurar a coluna prompt, defina a coluna batchPrompt caso a sua coluna seja do tipo Array[String].

batch_completion = (
    OpenAIPrompt()
    .setSubscriptionKey(key)
    .setDeploymentName(deployment_name)
    .setCustomServiceName(service_name)
    .setMaxTokens(200)
    .setBatchPromptCol("batchPrompt")
    .setErrorCol("error")
    .setOutputCol("completions")
)

Na chamada para transformação, é feito um pedido por linha. Como cada linha contém múltiplos prompts, cada pedido envia todos os prompts dessa linha. Os resultados contêm uma linha para cada linha na solicitação.

completed_batch_df = batch_completion.transform(batch_df).cache()
display(completed_batch_df)

Usando um minibatcher automático

Se os seus dados estiverem em formato coluna, pode transpor-os para formato de linha usando o FixedMiniBatcherTransformerSynapseML .

from pyspark.sql.types import StringType
from synapse.ml.stages import FixedMiniBatchTransformer
from synapse.ml.core.spark import FluentAPI

completed_autobatch_df = (
    df.coalesce(
        1
    )  # Force a single partition so that our little 4-row dataframe makes a batch of size 4, you can remove this step for large datasets
    .mlTransform(FixedMiniBatchTransformer(batchSize=4))
    .withColumnRenamed("prompt", "batchPrompt")
    .mlTransform(batch_completion)
)

display(completed_autobatch_df)

Engenharia imediata para tradução

O serviço Azure OpenAI pode resolver muitas tarefas diferentes de linguagem natural através de prompt engineering. Este exemplo mostra a solicitação para tradução.

translate_df = spark.createDataFrame(
    [
        ("Japanese: Ookina hako \nEnglish: Big box \nJapanese: Midori tako\nEnglish:",),
        (
            "French: Quel heure et il au Montreal? \nEnglish: What time is it in Montreal? \nFrench: Ou est le poulet? \nEnglish:",
        ),
    ]
).toDF("prompt")

display(completion.transform(translate_df))

Solicitar resposta a perguntas

Este exemplo apresenta o modelo para responder a perguntas de conhecimento geral:

qa_df = spark.createDataFrame(
    [
        (
            "Q: Where is the Grand Canyon?\nA: The Grand Canyon is in Arizona.\n\nQ: What is the weight of the Burj Khalifa in kilograms?\nA:",
        )
    ]
).toDF("prompt")

display(completion.transform(qa_df))

Conteúdos relacionados

• Como construir um mecanismo de busca com SynapseML
• Como usar SynapseML e Foundry Tools para deteção de anomalias multivariadas - Analisar séries temporais
• Como usar o Kernel SHAP para explicar um modelo de classificação tabular