Agentes de consulta implementados en Azure Databricks

La forma en que consultas a un agente depende del servidor de agentes que lo atienda. Encuentra a tu agente en la siguiente tabla y luego sigue la sección correspondiente. Para aprender sobre servidores agentes, consulta Agent Server.

Tu agente Alojado en Cómo consultar
Usa DurableAgentServer Agent Runtime en Databricks Apps API de invocación de /api/invocations
Utiliza el MLflow AgentServer o LongRunningAgentServer (legacy) Aplicaciones de Databricks Cliente OpenAI de Databricks o la API de OpenAI Responses en /responses
Desplegado en Model Serving (heredado) Modelo de puntos de conexión de servicio Cliente OpenAI de Databricks, API REST, ai_query, o AI Playground

Los agentes alojados en aplicaciones Databricks requieren un token Databricks OAuth. Los tokens de acceso personal no funcionan para las aplicaciones Databricks. Para generar tokens OAuth a partir de un script, un principal de servicio, otra aplicación o un cuaderno, consulte Conectar a una API Databricks usando autenticación de token.

Consulta a un agente que use DurableAgentServer

Los agentes que usan DurableAgentServer atienden la API de invocación. Los agentes que creas con la CLI de Agent Bricks usan DurableAgentServer, y agentbricks deploy los despliega en Agent Runtime como una aplicación llamada agent-bricks-<name>. Cada solicitud a la API inicia una ejecución de tu agente, llamada invocación.

Endpoint Description
POST /api/invocations Inicia una invocación. Por defecto, la solicitud queda en espera y devuelve el resultado. Configura stream para recibir los eventos a medida que ocurren o background para regresar inmediatamente.
GET /api/invocations/<id> Devuelve el estado de una invocación y, una vez completada, su salida.
GET /api/invocations/<id>/events?after=<event-id> Transmite los eventos almacenados que vienen después de <event-id>. Usa este endpoint para reconectarte a un stream.

Cuerpo de la solicitud

El cuerpo POST /api/invocations de solicitud acepta los siguientes campos. El servidor rechaza solicitudes que contienen otros campos.

Campo Description
id Required. Un UUID que generas para cada invocación. El servidor trata el ID como una clave de idempotencia: reenviar la misma solicitud con el mismo ID devuelve la invocación existente en lugar de ejecutar el agente de nuevo. Reutilizar un ID para una solicitud diferente devuelve un error 409.
session_id La conversación a la que pertenece la invocación. Las invocaciones que comparten un ID de sesión se ejecutan una a una, en orden. Los agentes generados a partir de las plantillas CLI requieren este campo.
input La entrada para tu agente. Los agentes generados a partir de las plantillas CLI aceptan una lista de mensajes o un objeto con una lista messages.
stream Configúralo en true para recibir eventos como eventos enviados por el servidor (SSE).
background Configúralo en true para devolver una respuesta 202 inmediatamente con una URL de estado y luego consulta el resultado.

El manejador de tu agente define la estructura de input. Los agentes generados a partir de las plantillas CLI leen los siguientes campos cuando input es un objeto:

Campo Description
messages La conversación pasa a enviarse al agente.
actor La identidad de cuya memoria a largo plazo el agente lee y en cuya memoria escribe. Si no pasas un actor, el agente usa el ID de sesión, por lo que los recuerdos no se trasladan a una nueva sesión. Configura actor a partir del usuario que ha iniciado sesión en tu aplicación, no a partir del texto que escribe el usuario.
model El modelo a usar para esta invocación, en lugar del modelo configurado en el código del agente.
resume La respuesta a un agente que se detuvo para recibir una intervención humana, como la aprobación de una llamada a una herramienta. Cuando un agente hace una pausa, el/la status de la invocación es interrupted. Envía resume en una nueva invocación con la misma session_id para continuar.

Para que el agente recuerde lo que aprendió sobre un usuario a lo largo de las sesiones, pasa el ID del usuario 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"
  }
}

Ladrillos del agente

Para probar un agente desplegado desde tu terminal, usa agentbricks endpoint invoke. El comando busca la app y se autentica con tu perfil 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 la respuesta, añade "stream":true al cuerpo del JSON y pasa --sse. Para probar el agente mientras se ejecuta localmente con agentbricks dev, reemplaza el nombre de la app por --url http://localhost:8000.

REST API

  1. Obtén la URL de la app. El campo URL en la salida es la URL base para la API de invocaciones.

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

  2. Consigue un token OAuth para tu perfil. La salida contiene el token en el campo access_token.

    databricks auth token --profile <profile>

  3. Enviar una solicitud:

    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?" }]
  }'

La respuesta contiene la invocación id, su status y un campo output con el valor que devuelve tu agente. El servidor no requiere un esquema de salida: tu manejador puede devolver cualquier valor serializable en JSON. Los agentes generados a partir de las plantillas CLI devuelven un objeto con los siguientes campos:

  • output: los mensajes que el agente produjo en esta invocación.
  • status: completed, o interrupted si el agente hacía una pausa para recibir la entrada humana.

