mlflow.deployments

提供将 MLflow 模型部署到自定义服务工具的功能。

注意：目前可以通过 mlflow.sagemaker 模块将模型部署到 AWS Sagemaker。也可以使用 azureml library 将模型部署到 Azure。

MLflow 目前不提供对其他部署目标的内置支持，但可以通过第三方插件安装对自定义目标的支持。请参阅已知插件列表here。

该页面主要关注面向用户的部署 API。有关为自定义服务工具实现您自己的部署插件的说明，请参见 plugin docs。

class mlflow.deployments.BaseDeploymentClient(target_uri)

用于公开 Python 模型部署 APIs 的基类。

插件实现者应该在插件模块中通过继承BaseDeploymentClient来定义针对特定目标的部署逻辑，并使用与目标相关的信息自定义方法的文档字符串（docstrings）。

注意

子类在错误情况下应抛出 mlflow.exceptions.MlflowException（例如在部署模型失败时）。

abstract create_deployment(name, model_uri, flavor=None, config=None, endpoint=None)

Deploy a model to the specified target. By default, this method should block until deployment completes (i.e. until it’s possible to perform inference with the deployment). In the case of conflicts (e.g. if it’s not possible to create the specified deployment without due to conflict with an existing deployment), raises a mlflow.exceptions.MlflowException or an HTTPError for remote deployments. See target-specific plugin documentation for additional detail on support for asynchronous deployment and other configuration.

Parameters

• name – 用于部署的唯一名称。如果另一个部署存在相同的 name，则会引发 mlflow.exceptions.MlflowException

• model_uri – 要部署的模型的 URI

• flavor – (可选) 要部署的模型 flavor。如果未指定，将选择默认的 flavor。

• config – (可选) Dict 包含用于该部署的已更新的目标特定配置

• endpoint – (可选) 用于在其下创建部署的 endpoint。可能并非所有目标都支持

Returns

与已创建的部署对应的字典，必须包含 ‘name’ 键。

create_endpoint(name, config=None)

Create an endpoint with the specified target. By default, this method should block until creation completes (i.e. until it’s possible to create a deployment within the endpoint). In the case of conflicts (e.g. if it’s not possible to create the specified endpoint due to conflict with an existing endpoint), raises a mlflow.exceptions.MlflowException or an HTTPError for remote deployments. See target-specific plugin documentation for additional detail on support for asynchronous creation and other configuration.

Parameters

• name – 用于端点的唯一名称。如果存在另一个具有相同名称的端点，则会引发 mlflow.exceptions.MlflowException。

• config – （可选）Dict，包含用于该端点的目标特定配置。

Returns

Dict 对应于已创建的 endpoint，必须包含 ‘name’ 键。

abstract delete_deployment(name, config=None, endpoint=None)

从指定的目标删除名称为 name 的部署。

删除操作应具有幂等性（即，如果在不存在的部署上重试删除，不应失败）。

Parameters

• name – 要删除的部署名称

• config – （可选）dict，包含用于部署的更新的特定于目标的配置

• endpoint – (可选) 包含要删除的部署的 endpoint。可能并非所有目标都支持。

Returns

无

delete_endpoint(endpoint)

从指定目标删除该端点。删除操作应幂等（即在不存在的部署上重试删除时不应失败）。

Parameters

endpoint – 要删除的 endpoint 名称

Returns

无

explain(deployment_name=None, df=None, endpoint=None)

为已部署的模型生成针对指定输入 pandas Dataframe df 的预测解释。解释的输出格式因部署目标而异，可能包括用于理解/调试预测的特征重要性等详细信息。

Parameters

• deployment_name – 用于预测的部署名称

• df – Pandas DataFrame 用于解释模型预测中的特征重要性

• endpoint – 用于进行预测的 endpoint。可能并非所有目标都支持

Returns

一个可序列化为 JSON 的对象 (pandas dataframe, numpy array, dictionary)，或者如果部署目标的类中没有可用的实现，则为一个异常

