使用 Anthropic Claude 模型進行批次預測

您可以透過批次預測，將多個對延遲時間不敏感的提示傳送至 Anthropic Claude 模型。與線上預測相比，您可以在單一要求中批次處理大量輸入提示，而線上預測則是在每個要求中傳送一個輸入提示。

支援的 Anthropic Claude 模型

Vertex AI 支援下列 Anthropic Claude 模型的批次預測：

配額

根據預設，您在單一專案中可發出的並行批次要求數量為 4。

準備輸入內容

開始前，請在 BigQuery 資料表或 Cloud Storage 的 JSONL 檔案中準備輸入資料集。這兩種來源的輸入內容都必須採用 Anthropic Claude API 結構定義 JSON 格式，如下列範例所示：

{
  "custom_id": "request-1",
  "request":  {
    "messages": [{"role": "user", "content": "Hello!"}],
    "anthropic_version": "vertex-2023-10-16",
    "max_tokens": 50
  }
}

BigQuery

BigQuery 輸入資料表必須符合下列結構定義：

資料欄名稱 說明
custom_id 每個要求的 ID，用於將輸入內容與輸出內容配對。
要求 要求主體，也就是輸入提示，且必須遵循 Anthropic Claude API 架構
  • 輸入表格可以有其他資料欄，但批次工作會忽略這些資料欄。
  • 批次預測工作會保留兩個資料欄名稱，用於批次預測輸出：response(JSON) 和 status。請勿在輸入表格中使用這些資料欄。

Cloud Storage

如果是 Cloud Storage，輸入檔案必須是位於 Cloud Storage bucket 中的 JSONL 檔案。

要求批次預測

使用 BigQuery 或 Cloud Storage 的輸入內容，對 Claude 模型進行批次預測。您可以選擇將預測結果輸出至 BigQuery 資料表，或 Cloud Storage 儲存空間中的 JSONL 檔案。

BigQuery

指定 BigQuery 輸入資料表、模型和輸出位置。 批次預測工作和資料表必須位於相同區域。

Python

如要瞭解如何安裝或更新 Python 適用的 Vertex AI SDK，請參閱「安裝 Python 適用的 Vertex AI SDK」。 詳情請參閱 Python API 參考說明文件。

import time

from google import genai
from google.genai.types import CreateBatchJobConfig, JobState, HttpOptions

client = genai.Client(http_options=HttpOptions(api_version="v1"))

# TODO(developer): Update and un-comment below line
# output_uri = f"bq://your-project.your_dataset.your_table"

job = client.batches.create(
    # Check Anthropic Claude region availability in https://cloud.devsite.corp.google.com/vertex-ai/generative-ai/docs/partner-models/use-claude#regions
    # More about Anthropic model: https://console.cloud.google.com/vertex-ai/publishers/anthropic/model-garden/claude-3-5-haiku
    model="publishers/anthropic/models/claude-3-5-haiku",
    # The source dataset needs to be created specifically in us-east5
    src="bq://python-docs-samples-tests.anthropic_bq_sample.test_data",
    config=CreateBatchJobConfig(dest=output_uri),
)
print(f"Job name: {job.name}")
print(f"Job state: {job.state}")
# Example response:
# Job name: projects/%PROJECT_ID%/locations/us-central1/batchPredictionJobs/9876453210000000000
# Job state: JOB_STATE_PENDING

# See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.types.BatchJob
completed_states = {
    JobState.JOB_STATE_SUCCEEDED,
    JobState.JOB_STATE_FAILED,
    JobState.JOB_STATE_CANCELLED,
    JobState.JOB_STATE_PAUSED,
}

while job.state not in completed_states:
    time.sleep(30)
    job = client.batches.get(name=job.name)
    print(f"Job state: {job.state}")
# Example response:
# Job state: JOB_STATE_PENDING
# Job state: JOB_STATE_RUNNING
# Job state: JOB_STATE_RUNNING
# ...
# Job state: JOB_STATE_SUCCEEDED

REST

使用任何要求資料之前，請先替換以下項目：

  • LOCATION：支援所選 Anthropic Claude 模型的區域 (請參閱「Claude 區域」)。
  • PROJECT_ID：您的專案 ID。
  • MODEL：模型的名稱。
  • INPUT_URI：批次預測輸入內容所在的 BigQuery 資料表，例如 bq://myproject.mydataset.input_table。
  • OUTPUT_FORMAT：如要輸出至 BigQuery 資料表，請指定 bigquery。如要輸出至 Cloud Storage 值區，請指定 jsonl。
  • DESTINATION：如果是 BigQuery，請指定 bigqueryDestination。如果是 Cloud Storage，請指定 gcsDestination。
  • OUTPUT_URI_FIELD_NAME： 如果是 BigQuery，請指定 outputUri。如果是 Cloud Storage，請指定 outputUriPrefix。
  • OUTPUT_URI：如果是 BigQuery，請指定資料表位置，例如 bq://myproject.mydataset.output_result。如果是 Cloud Storage，請指定值區和資料夾位置，例如 gs://mybucket/path/to/outputfile。

HTTP 方法和網址：

POST https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs

JSON 要求主體：

'{
  "displayName": "JOB_NAME",
  "model": "publishers/anthropic/models/MODEL",
  "inputConfig": {
    "instancesFormat":"bigquery",
    "bigquerySource":{
      "inputUri" : "INPUT_URI"
    }
  },
  "outputConfig": {
    "predictionsFormat":"OUTPUT_FORMAT",
    "DESTINATION":{
      "OUTPUT_URI_FIELD_NAME": "OUTPUT_URI"
    }
  }
}'

