Azure OpenAI 適用於大數據

你可以使用 Azure OpenAI 服務，透過提示完成 API 來解決許多自然語言任務。 為了讓你更容易將提示工作流程從少數範例擴展到大量範例資料集，Azure OpenAI 服務整合了分散式機器學習函式庫 SynapseML。 透過此整合，您可以使用 Apache Spark 分散式運算框架，透過 OpenAI 服務處理數百萬個提示。 本教學展示了如何利用 Azure OpenAI 與 Microsoft Fabric 在分散式尺度應用大型語言模型。

必要條件

此快速入門的關鍵條件包括一個可運作的 Azure OpenAI 資源，以及安裝 SynapseML 的 Apache Spark 叢集。

• 訂閱Microsoft Fabric。 或者，註冊免費的Microsoft Fabric試用。

• 登入Microsoft Fabric。

• 使用首頁左下角的體驗切換器切換到 Fabric。

  

• 前往 Microsoft Fabric 中的資料科學工作負載。
• 建立新的筆記本。
• 一個Azure OpenAI 資源 - create a resource

匯入此指南作為筆記本

下一個步驟是將此程式碼新增至您的 Spark 叢集。 您可以在 Spark 平台中建立筆記本，然後將程式碼複製到此筆記本中，以執行示範。

1. 下載此示範為筆記本 （選擇 RAW，然後儲存檔案）。
2. 匯入到 Fabric 工作區。
3. 使用 安裝指南在叢集上安裝 SynapseML。 這步驟需要在你匯入的筆記本頂端多貼一個儲存格。
4. 將您的筆記本連線至叢集，然後依序編輯和執行儲存格。

填入服務資訊

接著，編輯筆記本中的儲存格以指向您的服務。 將 service_name、deployment_name、location 和 key 變數設置為與你的 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

建立提示的資料集

接著，建立一個由一系列列組成的資料框架，每列有一個提示。

您也可以直接從 ADLS 或其他資料庫載入資料。 欲了解更多關於載入與準備 Spark DataFrame 的資訊，請參閱 Apache Spark 資料載入指南。

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

建立 OpenAIPrompt Apache Spark 客戶端

要將 Azure OpenAI 服務套用到 DataFrame，請建立OpenAIPrompt一個物件，作為分散式用戶端。 使用 OpenAIPrompt 物件上的適當 setter 方法，將服務參數設為單一值或 DataFrame 欄。 在這個例子中，設 maxTokens 為 200。 一個代幣約有四個字元，這個限制適用於提示詞與結果的總和。 在 DataFrame 中設定 promptCol 提示欄名稱的參數。

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

透過使用 OpenAIPrompt 用戶端轉換資料框架

建立 DataFrame 和提示用戶端後，轉換輸入資料集，並新增 completions 一個欄位，名稱包含服務所新增的所有資訊。 僅選取文字以求簡單。

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

您的輸出看起來應該像這樣。 完成文字與範例不同。

更多使用方式範例

產生文字嵌入

除了完成文字外，你也可以嵌入文字，用於下游演算法或向量檢索架構。 透過建立嵌入，你可以從大型資料庫中搜尋和檢索文件。 當提示工程不足以應付任務時，請採用此方法。 欲了解更多使用 OpenAIEmbedding資訊，請參閱 嵌入指南。

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

對話完成

像 GPT-4o 和 GPT-4.1 這類模型理解的是聊天，而非單一提示。 OpenAIChatCompletion 轉換器會大規模公開此功能。

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

透過要求批次來改善吞吐量

該範例會對服務提出數個要求，每個提示各一個。 若要在單一要求中完成多個提示，請使用批次模式。 首先，在物件中 OpenAIPrompt ，不要將 Prompt 欄位設為「Prompt」，而是將 BatchPrompt 欄位指定為「batchPrompt」。 建立一個 DataFrame，每列列出一個提示清單。

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

接著，建立物件 OpenAIPrompt 。 如果您的資料行屬於 Array[String] 類型，請不要設定提示資料行，而是設定 batchPrompt 資料行。

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

在呼叫轉換時，每列發出一個請求。 由於每一列包含多個提示，每個請求都會傳送該列中的所有提示。 結果中包含要求的每一列資料。

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

使用自動的小型批次處理器

如果您的資料是欄格式，可以使用 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)

翻譯的提示工程技術

Azure OpenAI 服務能透過 prompt engineering 解決許多不同的自然語言任務。 以下範例展示了語言翻譯的提示：

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

問題解答的提示

此範例提示一般知識問答模型：

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

相關內容

• 如何使用 SynapseML 組建搜尋引擎
• 如何使用 SynapseML 與 Foundry 工具進行多變量異常偵測 - 分析時間序列
• 如何使用 Kernel SHAP 來說明表格式分類模型