Python

El siguiente ejemplo utiliza el SDK Databricks para buscar la URL de la app y generar un token OAuth, y luego llama a la API de invocaciones. WorkspaceClient debe usar autenticación 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 la conversación, envía el siguiente mensaje con el mismo session_id y un nuevo id.

Transmite, ejecuta en segundo plano y vuelve a conectar

  • Transmisión: Establecer "stream": true. La respuesta es un flujo SSE que incluye eventos run.started y run.failed (o delta), además de los eventos que emite tu agente, como eventos run.completed con texto transmitido. Cada evento tiene un ID.
  • Ejecutar en segundo plano: Establecer "background": true. El servidor devuelve una respuesta 202 con un status_url. Consultar periódicamente GET /api/invocations/<id> hasta que el estado sea completed. Si también configuras "stream": true, la respuesta incluye un events_url de donde puedes leer eventos.
  • Reconectar: Si una transmisión se desconecta, llama a GET /api/invocations/<id>/events?after=<event-id> con el ID del último evento que recibiste.

El siguiente ejemplo de Python devuelve una respuesta en 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: "):])

Si despliegas el agente con más de una instancia, envía el ID de sesión en una X-Routing-Key cabecera para enrutar todas las solicitudes de una sesión a la misma instancia.

Consulta un agente que utilice el MLflow heredado AgentServer

Utiliza esta sección para los agentes que despliegas en las aplicaciones de Databricks con el servidor de agentes heredado: MLflow AgentServer o LongRunningAgentServer, con la ResponsesAgent interfaz. Estos agentes dan servicio a la API OpenAI Responses en /responses.

LongRunningAgentServer utiliza la misma API, por lo que los siguientes ejemplos también se aplican a ella. También soporta ejecuciones en segundo plano: establece background en true en la petición y luego se recupera la respuesta con GET /responses/<response-id>?stream=true&starting_after=<sequence-number>, que transmite los eventos posteriores a ese número de secuencia.

Cliente OpenAI de Databricks

Databricks recomienda el cliente OpenAI de Databricks para estos agentes. Incluye el apps/ prefijo en el nombre del 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 pasar custom_inputs, utiliza el extra_body parámetro:

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

Para obtener el ID de traza de una solicitud, incluye el x-mlflow-return-trace-id encabezado. A continuación, utilice MLflow get_trace para recuperar la traza completa.

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)

REST API

Envía peticiones a la /responses ruta de la URL de la app. El cuerpo de la solicitud sigue la API de Respuestas de OpenAI, por lo que puedes usar cualquier cliente HTTP o herramienta que la soporte.

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 pasarlos a custom_inputs, añádelos al cuerpo de la solicitud:

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 obtener el ID de seguimiento, incluye la x-mlflow-return-trace-id: true cabecera. El cuerpo de respuesta incluye el ID de traza en un campo metadata.trace_id. Para solicitudes de streaming, el ID de traza llega como un evento SSE separado (data: {"trace_id": "tr-..."}) cerca del final del stream.

Consulta un agente heredado en Model Serving

Utiliza esta sección para agentes heredados desplegados en endpoints de Model Serving. Puedes autenticarte con un token OAuth de Databricks o un token de acceso personal. Para mover estos agentes a Databricks Apps, consulte Migrar un agente de Model Serving a Databricks Apps.

Cliente OpenAI de Databricks

Para los agentes que usan la interfaz ResponsesAgent, llama a responses.create con el nombre del 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 los agentes que usan las interfaces heredadas ChatAgent o ChatModel, utiliza el cliente de completación 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)

Con cualquiera de los dos clientes, pasa custom_inputs o databricks_options mediante el parámetro extra_body. Por ejemplo, extra_body={"databricks_options": {"return_trace": True}} devuelve la traza con la respuesta.

REST API

Para los agentes que usan la ResponsesAgent interfaz, envía una petición a /serving-endpoints/responses con el nombre del 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 los agentes que usan las interfaces ChatAgent o ChatModel, envíe una solicitud a /serving-endpoints/chat/completions con una lista messages en lugar de input. Para pasarlos custom_inputs o databricks_options, añádelos al cuerpo de la solicitud. También puedes enviar solicitudes a la URL del /serving-endpoints/<agent-endpoint-name>/invocations endpoint. Consulta modelos individuales detrás de un endpoint.

AI Playground

Para chatear con un agente en Model Serving sin escribir código, abre AI Playground y selecciona el endpoint de servicio del agente. Para pasar custom_inputs al agente desde AI Playground, consulta Proporcionar custom_inputs en la aplicación de AI Playground y la aplicación Review.

SQL con

Usa ai_query para consultar un agente en Model Serving desde SQL. Ver la ai_query function para la sintaxis y los parámetros.

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

Recursos adicionales

  • Se actualizó por última vez el