Azure OpenAI per Big Data

È possibile usare il servizio OpenAI Azure per risolvere molte attività in linguaggio naturale richiedendo l'API di completamento. Per semplificare la scalabilità dei flussi di lavoro richiesti da alcuni esempi a set di dati di grandi dimensioni, il servizio OpenAI Azure si integra con la libreria di Machine Learning distribuita SynapseML. Usando questa integrazione, è possibile usare il framework di elaborazione distribuito apache Spark per elaborare milioni di richieste con il servizio OpenAI. Questa esercitazione illustra come applicare modelli linguistici di grandi dimensioni su larga scala usando Azure OpenAI e Microsoft Fabric.

Prerequisiti

I prerequisiti principali per questa guida introduttiva includono una risorsa Azure OpenAI funzionante e un cluster Apache Spark con SynapseML installato.

• Ottenere una sottoscrizione Microsoft Fabric. In alternativa, iscriversi per ottenere una versione di valutazione gratuita Microsoft Fabric.

• Accedere a Microsoft Fabric.

• Passare a Fabric usando il commutatore dell'esperienza in basso a sinistra della tua home page.

  

• Passare all'esperienza di data science in Microsoft Fabric.
• Creare un nuovo notebook.
• Una risorsa OpenAI Azure - creare una risorsa

Importare questa guida come notebook

Il passaggio successivo consiste nell'aggiungere questo codice al cluster Spark. È possibile creare un notebook nella piattaforma Spark e copiare il codice in questo notebook per eseguire la demo. In alternativa, scaricare il notebook e importarlo in Synapse Analytics.

1. Scaricare questa demo come notebook, (selezionare Raw e poi salvare il file)
2. Importare il notebook nell'area di lavoro di Synapse o, se si usa Fabric, importarlo nell'area di lavoro di Fabric
3. Installare SynapseML nel cluster. Vedere le istruzioni di installazione per Synapse nella parte inferiore del sito Web SynapseML. Se si usa Fabric, vedere Guida all'installazione. Questo passaggio richiede di incollare una cella aggiuntiva nella parte superiore del notebook importato.
4. Collega il tuo notebook a un cluster e segui le istruzioni per modificare ed eseguire le celle.

Inserire le informazioni sul servizio

Successivamente, modifica la cella nel notebook in modo che punti al tuo servizio. Impostare le service_namevariabili , deployment_namelocation, e key in modo che corrispondano al servizio 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

Crea un insieme di dati di prompt

Creare quindi un dataframe costituito da una serie di righe, con un prompt per riga.

È anche possibile caricare i dati direttamente da ADLS o da altri database. Per ulteriori informazioni sul caricamento e la preparazione dei dataframe Spark, vedere la Guida al caricamento dei dati di Apache Spark.

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

Creare il client Apache Spark OpenAIPrompt

Per applicare il servizio OpenAI Azure al dataframe, creare un oggetto OpenAIPrompt, che funge da client distribuito. Impostare i parametri del servizio con un singolo valore o una colonna del dataframe usando i setter appropriati nell'oggetto OpenAIPrompt . In questo esempio, imposta maxTokens su 200. Un token è di circa quattro caratteri e questo limite si applica alla somma della richiesta e del risultato. Imposta il parametro promptCol con il nome della colonna del prompt nel 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")
)

Trasformare il dataframe usando il client OpenAIPrompt

Dopo aver creato il dataframe e il client del prompt, trasformare il set di dati di input e aggiungere una colonna denominata completions con tutte le informazioni aggiunte dal servizio. Selezionare solo il testo per semplicità.

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

L'output dovrebbe risultare simile a quanto segue. Il testo di completamento è diverso dall'esempio.

Altri esempi di utilizzo

Generazione di incorporamenti di testo

Oltre a completare il testo, è anche possibile incorporare testo da usare negli algoritmi downstream o nelle architetture di recupero vettoriale. Creando incorporamenti, è possibile cercare e recuperare documenti da raccolte di grandi dimensioni. Utilizzare questo approccio quando la progettazione di prompt non è sufficiente per il compito. Per altre informazioni sull'uso di OpenAIEmbedding, vedere la guida all'incorporamento.

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

Completamento della chat

I modelli come GPT-4o e GPT-4.1 comprendono le chat anziché le singole richieste. Il trasformatore OpenAIChatCompletion espone questa funzionalità su larga scala.

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

Migliorare la velocità effettiva con l'invio in batch delle richieste

Nell'esempio vengono eseguite diverse richieste al servizio, una per ogni prompt. Per completare più prompt in una singola richiesta, usare la modalità in batch. Prima di tutto, nell'oggetto OpenAIPrompt anziché impostare la colonna Prompt su "Prompt", specificare "batchPrompt" per la colonna BatchPrompt. A tale scopo, creare un dataframe con un elenco di prompt per riga.

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

Creare quindi l'oggetto OpenAIPrompt . Piuttosto che configurare la colonna prompt, configura la colonna batchPrompt se la tua colonna è di tipo Array[String].

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

Nella chiamata alla trasformazione viene effettuata una richiesta per riga. Poiché ogni riga contiene più richieste, ogni richiesta invia tutte le richieste in tale riga. I risultati contengono una riga per ogni riga nella richiesta.

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

Uso di un minibatcher automatico

Se i tuoi dati sono in formato di colonna, puoi trasporli in formato riga utilizzando SynapseML FixedMiniBatcherTransformer.

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)

Progettazione di prompt per la traduzione

Il servizio OpenAI Azure può risolvere molte diverse attività in linguaggio naturale tramite prompt engineering. Questo esempio mostra la richiesta di tradurre una lingua:

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

Prompt per rispondere alle domande

In questo esempio viene richiesto al modello di rispondere a domande di conoscenza generale.

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

Contenuto correlato

• Come creare un motore di ricerca con SynapseML
• Come usare Gli strumenti SynapseML e Foundry per il rilevamento delle anomalie multivariate - Analizzare le serie temporali
• Come usare Kernel SHAP per spiegare un modello di classificazione tabulare