为 Azure OpenAI Batch 配置Azure Blob 存储（经典版）

仅适用于：Foundry（经典版）门户。 本文不适用于新的 Foundry 门户。 详细了解新门户。

注释

本文中的链接可能会打开新 Microsoft Foundry 文档中的内容，而不是你现在正在查看的 Foundry （经典）文档。

将 Azure Blob 存储 用于 Azure OpenAI Batch 的输入和输出文件。 通过使用自己的存储，可以删除对输入文件数的 Batch API 限制。 本文介绍如何配置存储访问一次，然后创建、监视和下载批处理作业，而无需返回到Azure门户。

先决条件

• 一个 Azure Blob 存储帐户。
• 位于支持批量部署的区域中的 Azure OpenAI 资源，其中某个模型部署为Global-Batch或DataZoneBatch。 有关部署说明，请参阅 “创建资源并部署模型”。
• 一个能够更新 Azure OpenAI 资源并在存储帐户上分配 Azure 角色的帐户。
• Azure OpenAI 资源的认知服务 OpenAI 用户或认知服务 OpenAI 参与者角色。
• Python 3.9 或更高版本。
• Azure CLI。

了解工作流

完成存储访问设置一次。 Azure OpenAI 资源使用其系统分配的托管标识读取输入 Blob 并写入结果 Blob。 用户标识将创建输入文件、上传、提交批处理作业、监视作业并下载结果。

使用Azure CLI或Azure门户完成下一部分中的所有步骤。 完成一个安装路径后，请继续安装Python包。

设置存储访问权限

为 Azure OpenAI 资源启用系统分配的托管标识。 然后在存储帐户范围内，将 存储 Blob 数据参与者 角色分配给资源标识和你的用户标识。

注释

目前，与 Blob 存储 结合使用的 Azure OpenAI Batch 不支持用户分配托管标识。

Azure CLI

登录到Azure并设置以下命令使用的值：

az login

RESOURCE_GROUP="<resource-group-name>"
AZURE_OPENAI_RESOURCE="<azure-openai-resource-name>"
STORAGE_ACCOUNT="<storage-account-name>"

启用系统分配的托管标识并检索其主体 ID：

az cognitiveservices account identity assign \
  --name "$AZURE_OPENAI_RESOURCE" \
  --resource-group "$RESOURCE_GROUP" \
  --output none

OPENAI_PRINCIPAL_ID=$(az cognitiveservices account show \
  --name "$AZURE_OPENAI_RESOURCE" \
  --resource-group "$RESOURCE_GROUP" \
  --query identity.principalId \
  --output tsv)

echo "Managed identity: $OPENAI_PRINCIPAL_ID"

Managed identity: <principal-id>

参考： az cognitiveservices account identity assign

将存储 Blob 数据参与者角色分配给资源标识和当前登录的用户：

STORAGE_ACCOUNT="<storage-account-name>"
STORAGE_ID=$(az storage account show \
  --name "$STORAGE_ACCOUNT" \
  --query id \
  --output tsv)
USER_ID=$(az ad signed-in-user show --query id --output tsv)

az role assignment create \
  --assignee-object-id "$OPENAI_PRINCIPAL_ID" \
  --assignee-principal-type ServicePrincipal \
  --role "Storage Blob Data Contributor" \
  --scope "$STORAGE_ID" \
  --output none

az role assignment create \
  --assignee-object-id "$USER_ID" \
  --assignee-principal-type User \
  --role "Storage Blob Data Contributor" \
  --scope "$STORAGE_ID" \
  --output none

echo "Storage roles assigned."

Storage roles assigned.

参考：az role assignment create

角色分配可能需要几分钟才能生效。 在分配生效后创建输入和输出容器：

STORAGE_ACCOUNT="<storage-account-name>"

az storage container create \
  --account-name "$STORAGE_ACCOUNT" \
  --name batch-input \
  --auth-mode login \
  --output none

az storage container create \
  --account-name "$STORAGE_ACCOUNT" \
  --name batch-output \
  --auth-mode login \
  --output none

