Query agents deployed on Azure Databricks

Wie man einen Agenten abfragt, hängt vom Agentenserver ab, der ihn bedient. Finden Sie Ihren Agenten in der folgenden Tabelle und folgen Sie dann dem passenden Abschnitt. Um mehr über Agent-Server zu erfahren, siehe Agent Server.

Agenten, die auf Databricks Apps gehostet werden, benötigen ein Databricks OAuth-Token. Persönliche Zugriffstoken funktionieren nicht für Databricks-Apps. Um OAuth-Tokens aus einem Skript, einem Service Principal, einer anderen App oder einem Notizbuch zu generieren, siehe Verbinden mit einer API-Databricks-App mittels Token-Authentifizierung.

Abfragen eines Agents, der DurableAgentServer verwendet

Agents, die DurableAgentServer verwenden, stellen die Aufruf-API bereit. Agenten, die du mit der Agent Bricks CLI erstellst, verwenden DurableAgentServer, und agentbricks deploy stellt sie in Agent Runtime als App namens agent-bricks-<name> bereit. Jede Anforderung an die API startet eine Ausführung Ihres Agents, die als Aufruf bezeichnet wird.

Anforderungstext

Der Request-Body für POST /api/invocations akzeptiert die folgenden Felder. Der Server lehnt Anfragen ab, die andere Felder enthalten.

Der Handler Ihres Agenten definiert die Struktur von input. Agenten, die aus den CLI-Vorlagen generiert werden, lesen folgende Felder, wenn input ein Objekt ist:

Damit der Agent sich daran erinnert, was er über einen Benutzer über Sitzungen hinweg erfahren hat, geben Sie die Benutzer-ID als actor weiter:

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

Agent Bricks

Um einen eingesetzten Agenten von Ihrem Terminal aus zu testen, verwenden Sie agentbricks endpoint invoke. Der Befehl sucht die App und authentifiziert sich mit Ihrem CLI-Profil.

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

Um die Antwort zu streamen, fügen Sie "stream":true dem JSON-Körper hinzu und übergeben Sie --sse. Um den Agenten zu testen, während er lokal mit agentbricks dev läuft, ersetzen Sie den App-Namen durch --url http://localhost:8000.

REST API

1. Rufen Sie die URL der App ab. Das URL-Feld in der Ausgabe ist die Basis-URL für die Aufruf-API.

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

2. Hol dir ein OAuth-Token für dein Profil. Die Ausgabe enthält das Token im Feld access_token .

   databricks auth token --profile <profile>
   

3. Senden einer Anforderung:

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

Die Antwort enthält den Aufruf id, sein status, und ein output Feld mit dem Wert, den dein Agent zurückgibt. Der Server benötigt kein Ausgabeschema: Ihr Handler kann jeden JSON-serialisierbaren Wert zurückgeben. Agenten, die aus den CLI-Vorlagen generiert wurden, liefern ein Objekt mit folgenden Feldern zurück:

• output: die Nachrichten, die der Agent in diesem Aufruf erzeugt hat.
• status: completed, oder interrupted, wenn der Agent für menschliche Eingaben pausierte.

Python

Das folgende Beispiel verwendet das Databricks SDK, um die App-URL nachzuschlagen und ein OAuth-Token zu generieren und dann die Invocations-API aufzurufen. Das WorkspaceClient muss die OAuth-Authentifizierung verwenden.

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"])

Um das Gespräch fortzusetzen, senden Sie die nächste Nachricht mit dem gleichen session_id und einem neuen id.

Streamen, im Hintergrund ausführen und erneut verbinden

• Stream: Set "stream": true. Die Antwort ist ein SSE-Stream, der run.completed- und run.failed-Ereignisse (oder run.started-Ereignisse) sowie die Ereignisse enthält, die Ihr Agent aussendet, wie zum Beispiel delta-Ereignisse mit gestreamtem Text. Jedes Event hat eine ID.
• Im Hintergrund ausführen: Legen Sie "background": true fest. Der Server gibt eine 202 Antwort mit einem status_urlzurück. Rufen Sie GET /api/invocations/<id> ab, bis der Status completed lautet. Wenn du auch "stream": true setzt, enthält die Antwort ein events_url, aus dem du Ereignisse lesen kannst.
• Verbinden Sie sich wieder: Wenn ein Stream abbricht, rufen Sie GET /api/invocations/<id>/events?after=<event-id> mit der ID des zuletzt erhaltenen Ereignisses auf.

Das folgende Python-Beispiel streamt eine Antwort:

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: "):])

Wenn du den Agenten mit mehr als einer Instanz deployest, sende die Session-ID in einem X-Routing-Key Header, um alle Anfragen einer Sitzung an dieselbe Instanz zu routen.

