Anthropic Claude モデルを使用したバッチ予測

バッチ予測を使用すると、レイテンシの影響を受けにくい複数のプロンプトを Anthropic Claude モデルに送信できます。リクエストごとに 1 つの入力プロンプトを送信するオンライン予測とは異なり、バッチ予測では 1 つのリクエストで大量の入力プロンプトをバッチ処理できます。

サポートされている Anthropic Claude モデル

Vertex AI は、次の Anthropic Claude モデルのバッチ予測をサポートしています。

割り当て

デフォルトでは、1 つのプロジェクトで実行できる同時バッチ リクエストの数は 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 の 2 つの列名を予約します。入力テーブルではこれらの列を使用しないでください。

Cloud Storage

Cloud Storage の場合、入力ファイルは Cloud Storage バケットにある JSONL ファイルである必要があります。

バッチ予測をリクエストする

BigQuery または Cloud Storage からの入力を使用して、Claude モデルに対してバッチ予測を行います。予測の出力先として、BigQuery テーブルまたは Cloud Storage バケット内の 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.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 メソッドと URL:

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 バケットに出力するには、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 メソッドと URL:

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 メソッドと URL:

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 の場合、出力は出力 Cloud Storage のロケーションに JSONL ファイルとして保存されます。

バッチ予測の結果全体には、すべての行が完了した後、または 24 時間後にアクセスできます。