echo "Created batch-input and batch-output."

Created batch-input and batch-output.

参考： az storage container create

在继续Python工作流之前完成所有门户配置：

1. 登录到 Azure 门户。
2. 打开 Azure OpenAI 资源。
3. 在资源管理下，选择标识>系统分配。
4. 将 状态 设置为 “打开”，然后选择“ 保存”。
5. 打开存储帐户，然后选择访问控制（IAM）>添加>角色分配。
6. 在“ 角色 ”选项卡上，选择“ 存储 Blob 数据参与者”，然后选择“ 下一步”。
7. 在“成员”选项卡上，选择“托管身份”>选择成员。
8. 选择Azure OpenAI 资源，然后选择“查看 + 分配”。
9. 向你的用户帐户再添加一个存储 Blob 数据参与者角色分配。
10. 在存储帐户中，选择 “数据存储>容器>+ 容器”。
11. 创建名为 batch-input 和 batch-output 的容器。

如果你使用自定义角色而不是 Storage Blob Data Contributor，请授予以下权限：

• 输入数据块：Microsoft.Storage/storageAccounts/blobServices/containers/blobs/read
• 输出 Blob：Microsoft.Storage/storageAccounts/blobServices/containers/blobs/read 和 Microsoft.Storage/storageAccounts/blobServices/containers/blobs/write

安装 Python 包

登录到 Azure，为 DefaultAzureCredential 提供本地凭据：

az login

然后安装 OpenAI、Azure标识和Azure Blob 存储客户端库：

python -m pip install --upgrade openai azure-identity azure-storage-blob

创建输入文件

创建 test.jsonl，其中包含三个聊天自动补全请求。 将 BATCH_DEPLOYMENT 设置为你的批处理模型部署的名称：

import json

BATCH_DEPLOYMENT = "<batch-deployment-name>"
INPUT_FILE = "test.jsonl"
SYSTEM_MESSAGE = "You are an AI assistant that helps people find information."
PROMPTS = [
  "When was Microsoft founded?",
  "When was the first Xbox released?",
  "What is Altair BASIC?",
]

with open(INPUT_FILE, "w", encoding="utf-8") as batch_file:
  for index, prompt in enumerate(PROMPTS):
    request = {
      "custom_id": f"task-{index}",
      "method": "POST",
      "url": "/v1/chat/completions",
      "body": {
        "model": BATCH_DEPLOYMENT,
        "messages": [
          {"role": "system", "content": SYSTEM_MESSAGE},
          {"role": "user", "content": prompt},
        ],
      },
    }
    batch_file.write(json.dumps(request) + "\n")

print(f"Created {INPUT_FILE} with {len(PROMPTS)} requests.")

Created test.jsonl with 3 requests.

提交批处理作业后，请勿修改输入 Blob。 如果 blob 在作业运行期间发生更改，作业会因出现 input_modified 错误而失败。

上传输入文件

使用您的用户标识将 test.jsonl 上传到 batch-input 容器。 将 STORAGE_ACCOUNT 设置为你的存储帐户名称：

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobClient

STORAGE_ACCOUNT = "<storage-account-name>"
INPUT_CONTAINER = "batch-input"
INPUT_FILE = "test.jsonl"
ACCOUNT_URL = f"https://{STORAGE_ACCOUNT}.blob.core.windows.net"

credential = DefaultAzureCredential()
input_blob = BlobClient(
  account_url=ACCOUNT_URL,
  container_name=INPUT_CONTAINER,
  blob_name=INPUT_FILE,
  credential=credential,
)

with open(INPUT_FILE, "rb") as data:
  input_blob.upload_blob(data, overwrite=True)

print(f"Uploaded: {input_blob.url}")

Uploaded: https://<storage-account-name>.blob.core.windows.net/batch-input/test.jsonl

参考： BlobClient.upload_blob

提交批处理作业

提交输入 Blob 并指定输出容器。 将 Azure OpenAI 资源和存储帐户占位符替换为你的值。

注释

Blob 存储 集成目前不支持 metadata。