abstract get_deployment(name, endpoint=None)

Returns a dictionary describing the specified deployment, throwing either a mlflow.exceptions.MlflowException or an HTTPError for remote deployments if no deployment exists with the provided ID. The dict is guaranteed to contain an ‘name’ key containing the deployment name. The other fields of the returned dictionary and their types may vary across deployment targets.

Parameters

• name – 要获取的部署 ID。

• endpoint – （可选）包含要获取的部署的端点。可能并非所有目标都支持。

Returns

对应于检索到的部署的字典。该字典保证包含一个‘name’键，对应于部署名称。返回字典的其他字段及其类型可能因目标而异。

get_endpoint(endpoint)

Returns a dictionary describing the specified endpoint, throwing a py:class:mlflow.exception.MlflowException or an HTTPError for remote deployments if no endpoint exists with the provided name. The dict is guaranteed to contain an ‘name’ key containing the endpoint name. The other fields of the returned dictionary and their types may vary across targets.

Parameters

endpoint – 要获取的 endpoint 的名称

Returns

与检索到的端点对应的 dict。该 dict 保证包含一个名为 ‘name’ 的键，对应端点的名称。返回字典的其他字段及其类型可能因目标而异。

abstract list_deployments(endpoint=None)

列出部署。

此方法预计返回所有 deployments 的未分页列表（另一种做法是返回一个字典，该字典带有一个 ‘deployments’ 字段以包含实际的 deployments，插件可以在返回的字典中指定其他字段，例如用于分页的 next_page_token 字段，并接受一个 pagination_args 参数以向此方法传递与分页相关的参数）。

Parameters

endpoint – （可选）列出指定 endpoint 中的部署。可能并非所有目标都支持。

Returns

一个对应于部署的字典（dict）列表。每个字典均保证包含一个‘name’键，其中包含部署名称。返回字典的其他字段及其类型可能因部署目标而异。

list_endpoints()

列出指定目标中的 endpoints。此方法预计返回所有 endpoints 的非分页列表（另一种方法是返回一个包含实际 endpoints 的 ‘endpoints’ 字段的 dict，插件可以在返回的字典中指定其他字段，例如用于分页的 next_page_token 字段，并接受一个 pagination_args 参数以将与分页相关的参数传递给此方法）。

Returns

对应端点的字典列表。每个字典都保证包含一个‘name’键，该键包含端点的名称。返回字典的其他字段及其类型可能在不同目标之间有所不同。

abstract predict(deployment_name=None, inputs=None, endpoint=None)

使用指定的部署或模型端点对输入进行预测。

请注意，此方法的输入/输出类型与 mlflow pyfunc predict 相同。

Parameters

• deployment_name – 要对其进行预测的部署的名称。

• inputs – 要传递给部署或模型端点以进行推理的输入数据（或参数）。

• endpoint – 用于进行预测的端点。可能并非所有目标都支持。

Returns

一个 mlflow.deployments.PredictionsResponse 实例，表示预测结果及相关模型服务器响应元数据。

predict_stream(deployment_name=None, inputs=None, endpoint=None)

向已配置的提供者端点提交查询，并获取流式响应

Parameters

• deployment_name – 要对其进行预测的部署名称。

• inputs – 查询的输入，作为字典。

• endpoint – 要查询的 endpoint 的名称。

Returns

一个字典迭代器，包含来自端点的响应。

abstract update_deployment(name, model_uri=None, flavor=None, config=None, endpoint=None)

使用指定名称更新部署。您可以更新模型的 URI、已部署模型的 flavor（在这种情况下也必须指定模型 URI），以及/或者部署的任何特定目标属性（通过 config）。默认情况下，此方法应阻塞直到部署完成（即直到可以对更新后的部署执行推理）。有关异步部署和其他配置支持的更多详细信息，请参阅特定目标插件文档。

