查询 Azure Databricks 上部署的代理

了解如何将请求发送到部署到 Databricks 应用或模型服务终结点的代理。 Databricks 提供了多个查询方法，以满足不同的用例和集成需求。

选择最适合用例的查询方法：

方法	主要优点
Databricks OpenAI 客户端（建议）	原生集成、全面功能支持、流媒体处理功能
REST API	与 OpenAI 兼容、与语言无关、适用于现有工具
AI 函数： `ai_query`	仅在模型服务端点上托管的与 OpenAI 兼容的查询遗留代理

Databricks 建议为新应用程序使用 Databricks OpenAI 客户端 。与需要 OpenAI 兼容终结点的平台集成时，请选择 REST API 。

Databricks OpenAI 客户端（建议）

Databricks 建议使用 DatabricksOpenAI 客户端查询已部署的代理。根据您已部署代理的 API，您将使用响应客户端或聊天完成功能客户端：

部署到应用的代理

根据ResponsesAgent接口，为Databricks Apps 上托管的代理使用以下示例，这是构建代理的建议方法。必须使用 Databricks OAuth 令牌来查询 Databricks Apps 上托管的代理。

from databricks.sdk import WorkspaceClient
from databricks_openai import DatabricksOpenAI

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

# The WorkspaceClient must be configured with OAuth authentication
# See: https://docs.databricks.com/aws/en/dev-tools/auth/oauth-u2m.html
w = WorkspaceClient()

client = DatabricksOpenAI(workspace_client=w)

# Run for non-streaming responses. Calls the "invoke" method
# Include the "apps/" prefix in the model name
response = client.responses.create(model=f"apps/{app_name}", input=input_msgs)
print(response)

# Include stream=True for streaming responses. Calls the "stream" method
# Include the "apps/" prefix in the model name
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 参数来添加它们。

streaming_response = client.responses.create(
    model=f"apps/{app_name}",
    input=input_msgs,
    stream=True,
    extra_body={
        "custom_inputs": {"id": 5},
    },
)
for chunk in streaming_response:
    print(chunk)

模型服务中的代理

使用以下示例来处理在模型服务上托管并遵循ResponsesAgent接口的旧版代理。可以使用 Databricks OAuth 令牌或个人访问令牌（PAT）查询模型服务上托管的代理。

from databricks_openai import DatabricksOpenAI

input_msgs = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>" # TODO: update this with your endpoint name

client = DatabricksOpenAI()

# Run for non-streaming responses. Invokes `predict`
response = client.responses.create(model=endpoint, input=input_msgs)
print(response)

# Include stream=True for streaming responses. Invokes `predict_stream`
streaming_response = client.responses.create(model=endpoint, input=input_msgs, stream=True)
for chunk in streaming_response:
  print(chunk)

如果想传入 custom_inputs 或 databricks_options，可以通过 extra_body 参数来添加它们：

streaming_response = client.responses.create(
    model=endpoint,
    input=input_msgs,
    stream=True,
    extra_body={
        "custom_inputs": {"id": 5},
        "databricks_options": {"return_trace": True},
    },
)
for chunk in streaming_response:
    print(chunk)

对遵循 ChatAgent 或 ChatModel 接口的模型服务上的旧代理使用以下示例。

from databricks.sdk import WorkspaceClient

messages = [{"role": "user", "content": "What does Databricks do?"}]
endpoint = "<agent-endpoint-name>" # TODO: update this with your endpoint name

ws_client = WorkspaceClient()
client = ws_client.serving_endpoints.get_open_ai_client()

# Run for non-streaming responses. Invokes `predict`
response = client.chat.completions.create(model=endpoint, messages=messages)
print(response)

# Include stream=True for streaming responses. Invokes `predict_stream`
streaming_response = client.chat.completions.create(model=endpoint, messages=messages, stream=True)
for chunk in streaming_response:
  print(chunk)