from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI
AZURE_OPENAI_BASE_URL = (
  "https://<azure-openai-resource-name>.openai.azure.com/openai/v1/"
)
STORAGE_ACCOUNT = "<storage-account-name>"
INPUT_BLOB_URL = (
  f"https://{STORAGE_ACCOUNT}.blob.core.windows.net/batch-input/test.jsonl"
)
OUTPUT_CONTAINER_URL = (
  f"https://{STORAGE_ACCOUNT}.blob.core.windows.net/batch-output"
)

token_provider = get_bearer_token_provider(
  DefaultAzureCredential(), "https://ai.azure.com/.default"
)
openai = OpenAI(base_url=AZURE_OPENAI_BASE_URL, api_key=token_provider)

batch = openai.batches.create(
  input_file_id=None,
  endpoint="/chat/completions",
  completion_window="24h",
  extra_body={
    "input_blob": INPUT_BLOB_URL,
    "output_folder": {"url": OUTPUT_CONTAINER_URL},
  },
)

print(f"Batch ID: {batch.id}")
print(f"Status: {batch.status}")

Batch ID: <batch-id>
Status: validating

保存该批处理 ID 以用于下一步。

参考：Azure OpenAI Batch

监控批处理作业

将 BATCH_ID 设置为提交作业时返回的标识符。 脚本每隔 60 秒检查一次状态，直到作业达到终端状态：

import time

from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI

AZURE_OPENAI_BASE_URL = (
  "https://<azure-openai-resource-name>.openai.azure.com/openai/v1/"
)
BATCH_ID = "<batch-id>"
TERMINAL_STATES = {"completed", "failed", "cancelled", "expired"}

token_provider = get_bearer_token_provider(
  DefaultAzureCredential(), "https://ai.azure.com/.default"
)
openai = OpenAI(base_url=AZURE_OPENAI_BASE_URL, api_key=token_provider)

batch = openai.batches.retrieve(BATCH_ID)
while batch.status not in TERMINAL_STATES:
  print(f"Batch {BATCH_ID}: {batch.status}")
  time.sleep(60)
  batch = openai.batches.retrieve(BATCH_ID)

print(f"Batch {BATCH_ID}: {batch.status}")
if batch.status == "completed":
  print(f"Output blob: {batch.output_blob}")
  print(f"Error blob: {batch.error_blob}")
elif batch.errors:
  for error in batch.errors.data:
    print(f"Error {error.code}: {error.message}")

Batch <batch-id>: validating
Batch <batch-id>: in_progress
Batch <batch-id>: finalizing
Batch <batch-id>: completed
Output blob: https://<storage-account-name>.blob.core.windows.net/batch-output/<batch-output-path>/results.jsonl
Error blob: https://<storage-account-name>.blob.core.windows.net/batch-output/<batch-output-path>/errors.jsonl

参考：Azure OpenAI Batch 状态值

下载结果

从已完成的批处理响应中复制输出 Blob URL。 使用用户标识下载结果：

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobClient

OUTPUT_BLOB_URL = (
  "https://<storage-account-name>.blob.core.windows.net/"
  "batch-output/<batch-output-path>/results.jsonl"
)
OUTPUT_FILE = "results.jsonl"

credential = DefaultAzureCredential()
output_blob = BlobClient.from_blob_url(
  blob_url=OUTPUT_BLOB_URL,
  credential=credential,
)

with open(OUTPUT_FILE, "wb") as results_file:
  results_file.write(output_blob.download_blob().readall())

print(f"Downloaded: {OUTPUT_FILE}")

Downloaded: results.jsonl

批处理响应始终包括 output_blob 和 error_blob URL。 仅当 Blob 有内容时，才会创建它。 例如，如果每个请求都成功， errors.jsonl 则不会创建。

参考： BlobClient.download_blob

相关内容

• 使用 Azure OpenAI Batch
• 将 Azure OpenAI Batch 与 REST API 配合使用
• 使用 Microsoft Entra ID向 Azure OpenAI 进行身份验证