メインコンテンツまでスキップ
に最終更新

Databricksにデプロイされたエージェントにクエリーを実行する

エージェントに対してどのようにクエリーを実行するかは、それをサービングするエージェントサーバーによって異なります。以下の表からエージェントを見つけ、対応するセクションを参照してください。エージェントサーバーの詳細については、エージェント サーバーを参照してください。

エージェント

ホスト元：

クエリーの方法

使用 DurableAgentServer

Databricks Apps上のエージェントランタイム

呼び出しAPIのエンドポイント： /api/invocations

MLflow AgentServer または LongRunningAgentServer を使用する (レガシー)

Databricks Apps

Databricks OpenAI クライアントまたは次の場所にある OpenAI Responses API /responses

Model Serving にデプロイ済み (レガシー)

モデルサービングエンドポイント

Databricks OpenAIクライアント、REST API、ai_query、またはAI Playground

エージェント

ホスト元：

クエリーの方法

使用 DurableAgentServer

Databricks Apps上のエージェントランタイム

呼び出しAPIのエンドポイント： /api/invocations

MLflow AgentServer または LongRunningAgentServer を使用する (レガシー)

Databricks Apps

Databricks OpenAI クライアントまたは次の場所にある OpenAI Responses API /responses

Model Serving にデプロイ済み (レガシー)

モデルサービングエンドポイント

Databricks OpenAIクライアント、REST API、ai_query、またはAI Playground

Databricks Apps でホストされるエージェントには、Databricks OAuth トークンが必要です。パーソナル アクセストークンは Databricks Apps では機能しません。スクリプト、Service Principal、別のアプリ、またはノートブックから OAuth トークンを生成する方法については、トークン認証を使用して API Databricks アプリに接続するを参照してください。

次を使用するエージェントをクエリーする DurableAgentServer​

DurableAgentServer を使用するエージェントが呼び出し API を提供します。Agent Bricks CLI で作成したエージェントは DurableAgentServer を使用し、agentbricks deploy はそれらを agent-bricks-<name> という名前のアプリとして Agent ランタイム にデプロイします。API への各リクエストによって、エージェントの 1 つのランが起動されます。これは呼び出しと呼ばれます。

エンドポイント

説明

POST /api/invocations

呼び出しを起動します。By default では、リクエストは待機して結果を返します。イベントが発生したときに受け取るには stream に設定し、すぐに返すには background に設定します。

GET /api/invocations/<id>

呼び出しのステータスを返し、完了後に出力を返します。

GET /api/invocations/<id>/events?after=<event-id>

<event-id> の後にある保存済みイベントをストリームします。ストリームに再接続するには、この Endpoint を使用します。

エンドポイント

説明

POST /api/invocations

呼び出しを起動します。By default では、リクエストは待機して結果を返します。イベントが発生したときに受け取るには stream に設定し、すぐに返すには background に設定します。

GET /api/invocations/<id>

呼び出しのステータスを返し、完了後に出力を返します。

GET /api/invocations/<id>/events?after=<event-id>

<event-id> の後にある保存済みイベントをストリームします。ストリームに再接続するには、この Endpoint を使用します。

リクエスト本文​

POST /api/invocations のリクエスト本文では、以下のフィールドを指定できます。サーバーは、他のフィールドが含まれているリクエストを拒否します。

フィールド

説明

id

必須。各呼び出しに対して生成するUUID。サーバーはIDをべき等性キーとして扱います。同じIDで同じリクエストを再送信すると、エージェントを再実行する代わりに既存の呼び出しが返されます。異なるリクエストでIDを再利用すると、409 エラーが返されます。

session_id

呼び出しが属する会話。セッション ID を共有する呼び出しは、順番に 1 つずつ実行されます。CLI Templateから生成されたエージェントでは、このフィールドが必要です。

input

エージェントへの入力です。CLI Templateから生成されたエージェントは、メッセージのリスト、または messages リストを持つオブジェクトを受け入れます。

stream

Server-Sent Events（SSE）としてイベントを受信するには、true に設定します。

background

trueに設定すると、ステータスURLを含む202応答がすぐに返され、その後結果がポーリングされます。

フィールド

説明

id

必須。各呼び出しに対して生成するUUID。サーバーはIDをべき等性キーとして扱います。同じIDで同じリクエストを再送信すると、エージェントを再実行する代わりに既存の呼び出しが返されます。異なるリクエストでIDを再利用すると、409 エラーが返されます。

session_id

呼び出しが属する会話。セッション ID を共有する呼び出しは、順番に 1 つずつ実行されます。CLI Templateから生成されたエージェントでは、このフィールドが必要です。

input

エージェントへの入力です。CLI Templateから生成されたエージェントは、メッセージのリスト、または messages リストを持つオブジェクトを受け入れます。

stream

Server-Sent Events（SSE）としてイベントを受信するには、true に設定します。

background

trueに設定すると、ステータスURLを含む202応答がすぐに返され、その後結果がポーリングされます。

