Usar rotas personalizadas arbitrárias em uma única implantação

No serviço moderno de LLM, os servidores de modelo implementam e oferecem suporte a várias rotas de inferência para fins diferentes. Para esses casos de uso, a inferência da Vertex recomenda usar o método invoke para acessar várias rotas em uma única implantação.

O método invoke pode ser ativado ao fazer upload de um Model definindo invokeRoutePrefix como "/*". Depois que o modelo é implantado em um endpoint, qualquer rota não raiz no servidor de modelo fica acessível com a chamada HTTP de invocação. Por exemplo, "/invoke/foo/bar" seria encaminhado como "/foo/bar" ao servidor do modelo.

Esse recurso está em pré-lançamento público e tem as seguintes restrições:

  • Os modelos ativados para invocação só podem ser implantados em um endpoint dedicado.
  • Somente chamadas HTTP são compatíveis com modelos ativados por invocação, e RPC não é compatível.
  • Ao fazer upload de um modelo, apenas uma das opções predictRoute ou invokeRoutePrefix pode ser definida. O valor padrão é predictRoute. Se o campo invokeRoutePrefix for definido em um modelo, todas as outras rotas da Vertex, além de invoke (por exemplo, :predict, :rawPredict etc.), serão desativadas após a implantação.
  • "/*" é o único valor permitido para invokeRoutePrefix e expõe todos os caminhos não raiz. É recomendável ter cuidado ao lidar com rotas que você não quer expor.

Como fazer upload de um modelo com invocação ativada

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,
)

Como implantar um modelo com invocação ativada

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

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

Como fazer uma solicitação de inferência em uma rota personalizada arbitrária

A rota de invocação permite o acesso a todos os caminhos de solicitação não raiz na implantação. Por exemplo, /invoke/foo/bar será encaminhado como /foo/bar para o servidor do modelo. Há duas maneiras de acessar a rota.

Solicitação de rota personalizada para um endpoint dedicado

A solicitação de invocação para um endpoint dedicado será encaminhada para um dos modelos implantados com base na configuração de divisão de tráfego.

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)

Solicitação de rota personalizada para um modelo implantado

A solicitação de invocação pode ser feita para segmentar um modelo implantado específico. Isso pode ser útil para testes e depuração.

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)