Use rotas personalizadas arbitrárias numa única implementação

Na publicação de LLMs moderna, os servidores de modelos implementam e suportam várias rotas de inferência para diferentes fins. Para esses exemplos de utilização, a Vertex Inference recomenda que use o método invoke para aceder a várias rotas numa única implementação.

O método invoke pode ser ativado quando faz o upload de um Model definindo invokeRoutePrefix como "/*". Depois de o modelo ser implementado num ponto final, qualquer rota não raiz no servidor do modelo fica acessível com a chamada HTTP de invocação. Por exemplo, "/invoke/foo/bar" seria encaminhado como "/foo/bar" para o servidor do modelo.

Esta funcionalidade está na fase de pré-visualização pública e tem as seguintes restrições:

  • Os modelos ativados para invocação só podem ser implementados num ponto final dedicado.
  • Apenas são suportadas chamadas HTTP para modelos ativados por invocação, e o RPC não é suportado.
  • Quando carregar um modelo, só é possível definir um dos seguintes elementos: predictRoute ou invokeRoutePrefix. O valor predefinido é predictRoute. Se o campo invokeRoutePrefix estiver definido num modelo, todas as outras rotas do Vertex, exceto invoke (por exemplo, :predict, :rawPredict…), são desativadas após a implementação.
  • "/*" é o único valor permitido para invokeRoutePrefix e expõe todo o caminho não raiz. Recomendamos um manuseamento cauteloso para rotas que não quer expor.

Carregar um modelo ativado por invocação

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

Implementar um modelo com a função invoke ativada

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

Fazer um pedido de inferência numa rota personalizada arbitrária

A rota de invocação permite o acesso a todos os caminhos de pedidos não raiz na implementação. Por exemplo, /invoke/foo/bar vai ser encaminhado como /foo/bar para o servidor do modelo. Existem duas formas de aceder ao trajeto.

Encaminhe o pedido personalizado para um ponto final dedicado

O pedido de invocação a um ponto final dedicado é encaminhado para um dos modelos implementados 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)

Encaminhe o pedido personalizado para um modelo implementado

O pedido de invocação pode ser feito para segmentar um modelo implementado específico. 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)