Parameters

• name – 要更新的部署的唯一名称。

• model_uri – 要部署的新模型的 URI。

• flavor – (可选) 用于部署的新模型 flavor。如果提供，model_uri 也必须被指定。如果未指定 flavor 但指定了 model_uri，将选择一个默认的 flavor，并使用该 flavor 更新部署。

• config – (可选) dict，包含更新后的针对特定目标的部署配置。

• endpoint – (可选) 包含要更新的部署的 endpoint。可能并非所有目标都支持。

Returns

无

update_endpoint(endpoint, config=None)

使用指定名称更新端点。您可以更新端点的任何特定目标属性（通过 config）。默认情况下，此方法应阻塞直到更新完成（即直到可以在该端点中创建部署）。有关对异步更新及其他配置支持的更多详细信息，请参阅特定目标的插件文档。

Parameters

• endpoint – 要更新的 endpoint 的唯一名称

• config –（可选）包含针对该端点的目标特定配置的 dict

Returns

无

class mlflow.deployments.DatabricksDeploymentClient(target_uri)

用于与 Databricks 的 Serving 端点交互的客户端。

示例：

首先，设置用于身份验证的凭据：

export DATABRICKS_HOST=...
export DATABRICKS_TOKEN=...

另请参见

有关其他身份验证方法，请参见 https://docs.databricks.com/en/dev-tools/auth.html。

然后，创建一个部署客户端并使用它与 Databricks 的 Serving 端点进行交互：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
endpoints = client.list_endpoints()
assert endpoints == [
    {
        "name": "chat",
        "creator": "alice@company.com",
        "creation_timestamp": 0,
        "last_updated_timestamp": 0,
        "state": {...},
        "config": {...},
        "tags": [...],
        "id": "88fd3f75a0d24b0380ddc40484d7a31b",
    },
]

create_deployment(name, model_uri, flavor=None, config=None, endpoint=None)

警告

此方法未为 DatabricksDeploymentClient 实现。

create_endpoint(name=None, config=None, route_optimized=False)

使用提供的名称和配置创建一个新的服务端点。

参见 https://docs.databricks.com/api/workspace/servingendpoints/create 了解请求/响应模式。

Parameters

• name –

  要创建的服务端点的 name。

  警告

  已弃用。请在 config 中包含 name。

• config – 一个字典，包含完整的 API 请求负载或要创建的服务端点的配置。

• route_optimized –

  一个布尔值，用于定义 databricks serving endpoint 是否针对路由流量进行了优化。仅在已弃用的方法中使用。

  警告

  已弃用。请改为将 route_optimized 包含在 config 中。

Returns

一个 DatabricksEndpoint 对象，包含请求的响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
endpoint = client.create_endpoint(
    config={
        "name": "test",
        "config": {
            "served_entities": [
                {
                    "external_model": {
                        "name": "gpt-4",
                        "provider": "openai",
                        "task": "llm/v1/chat",
                        "openai_config": {
                            "openai_api_key": "{{secrets/scope/key}}",
                        },
                    },
                }
            ],
            "route_optimized": True,
        },
    },
)
assert endpoint == {
    "name": "test",
    "creator": "alice@company.com",
    "creation_timestamp": 0,
    "last_updated_timestamp": 0,
    "state": {...},
    "config": {...},
    "tags": [...],
    "id": "88fd3f75a0d24b0380ddc40484d7a31b",
    "permission_level": "CAN_MANAGE",
    "route_optimized": False,
    "task": "llm/v1/chat",
    "endpoint_type": "EXTERNAL_MODEL",
    "creator_display_name": "Alice",
    "creator_kind": "User",
}

delete_deployment(name, config=None, endpoint=None)

警告

该方法尚未为 DatabricksDeploymentClient 实现。

delete_endpoint(endpoint)

删除指定的服务端点。 请参见 https://docs.databricks.com/api/workspace/servingendpoints/delete 了解请求/响应模式。