如要傳送要求，請選擇以下其中一個選項：

curl

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs"

PowerShell

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs" | Select-Object -Expand Content

您應該會收到類似如下的 JSON 回應。

Cloud Storage

指定 JSONL 檔案的 Cloud Storage 位置、模型和輸出位置。

Python

如要瞭解如何安裝或更新 Python 適用的 Vertex AI SDK，請參閱「安裝 Python 適用的 Vertex AI SDK」。 詳情請參閱 Python API 參考說明文件。

import time

from google import genai
from google.genai.types import CreateBatchJobConfig, JobState, HttpOptions

client = genai.Client(http_options=HttpOptions(api_version="v1"))
# TODO(developer): Update and un-comment below line
# output_uri = "gs://your-bucket/your-prefix"

# See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.batches.Batches.create
job = client.batches.create(
    # More about Anthropic model: https://console.cloud.google.com/vertex-ai/publishers/anthropic/model-garden/claude-3-5-haiku
    model="publishers/anthropic/models/claude-3-5-haiku",
    # Source link: https://storage.cloud.google.com/cloud-samples-data/batch/anthropic-test-data-gcs.jsonl
    src="gs://cloud-samples-data/anthropic-test-data-gcs.jsonl",
    config=CreateBatchJobConfig(dest=output_uri),
)
print(f"Job name: {job.name}")
print(f"Job state: {job.state}")
# Example response:
# Job name: projects/%PROJECT_ID%/locations/us-central1/batchPredictionJobs/9876453210000000000
# Job state: JOB_STATE_PENDING

# See the documentation: https://googleapis.github.io/python-genai/genai.html#genai.types.BatchJob
completed_states = {
    JobState.JOB_STATE_SUCCEEDED,
    JobState.JOB_STATE_FAILED,
    JobState.JOB_STATE_CANCELLED,
    JobState.JOB_STATE_PAUSED,
}

while job.state not in completed_states:
    time.sleep(30)
    job = client.batches.get(name=job.name)
    print(f"Job state: {job.state}")
# Example response:
# Job state: JOB_STATE_PENDING
# Job state: JOB_STATE_RUNNING
# Job state: JOB_STATE_RUNNING
# ...
# Job state: JOB_STATE_SUCCEEDED

REST

使用任何要求資料之前，請先替換以下項目：

  • LOCATION：支援所選 Anthropic Claude 模型的區域 (請參閱「Claude 區域」)。
  • PROJECT_ID：您的專案 ID。
  • MODEL：模型的名稱。
  • INPUT_URIS：以逗號分隔的 JSONL 批次預測輸入內容 Cloud Storage 位置清單，例如 gs://bucketname/path/to/jsonl。
  • OUTPUT_FORMAT：如要輸出至 BigQuery 資料表，請指定 bigquery。如要輸出至 Cloud Storage 值區，請指定 jsonl。
  • DESTINATION：如果是 BigQuery，請指定 bigqueryDestination。如果是 Cloud Storage，請指定 gcsDestination。
  • OUTPUT_URI_FIELD_NAME： 如果是 BigQuery，請指定 outputUri。如果是 Cloud Storage，請指定 outputUriPrefix。
  • OUTPUT_URI：如果是 BigQuery，請指定資料表位置，例如 bq://myproject.mydataset.output_result。如果是 Cloud Storage，請指定值區和資料夾位置，例如 gs://mybucket/path/to/outputfile。

HTTP 方法和網址：

POST https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs

JSON 要求主體：

'{
  "displayName": "JOB_NAME",
  "model": "publishers/anthropic/models/MODEL",
  "inputConfig": {
    "instancesFormat":"jsonl",
    "gcsSource":{
      "uris" : "INPUT_URIS"
    }
  },
  "outputConfig": {
    "predictionsFormat":"OUTPUT_FORMAT",
    "DESTINATION":{
      "OUTPUT_URI_FIELD_NAME": "OUTPUT_URI"
    }
  }
}'

如要傳送要求，請選擇以下其中一個選項：

curl

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs"

PowerShell

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs" | Select-Object -Expand Content

您應該會收到類似如下的 JSON 回應。

取得批次預測工作狀態

取得批次預測工作狀態，確認工作是否已順利完成。

REST

使用任何要求資料之前，請先替換以下項目：

  • PROJECT_ID：您的專案 ID。
  • LOCATION：批次工作所在的區域。
  • JOB_ID：建立工作時傳回的批次工作 ID。

HTTP 方法和網址：

GET https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/JOB_ID

如要傳送要求，請選擇以下其中一個選項：

curl

執行下列指令：

curl -X GET \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/JOB_ID"

PowerShell

執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method GET `
    -Headers $headers `
    -Uri "https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/batchPredictionJobs/JOB_ID" | Select-Object -Expand Content

您應該會收到類似如下的 JSON 回應。

擷取批次預測輸出內容

批次預測工作完成後，請從您指定的位置擷取輸出內容。如果是 BigQuery，輸出內容會顯示在目的地 BigQuery 資料表的 response(JSON) 欄中。如果是 Cloud Storage，輸出內容會儲存為 JSONL 檔案，並存放在輸出 Cloud Storage 位置。

所有資料列完成預測後，或 24 小時後 (以先到者為準)，您就能存取完整批次預測結果。