在單一部署作業中使用任意自訂路徑

在現代 LLM 服務中，模型伺服器會實作並支援多個用於不同用途的推論路徑。在這些用途中，Vertex Inference 建議使用 invoke 方法，存取單一部署作業中的多個路徑。

上傳 Model 時，將 invokeRoutePrefix 設為 "/*"，即可啟用 invoke 方法。模型部署至端點後，模型伺服器上的任何非根路徑都可以透過叫用 HTTP 呼叫存取。例如，「/invoke/foo/bar」會轉送至模型伺服器，並顯示為「/foo/bar」。

這項功能目前處於公開測試階段，並有下列限制：

  • 已啟用叫用的模型只能部署至專屬端點。
  • 啟用呼叫功能的模型僅支援 HTTP 呼叫，不支援 RPC。
  • 上傳模型時，只能設定 predictRoute 或 invokeRoutePrefix 其中之一。預設值為 predictRoute。如果模型上設定了 invokeRoutePrefix 欄位，部署後除了 invoke 以外的所有其他 Vertex 路徑 (例如 :predict、:rawPredict 等) 都會停用。
  • "/*" 是 invokeRoutePrefix 的唯一允許值，可公開所有非根路徑。建議謹慎處理不想公開的路線。

上傳已啟用叫用的模型

from google.cloud import aiplatform

invoke_enabled_model = aiplatform.Model.upload(
    display_name="invoke-enabled-model",
    serving_container_image_uri=IMAGE_URI,
    serving_container_invoke_route_prefix="/*",
    serving_container_health_route=HEALTH_ROUTE,
    serving_container_environment_variables={"KEY": "VALUE"},
    serving_container_args=[],
    sync=True,
)

部署已啟用叫用功能的模型

dedicated_endpoint = aiplatform.Endpoint.create(
    display_name="dedicated-endpoint-for-invoke-enabled-model",
    dedicated_endpoint_enabled=True,
    sync=True,
)

dedicated_endpoint.deploy(
    model=model,
    traffic_percentage=100,
    machine_type=MACHINE_TYPE,
    accelerator_type=ACCELERATOR_TYPE,
    accelerator_count=1,
    max_replica_count=1,
)

在任意自訂路徑上發出推論要求

叫用路徑可存取部署作業中的所有非根要求路徑。 舉例來說，/invoke/foo/bar 會以 /foo/bar 形式轉送至模型伺服器。存取路徑的方式有兩種。

將自訂路徑要求傳送至專屬端點

系統會根據流量拆分設定，將對專屬端點的叫用要求轉送至其中一個已部署的模型。

def invoke_tabular_sample(
    project: str,
    location: str,
    endpoint_id: str,
    request_path: str,
    http_request_body: Dict[str, Any],
    stream: bool = False,
):
    aiplatform.init(project=project, location=location)

    dedicated_endpoint = aiplatform.Endpoint(endpoint_id)
    if stream:
        for chunk in dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            stream=True,
        ):
            print(chunk)
    else:
        response = dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        print(response)

將要求自訂路徑傳送至已部署的模型

您可以發出叫用要求，指定要使用的已部署模型。這項功能有助於測試和偵錯。

def invoke_direct_deployed_model_inference_tabular_sample(
    project: str,
    location: str,
    endpoint_id: str,
    request_path: str,
    http_request_body: Dict[str, Any],
    deployed_model_id: str,
    stream: bool = False,
):
    aiplatform.init(project=project, location=location)

    dedicated_endpoint = aiplatform.Endpoint(endpoint_id)
    if stream:
        for chunk in dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            deployed_model_id=deployed_model_id,
            stream=True,
        ):
            print(chunk)
    else:
        response = dedicated_endpoint.invoke(
            request_path=request_path,
            body=json.dumps(http_request_body).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            deployed_model_id=deployed_model_id,
        )
        print(response)