Parameters

endpoint – 要删除的 serving endpoint 的名称。

Returns

包含请求响应的 DatabricksEndpoint 对象。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
client.delete_endpoint(endpoint="chat")

get_deployment(name, endpoint=None)

警告

此方法未为 DatabricksDeploymentClient 实现。

get_endpoint(endpoint)

获取指定的服务端点。有关请求/响应模式，请参见 https://docs.databricks.com/api/workspace/servingendpoints/get。

Parameters

endpoint – 要获取的服务端点的名称。

Returns

一个 DatabricksEndpoint 对象，包含请求响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
endpoint = client.get_endpoint(endpoint="chat")
assert endpoint == {
    "name": "chat",
    "creator": "alice@company.com",
    "creation_timestamp": 0,
    "last_updated_timestamp": 0,
    "state": {...},
    "config": {...},
    "tags": [...],
    "id": "88fd3f75a0d24b0380ddc40484d7a31b",
}

list_deployments(endpoint=None)

警告

此方法未为 DatabricksDeploymentClient 实现。

list_endpoints()

检索所有服务端点。

有关请求/响应模式，请参见 https://docs.databricks.com/api/workspace/servingendpoints/list。

Returns

包含请求响应的 DatabricksEndpoint 对象列表。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
endpoints = client.list_endpoints()
assert endpoints == [
    {
        "name": "chat",
        "creator": "alice@company.com",
        "creation_timestamp": 0,
        "last_updated_timestamp": 0,
        "state": {...},
        "config": {...},
        "tags": [...],
        "id": "88fd3f75a0d24b0380ddc40484d7a31b",
    },
]

predict(deployment_name=None, inputs=None, endpoint=None)

使用所提供的模型输入查询一个 serving endpoint。 参见 https://docs.databricks.com/api/workspace/servingendpoints/query 以获取请求/响应模式。

Parameters

• deployment_name – 未使用。

• inputs – 一个字典，包含要查询的模型输入。

• endpoint – 要查询的 serving endpoint 的名称。

Returns

一个 DatabricksEndpoint 对象包含查询响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
response = client.predict(
    endpoint="chat",
    inputs={
        "messages": [
            {"role": "user", "content": "Hello!"},
        ],
    },
)
assert response == {
    "id": "chatcmpl-8OLm5kfqBAJD8CpsMANESWKpLSLXY",
    "object": "chat.completion",
    "created": 1700814265,
    "model": "gpt-4-0613",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "Hello! How can I assist you today?",
            },
            "finish_reason": "stop",
        }
    ],
    "usage": {
        "prompt_tokens": 9,
        "completion_tokens": 9,
        "total_tokens": 18,
    },
}

predict_stream(deployment_name=None, inputs=None, endpoint=None) → Iterator[dict[str, typing.Any]]

向已配置的提供者端点提交查询，并获取流式响应

Parameters

• deployment_name – 未使用。

• inputs – 查询的输入，作为字典。

• endpoint – 要查询的 endpoint 的名称。

Returns

一个字典迭代器，包含来自端点的响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
chunk_iter = client.predict_stream(
    endpoint="databricks-llama-2-70b-chat",
    inputs={
        "messages": [{"role": "user", "content": "Hello!"}],
        "temperature": 0.0,
        "n": 1,
        "max_tokens": 500,
    },
)
for chunk in chunk_iter:
    print(chunk)
    # Example:
    # {
    #     "id": "82a834f5-089d-4fc0-ad6c-db5c7d6a6129",
    #     "object": "chat.completion.chunk",
    #     "created": 1712133837,
    #     "model": "llama-2-70b-chat-030424",
    #     "choices": [
    #         {
    #             "index": 0, "delta": {"role": "assistant", "content": "Hello"},
    #             "finish_reason": None,
    #         }
    #     ],
    #     "usage": {"prompt_tokens": 11, "completion_tokens": 1, "total_tokens": 12},
    # }

