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

透過批次預測功能，您可以將多個對延遲不敏感的提示詞傳送至 Anthropic Claude 模型。相較於線上預測 (每次要求傳送一個輸入提示)，您可以將大量輸入提示批次處理成單一要求。

支援的 Anthropic Claude 模型

Gemini Enterprise Agent Platform 支援下列 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 bucket 中的 JSONL 檔案。

BigQuery

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

Python

如要瞭解如何安裝或更新 Vertex AI SDK for Python，請參閱「安裝 Vertex AI SDK for Python」。詳情請參閱 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.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 bucket，請指定 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，請指定 bucket 和資料夾位置，例如 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

如要瞭解如何安裝或更新 Vertex AI SDK for Python，請參閱「安裝 Vertex AI SDK for Python」。詳情請參閱 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：
  • MODEL：模型名稱。
  • INPUT_URIS：JSONL 批次預測輸入內容的 Cloud Storage 位置清單，以半形逗號分隔，例如 gs://bucketname/path/to/jsonl。
  • OUTPUT_FORMAT：如要輸出至 BigQuery 資料表，請指定 bigquery。如要輸出至 Cloud Storage bucket，請指定 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，請指定 bucket 和資料夾位置，例如 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：
  • 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 小時後 (以先到者為準)，您就能存取完整的批次預測結果。