Azure OpenAI voor big data

U kunt de Azure OpenAI-service gebruiken om veel taken in natuurlijke taal op te lossen door de voltooiings-API te vragen. De Azure OpenAI-service kan worden geïntegreerd met de gedistribueerde machine learning-bibliotheek SynapseML om de werkstromen gemakkelijker te schalen van enkele voorbeelden naar grote gegevenssets. Door deze integratie te gebruiken, kunt u het gedistribueerde Apache Spark-computingframework gebruiken om miljoenen prompts te verwerken met de OpenAI-service. In deze zelfstudie ziet u hoe u grote taalmodellen op een gedistribueerde schaal kunt toepassen met behulp van Azure OpenAI en Microsoft Fabric.

Vereisten

De belangrijkste vereisten voor deze quickstart zijn een werkende Azure OpenAI-resource en een Apache Spark-cluster waarop SynapseML is geïnstalleerd.

• Een Microsoft Fabric-abonnement ophalen. Of meld u aan voor een gratis Microsoft Fabric proefversie.

• Meld u aan bij Microsoft Fabric.

• Schakel over naar Fabric met behulp van de ervaringsschakelaar aan de linkerkant van de startpagina.

  

• Ga naar de Data Science-workload in Microsoft Fabric.
• Maak een nieuw notitieblok.
• Een Azure OpenAI-resource- een resource maken

Deze handleiding als notitieblok importeren

De volgende stap bestaat uit het toevoegen van deze code aan uw Spark-cluster. U kunt een notebook maken in uw Spark-platform en de code naar dit notebook kopiëren om de demo uit te voeren.

1. Download deze demo als notitieboekje (selecteer Raw, sla het bestand op).
2. Importeren naar de Fabric-werkruimte.
3. Gebruik de installatiehandleiding om SynapseML op je cluster te installeren. Deze stap vereist het plakken van een extra cel boven aan het notitieblok dat u hebt geïmporteerd.
4. Verbind uw notebook met een cluster en volg de instructies, bewerk de cellen en voer ze uit.

Servicegegevens invullen

Bewerk vervolgens de cel in het notebook zodat deze naar uw service wijst. Stel de variabelen service_name, deployment_name, location en key in om overeen te komen met uw OpenAI-service:

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

Een dataset creëren van prompts

Maak vervolgens een DataFrame bestaande uit een reeks rijen, met één prompt per rij.

U kunt gegevens ook rechtstreeks vanuit ADLS of andere databases laden. Voor meer informatie over het laden en voorbereiden van Spark DataFrames, zie de Apache Spark data-laadgids.

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

De OpenAIPrompt Apache Spark-client maken

Om de Azure OpenAI-service toe te passen op je DataFrame, maak je een OpenAIPrompt object dat fungeert als een gedistribueerde client. Stel de serviceparameters in met ofwel één enkele waarde of een DataFrame-kolom door de juiste setters op het OpenAIPrompt object te gebruiken. Stel in dit voorbeeld maxTokens in op 200. Een token is ongeveer vier tekens en deze limiet is van toepassing op de som van de prompt en het resultaat. Stel de promptCol parameter in met de naam van de promptkolom in de 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")
)

Transformeer de DataFrame met behulp van de OpenAIPrompt-client

Na het aanmaken van de DataFrame en de promptclient transformeer je je invoerdataset en voeg je een kolom completions toe met alle informatie die de service toevoegt. Selecteer alleen de tekst om het eenvoudig te maken.

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

Uw uitvoer ziet er ongeveer als volgt uit. De voltooiingstekst verschilt van het voorbeeld.

Meer gebruiksvoorbeelden

Tekst insluitingen genereren

Naast het voltooien van tekst kunt u ook tekst insluiten voor gebruik in downstreamalgoritmen of vector ophalen architecturen. Door insluitingen te maken, kunt u documenten zoeken en ophalen uit grote verzamelingen. Gebruik deze benadering wanneer prompt engineering niet voldoende is voor de taak. Zie de OpenAIEmbedding voor meer informatie over het gebruik.

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

Chataanvulling

Modellen zoals GPT-4o en GPT-4.1 begrijpen chats in plaats van enkele prompts. De OpenAIChatCompletion transformator maakt deze functionaliteit op schaal beschikbaar.

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

Doorvoer verbeteren met batchverwerking van aanvragen

In het voorbeeld worden verschillende aanvragen naar de service verzonden, één voor elke prompt. Als u meerdere prompts in één aanvraag wilt voltooien, gebruikt u de batchmodus. Geef eerst in het OpenAIPrompt object, in plaats van de kolom Prompt in te stellen op 'Prompt', 'batchPrompt' op voor de kolom BatchPrompt. Maak een DataFrame met een lijst van prompts per rij.

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

Maak vervolgens het OpenAIPrompt object. In plaats van de promptkolom in te stellen, stelt u de kolom batchPrompt in als uw kolom van het type Array[String]is.

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

In de aanroep om te transformeren wordt een aanvraag per rij ingediend. Omdat elke rij meerdere prompts bevat, verzendt elke aanvraag alle prompts in die rij. De resultaten bevatten een rij voor elke rij in de aanvraag.

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

Een automatische minibatcher gebruiken

Als uw gegevens in een kolomformaat zijn, kunt u ze naar een rijformaat transponeren met behulp van 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)

Prompt engineering voor vertaling

De Azure OpenAI-service kan veel verschillende natuurlijke taaltaken oplossen via prompt engineering. In dit voorbeeld wordt gevraagd om taalomzetting:

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

Uitnodiging tot beantwoording van vragen

In dit voorbeeld wordt het model gevraagd om algemene kennisvragen te beantwoorden:

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

Gerelateerde inhoud

• Een zoekmachine bouwen met SynapseML
• SynapseML en Foundry Tools gebruiken voor multivariate anomaliedetectie - Tijdreeks analyseren
• Kernel SHAP gebruiken om een tabellair classificatiemodel uit te leggen