update_deployment(name, model_uri=None, flavor=None, config=None, endpoint=None)

警告

该方法尚未为 DatabricksDeploymentClient 实现。

update_endpoint(endpoint, config=None)

警告

mlflow.deployments.databricks.DatabricksDeploymentClient.update_endpoint 已弃用。此方法将在未来的版本中移除。请使用 update_endpoint_config, update_endpoint_tags, update_endpoint_rate_limits, or update_endpoint_ai_gateway。

使用提供的配置更新指定的服务端点。 有关请求/响应 schema，请参阅 https://docs.databricks.com/api/workspace/servingendpoints/updateconfig。

Parameters

• endpoint – 要更新的 serving endpoint 的名称。

• config – 一个包含要更新的 serving endpoint 的配置的字典。

Returns

一个 DatabricksEndpoint 对象，包含请求的响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
endpoint = client.update_endpoint(
    endpoint="chat",
    config={
        "served_entities": [
            {
                "name": "test",
                "external_model": {
                    "name": "gpt-4",
                    "provider": "openai",
                    "task": "llm/v1/chat",
                    "openai_config": {
                        "openai_api_key": "{{secrets/scope/key}}",
                    },
                },
            }
        ],
    },
)
assert endpoint == {
    "name": "chat",
    "creator": "alice@company.com",
    "creation_timestamp": 0,
    "last_updated_timestamp": 0,
    "state": {...},
    "config": {...},
    "tags": [...],
    "id": "88fd3f75a0d24b0380ddc40484d7a31b",
}

rate_limits = client.update_endpoint(
    endpoint="chat",
    config={
        "rate_limits": [
            {
                "key": "user",
                "renewal_period": "minute",
                "calls": 10,
            }
        ],
    },
)
assert rate_limits == {
    "rate_limits": [
        {
            "key": "user",
            "renewal_period": "minute",
            "calls": 10,
        }
    ],
}

update_endpoint_ai_gateway(endpoint, config)

更新指定服务端点的 AI 网关配置。

Parameters

• endpoint (str) – 要更新的 serving endpoint 的名称。

• config (dict) – 一个包含要更新的 AI Gateway 配置的字典。

Returns

包含已更新 AI Gateway 配置的字典。

Return type

dict

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
name = "test"

gateway_config = {
    "usage_tracking_config": {"enabled": True},
    "inference_table_config": {
        "enabled": True,
        "catalog_name": "my_catalog",
        "schema_name": "my_schema",
    },
}

updated_gateway = client.update_endpoint_ai_gateway(
    endpoint=name, config=gateway_config
)
assert updated_gateway == {
    "usage_tracking_config": {"enabled": True},
    "inference_table_config": {
        "catalog_name": "my_catalog",
        "schema_name": "my_schema",
        "table_name_prefix": "test",
        "enabled": True,
    },
}

update_endpoint_config(endpoint, config)

更新指定服务端点的配置。请参见 https://docs.databricks.com/api/workspace/servingendpoints/updateconfig 以获取请求/响应模式。

Parameters

• endpoint – 要更新的 serving endpoint 的名称。

• config – 一个字典，包含要更新的服务端点的配置。

Returns

一个 DatabricksEndpoint 对象，包含请求的响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
updated_endpoint = client.update_endpoint_config(
    endpoint="test",
    config={
        "served_entities": [
            {
                "name": "gpt-4o-mini",
                "external_model": {
                    "name": "gpt-4o-mini",
                    "provider": "openai",
                    "task": "llm/v1/chat",
                    "openai_config": {
                        "openai_api_key": "{{secrets/scope/key}}",
                    },
                },
            }
        ]
    },
)
assert updated_endpoint == {
    "name": "test",
    "creator": "alice@company.com",
    "creation_timestamp": 1729527763000,
    "last_updated_timestamp": 1729530896000,
    "state": {"ready": "READY", "config_update": "NOT_UPDATING"},
    "config": {...},
    "id": "44b258fb39804564b37603d8d14b853e",
    "permission_level": "CAN_MANAGE",
    "route_optimized": False,
    "task": "llm/v1/chat",
    "endpoint_type": "EXTERNAL_MODEL",
    "creator_display_name": "Alice",
    "creator_kind": "User",
}