エージェントのハンドラーは input の構造を定義します。CLI Templateから生成されたエージェントは、input がオブジェクトである場合に次のフィールドを読み取ります：

フィールド

説明

messages

エージェントに送信する会話のターン。

actor

エージェントが読み書きする長期記憶の識別情報。actorを渡さない場合、エージェントはセッションIDを使用するため、メモリは新しいセッションに引き継がれません。ユーザーが入力したテキストからではなく、アプリケーションにサインインしたユーザーからactorを設定します。

model

エージェントコードで設定されたモデルの代わりに、この呼び出しで使用するモデル。

resume

ツール呼び出しの承認など、人間の入力を待つために停止するエージェントへの応答。エージェントが一時停止すると、呼び出しの status は interrupted になります。続行するには、同じ session_id を指定した新しい呼び出しで resume を送信します。

フィールド

説明

messages

エージェントに送信する会話のターン。

actor

エージェントが読み書きする長期記憶の識別情報。actorを渡さない場合、エージェントはセッションIDを使用するため、メモリは新しいセッションに引き継がれません。ユーザーが入力したテキストからではなく、アプリケーションにサインインしたユーザーからactorを設定します。

model

エージェントコードで設定されたモデルの代わりに、この呼び出しで使用するモデル。

resume

ツール呼び出しの承認など、人間の入力を待つために停止するエージェントへの応答。エージェントが一時停止すると、呼び出しの status は interrupted になります。続行するには、同じ session_id を指定した新しい呼び出しで resume を送信します。

エージェントがセッション間でユーザーについて学習した内容を記憶できるようにするには、ユーザーの ID を actor として渡します。

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

ターミナルからデプロイされたエージェントをテストするには、agentbricks endpoint invoke を使用します。このコマンドはアプリを検索し、CLIプロファイルで認証を行います。

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

応答をストリームするには、JSON本文に"stream":trueを追加し、--sseを渡します。agentbricks devを使用してローカルで実行中にエージェントをテストするには、アプリ名を--url http://localhost:8000に置き換えます。

ストリーム、バックグラウンドでのラン、および再接続​

  • ストリーム : セット "stream": true。レスポンスは、run.startedおよびrun.completed（またはrun.failed）イベントに加えて、ストリームテキストを含むdeltaイベントなど、エージェントが発行するイベントを含むSSEストリームです。各イベントには ID があります。
  • バックグラウンドでラン : "background": trueを設定します。サーバーは、status_urlを含む202応答を返します。ステータスがcompletedになるまで、GET /api/invocations/<id>をポーリングします。"stream": trueも設定すると、応答にイベントの読み取り元となるevents_urlが含まれます。
  • 再接続 ：ストリームが切断された場合は、受信した最後のエベントの ID を指定して GET /api/invocations/<id>/events?after=<event-id> を呼び出します。

次のPythonの例では、レスポンスをストリームします。

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

複数のインスタンスを使用してエージェントをデプロイする場合は、セッション内のすべてのリクエストを同じインスタンスにルーティングするために、X-Routing-KeyヘッダーでセッションIDを送信します。

レガシー MLflow を使用するエージェントにクエリーする AgentServer​

レガシーエージェントサーバーを使用して Databricks Apps にデプロイするエージェントの場合は、このセクションを使用します。MLflowAgentServer LongRunningAgentServerの または をResponsesAgent インターフェイスとともに使用します。これらのエージェントは、/responses で OpenAI Responses API を提供します。

LongRunningAgentServer 同じ API を提供するため、次の例も適用されます。また、バックグラウンドランもサポートされます。リクエストで background を true に設定し、そのシーケンス番号以降のイベントをストリームする GET /responses/<response-id>?stream=true&starting_after=<sequence-number> でレスポンスを取得します。

Databricksでは、これらのエージェントに対してDatabricks OpenAI clientをお勧めしています。モデル名に apps/ プレフィックスを含めます。

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

custom_inputs を渡すには、extra_body パラメーターを使用します。

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

リクエストのトレースIDを取得するには、x-mlflow-return-trace-idヘッダーを含めます。次に、MLflow の get_trace を使用して、完全なトレースを取得します。

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

Model Serving でレガシーエージェントをクエリーする​

Model Servingエンドポイントにデプロイされたレガシーエージェントについては、このセクションを使用してください。Databricks OAuthトークンまたは個人アクセストークンを使用して認証できます。これらのエージェントをDatabricks Appsに移行するには、Model ServingからDatabricks Appsへのエージェントの移行を参照してください。

ResponsesAgent インターフェイスを使用するエージェントの場合は、エンドポイント名をモデルとして指定して responses.create を呼び出します。

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

レガシーの ChatAgent または ChatModel インターフェイスを使用するエージェントの場合は、チャットコンプリーションクライアントを使用します。

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

どちらのクライアントでも、extra_bodyパラメーターを介してcustom_inputsまたはdatabricks_optionsを渡します。たとえば、extra_body={"databricks_options": {"return_trace": True}}は応答とともにトレースを返します。

その他のリソース​

このページの見出し