Agentes de consulta implementados no Azure Databricks

A forma como consultas um agente depende do servidor de agentes que o serve. Encontre o seu agente na tabela seguinte e siga a secção correspondente. Para saber mais sobre servidores de agente, consulte Agent Server.

O seu agente Hospedado em Como consultar
Usos DurableAgentServer Runtime do agente nas aplicações Databricks API de invocação no endereço /api/invocations
Utiliza o MLflow AgentServer ou LongRunningAgentServer (legacy) Aplicativos Databricks cliente OpenAI do Databricks ou a API OpenAI Responses em /responses
Implementado no Model Serving (legado) Endpoint de serviço do modelo cliente OpenAI da Databricks, API REST, ai_query ou AI Playground

Os agentes alojados em aplicações Databricks requerem um token Databricks OAuth. Tokens de acesso pessoal não funcionam para aplicações Databricks. Para gerar tokens OAuth a partir de um script, um principal de serviço, outra aplicação ou um notebook, consulte Ligar a uma API Databricks usando autenticação de token.

Consultar um agente que use DurableAgentServer

Os agentes que usam DurableAgentServer servem a API de invocação. Os agentes que crias com a CLI Agent Bricks usam DurableAgentServer, e agentbricks deploy implementa-os no Agent Runtime como uma aplicação chamada agent-bricks-<name>. Cada pedido à API inicia uma execução do seu agente, chamada invocação.

Endpoint Description
POST /api/invocations Inicia uma invocação. Por predefinição, a solicitação fica em espera e devolve o resultado. Configurar stream para receber eventos à medida que acontecem, ou background para regressar imediatamente.
GET /api/invocations/<id> Retorna o estado de uma invocação e, após a sua conclusão, a sua saída.
GET /api/invocations/<id>/events?after=<event-id> Transmite os eventos armazenados que vêm depois de <event-id>. Usa este endpoint para te reconectares a um stream.

Corpo do pedido

O corpo do pedido para POST /api/invocations aceita os seguintes campos. O servidor rejeita pedidos que contenham outros campos.

Field Description
id Required. Um UUID que gera para cada invocação. O servidor trata o ID como uma chave de idempotência: reenviar o mesmo pedido com o mesmo ID devolve a invocação existente em vez de executar o agente novamente. Reutilizar um ID para um pedido diferente devolve um erro 409.
session_id A conversa a que a invocação pertence. As invocações que partilham um ID de sessão executam-se uma de cada vez, por ordem. Os agentes gerados a partir dos templates CLI requerem este campo.
input A entrada para o seu agente. Os agentes gerados a partir dos templates CLI aceitam uma lista de mensagens, ou um objeto com uma lista messages.
stream Definir para true para receber eventos como Server-Sent Events (SSE).
background Defina para true devolver imediatamente uma 202 resposta com um URL de estado, e depois consulte o resultado.

O handler do seu agente define a estrutura de input. Os agentes gerados a partir dos templates CLI leem os seguintes campos quando input é um objeto:

Field Description
messages A conversa é encaminhada para o agente.
actor A identidade cuja memória de longo prazo o agente lê e na qual escreve. Se não passares um actor, o agente usa o ID da sessão, por isso as memórias não se transferem para uma nova sessão. Define actor a partir do utilizador com sessão iniciada da tua aplicação, não do texto que o utilizador escreve.
model O modelo a usar para esta invocação, em vez do modelo definido no código do agente.
resume A resposta a um agente que fez uma pausa para intervenção humana, como a aprovação de uma chamada de ferramenta. Quando um agente faz uma pausa, a invocação status é interrupted. Envia resume numa nova invocação com a mesma session_id para continuar.

Para permitir que o agente se lembre do que aprendeu sobre um utilizador ao longo das sessões, passe o ID do utilizador como actor:

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "session_id": "support-case-123",
  "input": {
    "messages": [{ "role": "user", "content": "What does Databricks do?" }],
    "actor": "user-42"
  }
}

Agente Bricks

Para testar um agente implementado a partir do seu terminal, use agentbricks endpoint invoke. O comando pesquisa a aplicação e autentica-se através do seu perfil da CLI.

agentbricks --profile <profile> endpoint invoke agent-bricks-<name> \
  --path /api/invocations \
  --json "{\"id\":\"$(uuidgen)\",\"session_id\":\"$(uuidgen)\",\"input\":[{\"role\":\"user\",\"content\":\"Hello\"}]}"