update_endpoint_rate_limits(endpoint, config)

更新指定 serving endpoint 的速率限制。请参见 https://docs.databricks.com/api/workspace/servingendpoints/put 以获取请求/响应架构。

Parameters

• endpoint – 要更新的 serving endpoint 的名称。

• config – 一个包含更新后的速率限制配置的字典。

Returns

一个 DatabricksEndpoint 对象，包含更新后的速率限制。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
name = "databricks-dbrx-instruct"
rate_limits = {
    "rate_limits": [{"calls": 10, "key": "endpoint", "renewal_period": "minute"}]
}
updated_rate_limits = client.update_endpoint_rate_limits(
    endpoint=name, config=rate_limits
)
assert updated_rate_limits == {
    "rate_limits": [{"calls": 10, "key": "endpoint", "renewal_period": "minute"}]
}

update_endpoint_tags(endpoint, config)

更新指定的 serving endpoint 的标签。有关请求/响应模式，请参见 https://docs.databricks.com/api/workspace/servingendpoints/patch。

Parameters

• endpoint – 要更新的 serving endpoint 的名称。

• config – 一个包含要添加和/或移除的标签的字典。

Returns

一个 DatabricksEndpoint 对象，包含请求的响应。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("databricks")
updated_tags = client.update_endpoint_tags(
    endpoint="test", config={"add_tags": [{"key": "project", "value": "test"}]}
)
assert updated_tags == {"tags": [{"key": "project", "value": "test"}]}

class mlflow.deployments.DatabricksEndpoint

一个类字典对象，表示一个 Databricks 服务端点。

endpoint = DatabricksEndpoint(
    {
        "name": "chat",
        "creator": "alice@company.com",
        "creation_timestamp": 0,
        "last_updated_timestamp": 0,
        "state": {...},
        "config": {...},
        "tags": [...],
        "id": "88fd3f75a0d24b0380ddc40484d7a31b",
    }
)
assert endpoint.name == "chat"

class mlflow.deployments.MlflowDeploymentClient(target_uri)

用于与 MLflow AI 网关交互的客户端。

示例：

首先，启动 MLflow AI Gateway：

mlflow gateway start --config-path path/to/config.yaml

然后，创建一个客户端并使用它与服务器交互：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("http://localhost:5000")
endpoints = client.list_endpoints()
assert [e.dict() for e in endpoints] == [
    {
        "name": "chat",
        "endpoint_type": "llm/v1/chat",
        "model": {"name": "gpt-4o-mini", "provider": "openai"},
        "endpoint_url": "http://localhost:5000/gateway/chat/invocations",
    },
]

create_deployment(name, model_uri, flavor=None, config=None, endpoint=None)

警告

此方法未在 MlflowDeploymentClient 中实现。

create_endpoint(name, config=None)

警告

该方法未为 MlflowDeploymentClient 实现。

delete_deployment(name, config=None, endpoint=None)

警告

此方法未为 MlflowDeploymentClient 实现。

delete_endpoint(endpoint)

警告

此方法尚未为 MlflowDeploymentClient 实现。

get_deployment(name, endpoint=None)

警告

该方法尚未在 MLflowDeploymentClient 中实现。

get_endpoint(endpoint) → Endpoint

获取为 MLflow AI Gateway 配置的指定端点。

Parameters

endpoint – 要检索的 endpoint 的名称。

Returns