Abfragen Sie einen Agenten, der das Legacy-MLflow verwendet AgentServer

Verwenden Sie diesen Abschnitt für Agents, die Sie in Databricks-Apps mit dem Legacy-Agent-Server bereitstellen: MLflow AgentServer oder LongRunningAgentServer mit der ResponsesAgent-Schnittstelle. Diese Agenten bedienen die OpenAI Responses API unter /responses.

LongRunningAgentServer bietet dieselbe API, daher gelten die folgenden Beispiele auch dafür. Es werden auch Hintergrundausführungen unterstützt: Legen Sie background in der Anforderung auf true fest, und rufen Sie dann die Antwort mit GET /responses/<response-id>?stream=true&starting_after=<sequence-number> ab, wodurch die Ereignisse nach dieser Sequenznummer gestreamt werden.

Databricks OpenAI-Client

Databricks empfiehlt den OpenAI-Client von Databricks für diese Agenten. Fügen Sie das apps/ Präfix dem Modellnamen voran.

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)

Verwenden Sie den Parameter extra_body, um custom_inputs zu übergeben:

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

Um die Trace-ID für eine Anfrage zu erhalten, fügen Sie den x-mlflow-return-trace-id Header hinzu. Verwenden Sie dann MLflow get_trace, um die vollständige Protokollierung abzurufen.

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

Sende Anfragen an den /responses Pfad der App-URL. Der Anforderungstext folgt der OpenAI Responses-API, sodass Sie jeden HTTP-Client oder jedes Tool verwenden können, das diese unterstützt.

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
  }'

Um custom_inputs zu übergeben, fügen Sie sie dem Anforderungstext hinzu:

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 }
  }'

Um die Trace-ID zu erhalten, fügen Sie den x-mlflow-return-trace-id: true Header hinzu. Der Antwort-Body enthält die Trace-ID in einem Feld metadata.trace_id . Bei Streaming-Anfragen erscheint die Trace-ID als separates SSE-Ereignis (data: {"trace_id": "tr-..."}) gegen Ende des Streams.

Legacy-Agent auf Model Serving abfragen

Verwenden Sie diesen Abschnitt für Legacy-Agenten, die auf Model Serving Endpunkte bereitgestellt werden. Du kannst dich mit einem Databricks OAuth-Token oder einem persönlichen Zugriffstoken authentifizieren. Um diese Agenten zu Databricks Apps zu verschieben, siehe Einen Agenten von Model Serving zu Databricks Apps migrieren.

Databricks OpenAI-Client

Rufen Sie für Agents, die die ResponsesAgent-Schnittstelle verwenden, responses.create mit dem Endpunktnamen als Modell auf:

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)

Verwenden Sie für Agents, die die Legacy-Schnittstellen ChatAgent oder ChatModel verwenden, den Chatvervollständigungsclient:

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)

Mit beiden Clients können Sie extra_body oder custom_inputs über den Parameter databricks_options übergeben. Zum Beispiel gibt extra_body={"databricks_options": {"return_trace": True}} den Trace mit der Antwort zurück.

REST API

Senden Sie für Agents, die die ResponsesAgent-Schnittstelle verwenden, eine Anforderung an /serving-endpoints/responses mit dem Endpunktnamen als Modell:

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
  }'

Für Agenten, die die Schnittstellen ChatAgent oder ChatModel verwenden, senden Sie eine Anfrage an /serving-endpoints/chat/completions mit einer messages Liste statt input. Um custom_inputs oder databricks_options zu übergeben, fügen Sie sie dem Request-Body hinzu. Du kannst auch Anfragen an die /serving-endpoints/<agent-endpoint-name>/invocations URL des Endpunkts senden. Siehe Einzelne Modelle hinter einem Endpunkt abfragen.

KI-Spielplatz

Um mit einem Agenten auf Model Serving zu chatten, ohne Code zu schreiben, öffnen Sie AI Playground und wählen Sie den Serving-Endpunkt des Agenten aus. Informationen zum Übergeben von custom_inputs an den Agent aus AI Playground finden Sie unter Bereitstellen von custom_inputs in AI Playground und der Überprüfungs-App.

SQL mit

Verwenden Sie ai_query, um einen Agent in der Modellbereitstellung aus SQL abzufragen. Siehe ai_query Funktion für Syntax und Parameter.

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

Weitere Ressourcen

• Agent Server
• Agent-Runtime
• Produktionsüberwachung einrichten
• Foundation- und Einbettungsmodelle abfragen: Fragen Sie Foundation-Modelle und andere Modelle direkt anstelle eines Agents ab.