如果想传入 custom_inputs 或 databricks_options，可以通过 extra_body 参数来添加它们：

streaming_response = client.chat.completions.create(
    model=endpoint,
    messages=messages,
    stream=True,
    extra_body={
        "custom_inputs": {"id": 5},
        "databricks_options": {"return_trace": True},
    },
)
for chunk in streaming_response:
    print(chunk)

REST API

Databricks REST API 为与 OpenAI 兼容的模型提供终结点。这样，就可以使用 Databricks 代理为需要 OpenAI 接口的应用程序提供服务。

此方法非常适合：

与语言无关的使用 HTTP 请求的应用程序
与需要 OpenAI 兼容的 API 的第三方平台集成
使用最少的代码更改从 OpenAI 迁移到 Databricks

使用 Databricks OAuth 令牌通过 REST API 进行身份验证。有关更多选项和信息，请参阅 Databricks 身份验证文档。

部署到应用的代理

根据ResponsesAgent，对Databricks Apps 上托管的代理使用以下示例，这是构建代理的推荐方法。必须使用 Databricks OAuth 令牌来查询 Databricks Apps 上托管的代理。

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

如果要传入 custom_inputs，可以将它们添加到请求正文：

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

模型服务上的代理

在ResponsesAgent接口之后，使用以下示例对模型服务上托管的旧代理进行处理。可以使用 Databricks OAuth 令牌或个人访问令牌（PAT）查询模型服务上托管的代理。 REST API 调用等效于：

使用 Databricks OpenAI 客户端与 responses.create。
将 POST 请求发送到特定终结点的 URL（例如： https://<host.databricks.com>/serving-endpoints/\<model-name\>/invocations）。有关详细信息，请参阅终结点的“模型服务”页和 “模型服务”文档。

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

如果要传入custom_inputs或databricks_options，可以将它们添加到请求正文中：

curl --request POST \
  --url https://<host.databricks.com\>/serving-endpoints/responses \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "model": "\<model-name\>",
    "input": [{ "role": "user", "content": "hi" }],
    "stream": true,
    "custom_inputs": { "id": 5 },
    "databricks_options": { "return_trace": true }
  }'

请将以下内容用于用旧版 ChatAgent 或 ChatModel 接口创建的代理。这等效于：

使用 Databricks OpenAI 客户端与 chat.completions.create。
将 POST 请求发送到特定终结点的 URL（例如： https://<host.databricks.com>/serving-endpoints/\<model-name\>/invocations）。有关详细信息，请参阅终结点的“模型服务”页和 “模型服务”文档。

curl --request POST \
  --url https://<host.databricks.com\>/serving-endpoints/chat/completions \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "model": "\<model-name\>",
    "messages": [{ "role": "user", "content": "hi" }],
    "stream": true
  }'

如果要传入custom_inputs或databricks_options，可以将它们添加到请求正文中：

curl --request POST \
  --url https://<host.databricks.com\>/serving-endpoints/chat/completions \
  --header 'Authorization: Bearer <OAuth token>' \
  --header 'content-type: application/json' \
  --data '{
    "model": "\<model-name\>",
    "messages": [{ "role": "user", "content": "hi" }],
    "stream": true,
    "custom_inputs": { "id": 5 },
    "databricks_options": { "return_trace": true }
  }'

AI 函数： `ai_query`

可以使用 ai_query SQL 查询在模型服务上托管的已部署代理。请参阅用于 SQL 语法和参数定义的 ai_query 函数。

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

后续步骤

在生产环境中监控 GenAI

反馈

此页面是否有帮助？

Last updated on 2026-02-14

通过

查询 Azure Databricks 上部署的代理

Databricks OpenAI 客户端（建议）

部署到应用的代理

模型服务中的代理

REST API

部署到应用的代理

模型服务上的代理

AI 函数： ai_query

后续步骤

反馈

其他资源

AI 函数： `ai_query`