一个Endpoint对象，表示该端点。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("http://localhost:5000")
endpoint = client.get_endpoint(endpoint="chat")
assert endpoint.dict() == {
    "name": "chat",
    "endpoint_type": "llm/v1/chat",
    "model": {"name": "gpt-4o-mini", "provider": "openai"},
    "endpoint_url": "http://localhost:5000/gateway/chat/invocations",
}

list_deployments(endpoint=None)

警告

此方法尚未为 MlflowDeploymentClient 实现。

list_endpoints() → list[Endpoint]

列出为 MLflow AI Gateway 配置的端点。

Returns

一组 Endpoint 对象。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("http://localhost:5000")

endpoints = client.list_endpoints()
assert [e.dict() for e in endpoints] == [
    {
        "name": "chat",
        "endpoint_type": "llm/v1/chat",
        "model": {"name": "gpt-4o-mini", "provider": "openai"},
        "endpoint_url": "http://localhost:5000/gateway/chat/invocations",
    },
]

predict(deployment_name=None, inputs=None, endpoint=None) → dict[str, typing.Any]

向已配置的提供者端点提交查询。

Parameters

• deployment_name – 未使用。

• inputs – 查询的输入，作为字典。

• endpoint – 要查询的 endpoint 的名称。

Returns

包含来自端点的响应的字典。

示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("http://localhost:5000")

response = client.predict(
    endpoint="chat",
    inputs={"messages": [{"role": "user", "content": "Hello"}]},
)
assert response == {
    "id": "chatcmpl-8OLoQuaeJSLybq3NBoe0w5eyqjGb9",
    "object": "chat.completion",
    "created": 1700814410,
    "model": "gpt-4o-mini",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "Hello! How can I assist you today?",
            },
            "finish_reason": "stop",
        }
    ],
    "usage": {
        "prompt_tokens": 9,
        "completion_tokens": 9,
        "total_tokens": 18,
    },
}

可以将对于给定提供者和端点配置有效的其他参数随请求一并包含，如下所示，使用 an openai completions endpoint request 作为示例：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("http://localhost:5000")
client.predict(
    endpoint="completions",
    inputs={
        "prompt": "Hello!",
        "temperature": 0.3,
        "max_tokens": 500,
    },
)

update_deployment(name, model_uri=None, flavor=None, config=None, endpoint=None)

警告

此方法未为MlflowDeploymentClient实现。

update_endpoint(endpoint, config=None)

警告

此方法未在 MlflowDeploymentClient 中实现。

class mlflow.deployments.OpenAIDeploymentClient(target_uri)

用于与 OpenAI 端点交互的客户端。

示例：

首先，设置用于身份验证的凭据：

export OPENAI_API_KEY=...

另请参见

请参阅 https://mlflow.org/docs/latest/python_api/openai/index.html 了解其他身份验证方法。

然后，创建一个部署客户端并使用它与 OpenAI 端点进行交互：

from mlflow.deployments import get_deploy_client

client = get_deploy_client("openai")
client.predict(
    endpoint="gpt-4o-mini",
    inputs={
        "messages": [
            {"role": "user", "content": "Hello!"},
        ],
    },
)

create_deployment(name, model_uri, flavor=None, config=None, endpoint=None)

警告

此方法未在OpenAIDeploymentClient中实现。

create_endpoint(name, config=None)

警告

此方法尚未为 OpenAIDeploymentClient 实现。

delete_deployment(name, config=None, endpoint=None)

警告

此方法未在 OpenAIDeploymentClient 中实现。

delete_endpoint(endpoint)

警告

此方法未为OpenAIDeploymentClient实现。

get_deployment(name, endpoint=None)

警告

此方法未在 OpenAIDeploymentClient 中实现。

get_endpoint(endpoint)

获取有关特定模型的信息。

list_deployments(endpoint=None)

警告

此方法未为OpenAIDeploymentClient实现。

list_endpoints()

列出当前可用的模型。

predict(deployment_name=None, inputs=None, endpoint=None)

查询 OpenAI 端点。有关更多信息，请参见 https://platform.openai.com/docs/api-reference。