Para transmitir em fluxo a resposta, adicione "stream":true ao corpo do JSON e passe --sse. Para testar o agente enquanto corre localmente com agentbricks dev, substitua o nome da aplicação por --url http://localhost:8000.

API REST

  1. Obtenha o URL da aplicação. O campo URL na saída é o URL base para a API de invocações.

    agentbricks --profile <profile> deployments get agent-bricks-<name>

  2. Obtenha um token OAuth para o seu perfil. A saída contém o token no campo access_token.

    databricks auth token --profile <profile>

  3. Envie um pedido:

    curl --request POST \
  --url <app-url>/api/invocations \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "session_id": "support-case-123",
    "input": [{ "role": "user", "content": "What does Databricks do?" }]
  }'

A resposta contém a invocação id, o seu status, e um campo output com o valor que o seu agente devolve. O servidor não requer um esquema de saída: o seu handler pode devolver qualquer valor serializável em JSON. Os agentes gerados a partir dos templates CLI retornam um objeto com os seguintes campos:

  • output: as mensagens que o agente produziu nesta invocação.
  • status: completed, ou interrupted se o agente fez uma pausa para intervenção humana.

Python

O exemplo seguinte usa o Databricks SDK para procurar a URL da aplicação e gerar um token OAuth, e depois chama a API de invocações. O WorkspaceClient deve usar autenticação OAuth.

import uuid

import requests
from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
app_url = w.apps.get("agent-bricks-<name>").url
session_id = str(uuid.uuid4())

response = requests.post(
    f"{app_url}/api/invocations",
    headers=w.config.authenticate(),
    json={
        "id": str(uuid.uuid4()),
        "session_id": session_id,
        "input": [{"role": "user", "content": "What does Databricks do?"}],
    },
)
response.raise_for_status()
print(response.json()["output"])

Para continuar a conversa, envie a próxima mensagem com a mesma session_id e uma nova id.

Transmitir, executa em segundo plano e reconecta

  • Stream: Definir "stream": true. A resposta é um fluxo SSE que inclui eventos run.started e run.failed (ou delta), mais os eventos que o seu agente emite, como eventos run.completed com texto transmitido. Cada evento tem um ID.
  • Executar em segundo plano: Definição "background": true. O servidor devolve uma resposta 202 com um status_url. Consulte GET /api/invocations/<id> até o status ser completed. Se também definires "stream": true, a resposta inclui um events_url de onde podes ler eventos.
  • Reconectar: Se um stream se desligar, chame GET /api/invocations/<id>/events?after=<event-id> com o ID do último evento que recebeu.

O seguinte exemplo em Python devolve uma resposta em streaming:

with requests.post(
    f"{app_url}/api/invocations",
    headers=w.config.authenticate(),
    json={
        "id": str(uuid.uuid4()),
        "session_id": session_id,
        "input": [{"role": "user", "content": "Summarize our last conversation."}],
        "stream": True,
    },
    stream=True,
) as response:
    response.raise_for_status()
    for line in response.iter_lines(decode_unicode=True):
        if line.startswith("data: "):
            print(line[len("data: "):])

Se implementares o agente com mais do que uma instância, envia o ID da sessão num cabeçalho X-Routing-Key para encaminhar todos os pedidos numa sessão para a mesma instância.

Consultar um agente que utilize o MLflow legado AgentServer

Utilize esta secção para agentes que implementar nas aplicações Databricks com o servidor de agentes legado: o MLflow AgentServer ou LongRunningAgentServer, com a interface ResponsesAgent. Estes agentes disponibilizam a API OpenAI Responses em /responses.

LongRunningAgentServer disponibiliza a mesma API, pelo que os seguintes exemplos também se aplicam a ela. Também suporta execuções em segundo plano: definir background para true no pedido, e depois recuperar a resposta com GET /responses/<response-id>?stream=true&starting_after=<sequence-number>, que transmite os eventos após esse número de sequência.

cliente OpenAI da Databricks

A Databricks recomenda o cliente Databricks OpenAI para estes agentes. Inclua o apps/ prefixo no nome do modelo.

from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI

input_msgs = [{"role": "user", "content": "What does Databricks do?"}]
app_name = "<agent-app-name>"

# The WorkspaceClient must use OAuth authentication.
w = WorkspaceClient()
client = DatabricksOpenAI(workspace_client=w)

# Non-streaming request
response = client.responses.create(model=f"apps/{app_name}", input=input_msgs)
print(response)

# Streaming request
streaming_response = client.responses.create(
    model=f"apps/{app_name}", input=input_msgs, stream=True
)
for chunk in streaming_response:
    print(chunk)

