에만 적용됩니다: 
Foundry(클래식) 포털. 이 문서는 새 Foundry 포털에서 사용할 수 없습니다.
새 포털에 대해 자세히 알아봅니다.
비고
이 문서의 링크는 현재 보고 있는 Foundry(클래식) 설명서 대신 새 Microsoft Foundry 설명서의 콘텐츠를 열 수 있습니다.
Azure OpenAI Batch 입력 및 출력 파일에는 Azure Blob Storage를 사용합니다. 사용자 고유의 스토리지를 사용하여 입력 파일 수에 대한 Batch API 제한을 제거합니다. 이 문서에서는 스토리지 액세스를 한 번 구성한 다음, Azure 포털로 돌아가지 않고 일괄 처리 작업을 만들고, 모니터링하고, 다운로드하는 방법을 보여 줍니다.
필수 조건
- Azure Blob Storage 계정
-
배치 배포를 지원하는 지역에 있는 Azure OpenAI 리소스로, 모델이
Global-Batch또는DataZoneBatch로 배포된 리소스. 배포 지침은 리소스 만들기 및 모델 배포를 참조하세요. - Azure OpenAI 리소스를 업데이트하고 스토리지 계정에 Azure 역할을 할당할 수 있는 계정입니다.
- Azure OpenAI 리소스에 대한 Cognitive Services OpenAI 사용자 또는 Cognitive Services OpenAI 기여자 역할입니다.
- Python 3.9 이상.
- Azure CLI입니다.
워크플로 이해
스토리지 액세스 설정을 한 번 완료합니다. Azure OpenAI 리소스는 시스템 할당 관리 ID를 사용하여 입력 Blob을 읽고 결과 Blob을 작성합니다. 사용자 ID는 입력 파일을 만들고, 업로드하고, 일괄 작업을 제출하고, 작업을 모니터링하고, 결과를 다운로드합니다.
Azure CLI 또는 Azure 포털을 사용하여 다음 섹션의 모든 단계를 완료합니다. 하나의 설정 경로를 완료한 후 Python 패키지 설치를 계속합니다.
스토리지 액세스 설정
Azure OpenAI 리소스에 대해 시스템 할당 관리 ID를 사용하도록 설정합니다. 그런 다음 스토리지 계정 범위에서 리소스 ID와 사용자 ID 모두에 Storage Blob 데이터 기여자 역할을 할당합니다.
비고
Blob Storage Azure OpenAI Batch는 현재 사용자 할당 관리 ID를 지원하지 않습니다.
로그인하여 Azure 다음 명령에서 사용하는 값을 설정합니다.
az login
RESOURCE_GROUP="<resource-group-name>"
AZURE_OPENAI_RESOURCE="<azure-openai-resource-name>"
STORAGE_ACCOUNT="<storage-account-name>"
시스템 할당 관리 ID를 사용하도록 설정하고 해당 주체 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 계정 ID 할당
리소스 ID 및 로그인한 사용자에게 Storage 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.
역할 할당은 전파하는 데 몇 분이 걸릴 수 있습니다. 할당이 적용된 후 입력 및 출력 컨테이너를 만듭니다.
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.
Python 패키지 설치
DefaultAzureCredential에 대한 로컬 자격 증명을 제공하려면 Azure에 로그인합니다.
az login
그런 다음 OpenAI, Azure ID 및 Azure Blob Storage 클라이언트 라이브러리를 설치합니다.
python -m pip install --upgrade openai azure-identity azure-storage-blob
입력 파일 만들기
채팅 완료 요청 3개로 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 작업이 실패합니다.
입력 파일 업로드
사용자 ID를 사용하여 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
일괄 처리 작업 제출
입력 Blob을 제출하고 출력 컨테이너를 지정합니다. Azure OpenAI 리소스 및 스토리지 계정 자리 표시자를 값으로 바꿉니다.
비고
metadata는 현재 Blob Storage 통합에서 지원되지 않습니다.
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를 저장합니다.
일괄 처리 작업 모니터링
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
결과 다운로드
완료된 일괄 처리 응답에서 출력 Blob URL을 복사합니다. 사용자 ID를 사용하여 결과를 다운로드합니다.
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 생성되지 않습니다.