Parameters

• deployment_name – 未使用。

• inputs – 一个包含要查询的模型输入的字典。

• endpoint – 要查询的 endpoint 的名称。

Returns

包含模型输出的字典。

update_deployment(name, model_uri=None, flavor=None, config=None, endpoint=None)

警告

此方法未在OpenAIDeploymentClient中实现。

update_endpoint(endpoint, config=None)

警告

此方法未在OpenAIDeploymentClient中实现。

mlflow.deployments.get_deploy_client(target_uri=None)

返回一个继承自 mlflow.deployments.BaseDeploymentClient 的子类，提供用于将模型部署到指定目标的标准 API。通过在返回的对象上调用 help() 或查看 mlflow.deployments.BaseDeploymentClient 的文档来查看可用的部署 API。你也可以通过 CLI 运行 mlflow deployments help -t 以获得有关特定目标配置选项的更多详细信息。

Parameters

target_uri – 可选要部署到的目标 URI。如果未提供目标 URI，MLflow 将尝试通过 get_deployments_target() 或 MLFLOW_DEPLOYMENTS_TARGET 环境变量来获取部署目标的设置。

Example

from mlflow.deployments import get_deploy_client
import pandas as pd

client = get_deploy_client("redisai")
# Deploy the model stored at artifact path 'myModel' under run with ID 'someRunId'. The
# model artifacts are fetched from the current tracking server and then used for deployment.
client.create_deployment("spamDetector", "runs:/someRunId/myModel")
# Load a CSV of emails and score it against our deployment
emails_df = pd.read_csv("...")
prediction_df = client.predict_deployment("spamDetector", emails_df)
# List all deployments, get details of our particular deployment
print(client.list_deployments())
print(client.get_deployment("spamDetector"))
# Update our deployment to serve a different model
client.update_deployment("spamDetector", "runs:/anotherRunId/myModel")
# Delete our deployment
client.delete_deployment("spamDetector")

mlflow.deployments.get_deployments_target() → str

当且仅当已设置时，返回当前配置的 MLflow 部署目标。 如果未使用 set_deployments_target 设置部署目标，则会引发 MlflowException。

mlflow.deployments.run_local(target, name, model_uri, flavor=None, config=None)

在本地部署指定的模型以进行测试。请注意，本地部署的模型不能被其他部署 APIs 管理（例如 update_deployment、delete_deployment 等）。

Parameters

• target – 要部署到的目标。

• name – 用于部署的名称

• model_uri – 要部署的模型的 URI

• flavor – (可选) 要部署的模型 flavor。如果未指定，将选择默认的 flavor。

• config –（可选）Dict，包含用于部署的针对特定目标的更新配置

Returns

无

mlflow.deployments.set_deployments_target(target: str)

为 MLflow 的部署设置目标客户端

Parameters

target – 运行中 MLflow AI Gateway 的完整 URI，或者如果在 Databricks 上运行，填写 “databricks”。

class mlflow.deployments.PredictionsResponse

表示在响应评分请求时返回的预测和元数据，例如发送到 MLflow Model Server 的 /invocations 端点的 REST API 请求。

get_predictions(predictions_format='dataframe', dtype=None)

以指定的格式获取来自 MLflow Model Server 的预测结果。

Parameters

• predictions_format – 返回预测结果的格式。可以是 "dataframe" 或 "ndarray"。

• dtype – 将预测强制转换为 NumPy 数据类型。仅在指定了 “ndarray” predictions_format 时使用。

Raises

Exception – If the predictions cannot be represented in the specified format.

Returns

以指定格式表示的预测结果。

to_json(path=None)

获取 MLflow Predictions Response 的 JSON 表示。

Parameters

path – 如果指定，JSON 表示会被写入到此文件路径。

Returns

如果 path 未指定，则为 MLflow Predictions Response 的 JSON 表示。否则，为 None。