Para passar custom_inputs, use o extra_body parâmetro:

response = client.responses.create(
    model=f"apps/{app_name}",
    input=input_msgs,
    extra_body={"custom_inputs": {"id": 5}},
)

Para obter o ID de rastreio de uma requisição, inclua o cabeçalho x-mlflow-return-trace-id. Depois usa o MLflow get_trace para recuperar o traçado completo.

response = client.responses.create(
    model=f"apps/{app_name}",
    input=input_msgs,
    extra_headers={"x-mlflow-return-trace-id": "true"},
)
trace_id = response.metadata["trace_id"]
trace = client.get_trace(trace_id)

API REST

Envie solicitações para o /responses caminho da URL da aplicação. O corpo do pedido segue a API de Respostas do OpenAI, por isso pode usar qualquer cliente HTTP ou ferramenta que a suporte.

curl --request POST \
  --url <app-url>/responses \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "input": [{ "role": "user", "content": "hi" }],
    "stream": true
  }'

Para passar custom_inputs, adicione-os ao corpo do pedido:

curl --request POST \
  --url <app-url>/responses \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "input": [{ "role": "user", "content": "hi" }],
    "custom_inputs": { "id": 5 }
  }'

Para obter o ID de rastreio, inclua o cabeçalho x-mlflow-return-trace-id: true. O corpo da resposta inclui o ID de rastreio num campo metadata.trace_id. Para pedidos de streaming, o trace ID chega como um evento SSE separado (data: {"trace_id": "tr-..."}) perto do final do fluxo.

Consultar um agente legado no Model Serving

Use esta secção para agentes legados implementados nos endpoints do Model Serving. Pode autenticar com um token Databricks OAuth ou um token de acesso pessoal. Para mover estes agentes para as Apps Databricks, veja Migrar um agente de Model Serving para Databricks Apps.

cliente OpenAI da Databricks

Para agentes que usam a interface ResponsesAgent, chame a função responses.create com o nome do endpoint como modelo:

from databricks_openai import DatabricksOpenAI

input_msgs = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>"

client = DatabricksOpenAI()

# Non-streaming request. Calls predict.
response = client.responses.create(model=endpoint, input=input_msgs)
print(response)

# Streaming request. Calls predict_stream.
streaming_response = client.responses.create(model=endpoint, input=input_msgs, stream=True)
for chunk in streaming_response:
    print(chunk)

Para agentes que usam as interfaces legadas ChatAgent ou ChatModel, utilize o cliente de completação de chat:

from databricks.sdk import WorkspaceClient

messages = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>"

client = WorkspaceClient().serving_endpoints.get_open_ai_client()
response = client.chat.completions.create(model=endpoint, messages=messages)
print(response)

Com qualquer cliente, passe custom_inputs ou databricks_options através do extra_body parâmetro. Por exemplo, extra_body={"databricks_options": {"return_trace": True}} devolve o trace com a resposta.

API REST

Para agentes que utilizam a interface ResponsesAgent, envie um pedido para /serving-endpoints/responses com o nome do endpoint como modelo:

curl --request POST \
  --url https://<workspace-url>/serving-endpoints/responses \
  --header 'Authorization: Bearer <token>' \
  --header 'content-type: application/json' \
  --data '{
    "model": "<agent-endpoint-name>",
    "input": [{ "role": "user", "content": "hi" }],
    "stream": true
  }'

Para agentes que utilizam as interfaces ChatAgent ou ChatModel, envie um pedido para /serving-endpoints/chat/completions com uma lista messages em vez de input. Para passar custom_inputs ou databricks_options, adicione-os ao corpo do pedido. Também pode enviar pedidos para a URL do /serving-endpoints/<agent-endpoint-name>/invocations endpoint. Veja Ver modelos individuais num endpoint.

Parque Infantil AI

Para conversar com um agente no Model Serving sem escrever código, abra o AI Playground e selecione o endpoint de serviço do agente. Para passar custom_inputs ao agente a partir do AI Playground, consulte Fornecer custom_inputs no AI Playground e na aplicação Review.

SQL com

Use ai_query para consultar um agente sobre o Model Serving a partir de SQL. Ver ai_query função para sintaxe e parâmetros.

SELECT ai_query(
  "<agent-endpoint-name>", question
) FROM (VALUES ('what is MLflow?'), ('how does MLflow work?')) AS t(question);

Recursos adicionais

  • Última atualização a