企业官网建设流程全解析

MLflow 部署 API 完全指南：从 get_deploy_client 到自定义部署插件（mlflow.deployments 模块深度解析）

【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow

本篇技术指南以 MLflow 官方 API 参考文档 mlflow.deployments.rst 为骨架，系统讲解mlflow.deployments模块的核心设计：如何通过统一的get_deploy_client()与run_local()接口将模型部署到自定义服务目标，BaseDeploymentClient提供的标准部署管理方法集，内置的 Databricks、OpenAI、MLflow AI Gateway 三类客户端，以及面向插件开发者的扩展机制。读完本文，你将掌握 MLflow 部署 API 的完整调用链路、内置客户端的使用方法与源码级实现原理，并能基于插件协议为自定义服务目标编写部署插件。

一、模块定位：面向自定义服务目标的统一部署接口

mlflow.deployments是 MLflow 中专门负责"把 MLflow 模型部署到自定义 Serving 工具"的模块。根据 mlflow/deployments/init.py 的模块 docstring，其定位有以下要点：

• 内置目标有限：AWS SageMaker 的部署通过独立的mlflow.sagemaker模块完成；Azure 部署可通过azureml-mlflow库完成；除此之外 MLflow不内置其他部署目标的支持。
• 自定义目标靠插件：对自定义部署目标（如 RedisAI、Ray Serve 等）的支持通过第三方插件安装获得。
• 模块聚焦用户侧 API：该页面主要介绍面向使用者的部署 API；如何自己编写部署插件，见插件开发文档。

从 mlflow/deployments/init.py 的__all__可以看出模块导出的全部公开符号：

值得注意的实现细节：MlflowDeploymentClient依赖可选依赖（openai等），因此init.py 使用contextlib.suppress(Exception)包裹其导入——若可选依赖未安装，导入失败会被静默忽略，不影响模块其他部分加载。

二、第一步：设置部署目标（Deployments Target）

在使用部署 API 之前，通常需要先确定"要部署到哪"——即部署目标。相关工具函数位于 mlflow/deployments/utils.py，并提供两种全局设定方式。

2.1set_deployments_target()与get_deployments_target()

from mlflow.deployments import set_deployments_target, get_deployments_target # 设置全局部署目标（模块级全局变量 _deployments_target） set_deployments_target("http://localhost:5000") # 或 Databricks 环境：set_deployments_target("databricks") # 读取当前目标 target = get_deployments_target()

按 utils.py 的实现：

• set_deployments_target()在写入前会用_is_valid_target()校验目标必须是合法 URI（含 scheme 与 netloc）或字符串"databricks"，否则抛出MlflowException（INVALID_PARAMETER_VALUE）。
• get_deployments_target()的取值优先级：先返回模块内_deployments_target（即代码中设置过的值）；若未设置，则回退读取环境变量MLFLOW_DEPLOYMENTS_TARGET（见 mlflow/environment_variables.py 中定义）；两者皆无则抛MlflowException提示设置方式。

2.2 环境变量方式：MLFLOW_DEPLOYMENTS_TARGET

export MLFLOW_DEPLOYMENTS_TARGET="http://localhost:5000"

2.3parse_target_uri()：从 URI 解析目标名

get_deploy_client()内部依赖 parse_target_uri() 从target_uri中解析目标名，规则如下：

• "target"（无 scheme、只有 path）→ 返回整个 path 作为目标名；
• "target:/suffix"→ 返回 scheme 即"target"；
• 空 scheme 且空 path → 抛异常，提示"Deployment URIs must be of the form 'target' or 'target:/suffix'"。

例如get_deploy_client("sagemaker:/my-profile")会解析出目标名sagemaker，而get_deploy_client("databricks")直接以databricks为目标名。

三、get_deploy_client()：统一入口与标准工作流

get_deploy_client(target_uri=None)定义在 mlflow/deployments/interface.py，返回BaseDeploymentClient的子类实例，暴露用于部署模型的标准 API。其目标解析优先级：

1. 显式传入target_uri参数；
2. 未传时尝试get_deployments_target()（即代码设置的全局目标或MLFLOW_DEPLOYMENTS_TARGET环境变量）；
3. 若两者皆无，打印提示日志并返回None。

模块 docstring 与接口 docstring 中给出了一个完整的标准工作流示例（以 RedisAI 插件为例）：

from mlflow.deployments import get_deploy_client import pandas as pd client = get_deploy_client("redisai") # 将 run 中 ID 为 'someRunId' 的实验下、artifact 路径 'myModel' 处的模型部署起来 # （模型产物从当前 tracking server 拉取） client.create_deployment("spamDetector", "runs:/someRunId/myModel") # 读取一封邮件的 CSV 并对其评分 emails_df = pd.read_csv("...") prediction_df = client.predict_deployment("spamDetector", emails_df) # 列出所有部署、查看单个部署详情 print(client.list_deployments()) print(client.get_deployment("spamDetector")) # 更新部署以服务另一个模型 client.update_deployment("spamDetector", "runs:/anotherRunId/myModel") # 删除部署 client.delete_deployment("spamDetector")

注：predict_deployment是旧版命名，当前基类中对应方法是predict()（见下文第四节）；以上代码体现的是 API 的使用范式。

实现上，get_deploy_client()通过plugin_store[target]按目标名取出插件，再用inspect.getmembers遍历插件模块成员，找到唯一一个BaseDeploymentClient的非抽象子类并实例化返回（见 interface.py）。

3.1run_local()：本地测试部署

from mlflow.deployments import run_local # 将模型在本地部署，用于测试 run_local( target="redisai", # 部署目标 name="spamDetector", # 部署名称 model_uri="runs:/someRunId/myModel", # 模型 URI flavor=None, # (可选) 模型 flavor，缺省自动选择 config=None, # (可选) 目标特有的配置字典 )

按 interface.py，run_local()直接调用插件的run_local方法。其签名与create_deployment非常相似，因为二者逻辑上做的是同类操作。注意：本地部署的模型无法被其他部署管理 API（update_deployment、delete_deployment等）管理，它只服务于测试目的。

3.2_target_help()：目标专属帮助

_target_help(target)（interface.py）返回目标专属的详细文档字符串，当用户执行mlflow deployments help -t <target-name>CLI 时展示，内容包括：

• 传给create_deployment/update_deployment的config中目标专属字段的解释；
• target_uri的写法（如 AWS SageMaker 的target_uri采用"sagemaker:/<aws-cli-profile-name>"形式，其中aws-cli-profile-name是 AWS CLI 配置文件名）；
• 其他目标专属细节。

四、BaseDeploymentClient：标准部署 API 契约

BaseDeploymentClient定义于 mlflow/deployments/base.py，被@developer_stable注解，是插件实现方必须继承的抽象基类。它既是用户侧调用的标准接口集合，也是插件侧的契约清单。

4.1 插件模块必须实现的三大要素

按 base.py 的模块 docstring，一个合法的部署插件模块必须实现：

1. 恰好一个继承自BaseDeploymentClient的客户端类，暴露管理部署的主要用户 API；
2. 模块级run_local(target, name, model_uri, flavor=None, config=None)函数，用于本地测试部署；
3. 模块级target_help()函数，返回描述目标 URI 格式与部署 config 的帮助消息。

run_local与target_help在 base.py 中仅作为"文档占位"存在——直接调用会抛NotImplementedError，真正的实现必须在插件模块的顶层命名空间中，以plugin_module.run_local/plugin_module.target_help方式可调用。

4.2 部署管理抽象方法（必须实现）

以下方法均为@abc.abstractmethod，插件必须实现：

4.3 可选/基类兜底方法

以下方法在基类中提供了默认行为（抛出MlflowException或NotImplementedError），由支持该能力的插件选择性覆写：

• predict_stream(deployment_name, inputs, endpoint)：向已配置的 provider 端点提交查询并获取流式响应，返回 dict 的迭代器（base.py）。
• explain(deployment_name, df, endpoint)：对输入 DataFrame 生成模型预测解释（如特征重要性），返回 JSON 可序列化对象（DataFrame / numpy 数组 / dict）；基类默认抛出 "Computing model explanations is not yet supported for this deployment target"（base.py）。
• 端点（Endpoint）管理族：create_endpoint、update_endpoint、delete_endpoint、list_endpoints、get_endpoint——分别对应端点的创建（阻塞至可用、返回含'name'的 dict）、更新、幂等删除、列出与查询。基类默认抛出 "Method is unimplemented in base client"（base.py）。

这些方法的公共参数约定：name为部署唯一名；config为目标专属配置字典；endpoint为可选的端点参数，并非所有目标都支持。

五、PredictionsResponse：统一预测响应封装

PredictionsResponse(dict)定义于 mlflow/deployments/init.py，以dict为基类，封装发送给 MLflow Model Server 的/invocations端点的评分请求所返回的预测与元数据。

5.1get_predictions(predictions_format="dataframe", dtype=None)

按指定格式取出预测结果：

• predictions_format="dataframe"（默认）：返回pandas.DataFrame。内部逻辑（init.py）：
  • 若predictions是字符串，包装为单行 DataFrame；
  • 若是 dict 且所有值均非一维 list-like，按"单行多列"构造（index=[0]）；
  • 否则直接由pd.DataFrame(data=predictions)构造。
• predictions_format="ndarray"：返回np.array(self["predictions"], dtype)，dtype为可选的 NumPy 数据类型。
• 其他格式：抛MlflowException（INVALID_PARAMETER_VALUE）。

5.2to_json(path=None)

返回 JSON 字符串表示；若指定path，则把 JSON 写入该文件路径并返回None（init.py）。

5.3from_json(json_str)（类方法）

从 JSON 字符串构造PredictionsResponse：

• JSON 解析失败 → 抛MlflowException("Predictions response contents are not valid JSON")；
• 解析结果不是 dict 或不含predictions字段 → 抛MlflowException指明必须为含'predictions'字段的字典（init.py）。

注：原 API 参考文档中autoclass指令以:exclude-members: from_json排除了from_json的渲染，但该方法是类上的真实公开能力，此处一并说明。

六、内置客户端之一：DatabricksDeploymentClient

用于与Databricks Serving Endpoints交互，定义于 mlflow/deployments/databricks/init.py。

6.1 认证与基本用法

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

from mlflow.deployments import get_deploy_client client = get_deploy_client("databricks") endpoints = client.list_endpoints() # 返回类似： # [{"name": "chat", "creator": "alice@company.com", "creation_timestamp": 0, # "last_updated_timestamp": 0, "state": {...}, "config": {...}, # "tags": [...], "id": "88fd3f75a0d24b0380ddc40484d7a31b"}]

6.2 与基类的差异：Deployment 方法全部未实现

DatabricksDeploymentClient覆写了create_deployment、update_deployment、delete_deployment、list_deployments、get_deployment五个部署管理方法，但全部直接raise NotImplementedError（databricks/init.py）。这是因为 Databricks 的模型服务以"端点"（Serving Endpoint）为管理单元，而非独立部署。因此该客户端的主打能力集中在端点管理与查询上。

6.3 推理：predict与predict_stream

client = get_deploy_client("databricks") # 普通推理：POST /api/2.0/serving-endpoints/{endpoint}/invocations response = client.predict( endpoint="chat", inputs={"messages": [{"role": "user", "content": "Hello!"}]}, ) # 返回 OpenAI 兼容格式的 dict（含 id/object/created/model/choices/usage 等字段） # 流式推理：请求体自动追加 stream=True 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) # 每个 chunk 是 OpenAI SSE 格式 "data: {...}" 解析出的 dict

实现要点（databricks/init.py）：

• predict通过_call_endpoint发 POST 请求，路由为{endpoint}/invocations，超时由环境变量MLFLOW_DEPLOYMENT_PREDICT_TIMEOUT（单请求）与MLFLOW_DEPLOYMENT_PREDICT_TOTAL_TIMEOUT（总重试时间）控制；
• predict_stream在请求体中注入stream=True，逐行解析响应——每行须为data: <value>格式，遇到"data: [DONE]"终止迭代；未知格式会抛MlflowException；
• 所有请求经http_request发送，带X-Databricks-Endpoints-API-Client头，重试码见 mlflow/deployments/constants.py：{429, 500, 502, 503}（特意移除超时，因为对代理 provider 的长超时重试通常意味着查询本身或模型配置有问题）。

6.4 端点管理 API

该客户端对端点提供了完整增删改查：

• create_endpoint(name=None, config=None, route_optimized=False)：创建 Serving Endpoint。推荐把name与route_optimized全部放进config字典（新风格，直接作为 API 请求体）；旧的"独立参数"风格会触发UserWarning弃用提示。示例（外部模型 gpt-4）：

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, }, }, )

• update_endpoint（已弃用）：按 config 是否为{"rate_limits": ...}决定走PUT .../rate-limits还是PUT .../config，弃用后应改用下面四个细分方法：
  • update_endpoint_config(endpoint, config)：PUT .../{endpoint}/config，更新 served_entities 等；
  • update_endpoint_tags(endpoint, config)：PATCH .../{endpoint}/tags，如{"add_tags": [{"key": "project", "value": "test"}]}；
  • update_endpoint_rate_limits(endpoint, config)：PUT .../{endpoint}/rate-limits，如{"rate_limits": [{"calls": 10, "key": "endpoint", "renewal_period": "minute"}]}；
  • update_endpoint_ai_gateway(endpoint, config)：PUT .../{endpoint}/ai-gateway，可配置usage_tracking_config与inference_table_config（推理结果落表）。
• delete_endpoint(endpoint)、list_endpoints()、get_endpoint(endpoint)：分别走DELETE /api/2.0/serving-endpoints/{endpoint}、GET /api/2.0/serving-endpoints、GET /api/2.0/serving-endpoints/{endpoint}，返回DatabricksEndpoint（继承自AttrDict的字典风格对象，支持endpoint.name属性访问）。

6.5DatabricksEndpoint

DatabricksEndpoint(AttrDict)（databricks/init.py）是一个字典风格对象，表示 Databricks Serving 端点，示例字段包括name、creator、creation_timestamp、last_updated_timestamp、state、config、tags、id，支持endpoint.name == "chat"这样的属性访问。

七、内置客户端之二：OpenAIDeploymentClient

用于与OpenAI / Azure OpenAI 端点交互，定义于 mlflow/deployments/openai/init.py。

7.1 认证与基本用法

export OPENAI_API_KEY=...

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

注意_check_openai_key()（openai/init.py）要求环境变量OPENAI_API_KEY必须存在，否则抛MlflowException（INVALID_PARAMETER_VALUE）。

7.2 能力边界：仅支持查询与模型列表

与 Databricks 客户端类似，该客户端把create_deployment、update_deployment、delete_deployment、list_deployments、get_deployment以及端点的创建/更新/删除方法全部实现为raise NotImplementedError。实际能力：

• predict(deployment_name, inputs, endpoint)：endpoint即模型名（如"gpt-4o-mini"），通过openaiSDK 的client.chat.completions.create(messages=inputs["messages"], model=endpoint).model_dump()完成。SDK 客户端构造逻辑（openai/init.py）：
  • 若环境变量OPENAI_API_TYPE为"azure"/"azure_ad"/"azuread"，构造AzureOpenAI客户端（使用OPENAI_API_BASE、OPENAI_API_VERSION、OPENAI_DEPLOYMENT_NAME）；
  • 否则构造标准OpenAI客户端，base_url来自OPENAI_API_BASE。
• list_endpoints()：请求GET https://api.openai.com/v1/models，返回可用模型列表；Azure OpenAI 下抛NotImplementedError。
• get_endpoint(endpoint)：请求GET https://api.openai.com/v1/models/{endpoint}，查询单个模型信息；Azure OpenAI 下抛NotImplementedError。

八、内置客户端之三：MlflowDeploymentClient（MLflow AI Gateway）

用于与MLflow AI Gateway交互——即通过mlflow gateway start --config-path ...启动的本地部署服务。定义于 mlflow/deployments/mlflow/init.py。

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() # [{"name": "chat", "endpoint_type": "llm/v1/chat", # "model": {"name": "gpt-4o-mini", "provider": "openai"}, # "endpoint_url": "http://localhost:5000/gateway/chat/invocations"}]

该客户端同样不实现部署/端点的增删改（全部NotImplementedError），主要提供对 Gateway 上已配置端点的**查询（predict）与列表/详情（list_endpoints、get_endpoint）**能力。其 HTTP 调用基于mlflow.utils.rest_utils.http_request，路由常量定义于 mlflow/deployments/server/constants.py（如 CRUD 基础路径、查询后缀等），并通过resolve_endpoint_url（utils.py）判断返回的是完整 URL 还是需要拼接在 base URL 之后。

重要前置条件：该客户端依赖openai等可选依赖，导入失败会被init.py 静默吞掉，因此使用前需确保安装了完整依赖集（如pip install mlflow[gateway]或mlflow[genai]）。

九、插件机制：DeploymentPlugins与 Entry Points 注册

部署目标通过基于 entry points 的插件注册机制动态发现，核心实现在 mlflow/deployments/plugin_manager.py。

9.1 注册流程

• PluginManager（@developer_stable）维护self._registry字典（目标名 → 插件对象），提供register(target_name, plugin_module)与register_entrypoints()两个方法（plugin_manager.py）。
• DeploymentPlugins在构造时以 entry points 组名"mlflow.deployments"调用register_entrypoints()，自动扫描所有安装了该 entry point 的包（plugin_manager.py）。
• 内置的 SageMaker 目标通过代码显式注册：plugin_store.register("sagemaker", "mlflow.sagemaker")（见 mlflow/deployments/interface.py）。

9.2 插件合法性校验（__getitem__）

当get_deploy_client(target_uri)通过plugin_store[target]取插件时（plugin_manager.py）：

1. 用parse_target_uri解析目标名并在注册表查找；找不到则抛MlflowException（RESOURCE_DOES_NOT_EXIST），提示安装合适的插件；
2. 通过 entry pointload()加载插件模块（加载失败抛RuntimeError）；
3. 校验插件接口完整性：模块必须同时提供target_help与run_local，且恰好一个BaseDeploymentClient的非抽象子类；缺失接口、没有子类或存在多个子类都会抛MlflowException。

这套校验保证了"任何注册的插件都符合 4.1 节的三要素契约"。

十、CLI 入口：mlflow deployments命令族

部署能力同时暴露为命令行接口，入口定义于 mlflow/deployments/cli.py。顶层命令组：

mlflow deployments --help

常用选项与子命令概览（源码 cli.py）：

• -t / --target <target-uri>（必填）：部署目标 URI，可配合mlflow deployments help --target-name <target-name>查看该目标的 URI 格式与 config 选项；命令组启动时会打印当前已安装的部署目标列表。
• --name：部署名称（部分子命令中可选）。
• -C / --config NAME=VALUE（可多次）：目标专属配置，_user_args_to_dict会按首个=拆分键值并拒绝重复参数（cli.py）。
• -I / --input-path（必填，predict 类命令）：输入预测负载文件路径，可为 JSON（Python dict）或 CSV（pandas DataFrame，需配--content-type csv）。
• -O / --output-path：结果输出 JSON 文件；缺省打印到 stdout。
• --endpoint：端点名（部分命令必填、部分可选）。

由此可知，mlflow deployments help -t <target>实际调用_target_help()展示目标专属帮助，其余子命令（create/update/delete/list/get/predict 等）最终都汇聚到get_deploy_client返回的客户端实例上执行。

十一、可靠性配置：重试码与超时环境变量

部署客户端在 HTTP 调用层面有专门的可靠性配置：

• 重试码：MLFLOW_DEPLOYMENT_CLIENT_REQUEST_RETRY_CODES = frozenset({429, 500, 502, 503})（constants.py），注释明确说明这是从 Tracking server 重试码中移除超时后的子集——因为对代理 provider 而言，长超时 + 重试通常意味着查询或模型参数配置有问题，不应盲目重试。
• 超时环境变量（见 mlflow/environment_variables.py 相关定义）：MLFLOW_DEPLOYMENT_PREDICT_TIMEOUT（单次预测请求超时）、MLFLOW_DEPLOYMENT_PREDICT_TOTAL_TIMEOUT（含重试的总超时）、MLFLOW_DEPLOYMENT_CLIENT_HTTP_REQUEST_TIMEOUT（部署客户端 HTTP 请求超时）、MLFLOW_HTTP_REQUEST_TIMEOUT（通用 HTTP 请求超时兜底）。Databricks 客户端在每次调用前会执行validate_deployment_timeout_config(timeout, retry_timeout_seconds)校验配置合法性。

十二、写在最后：快速上手路径与源码索引

围绕mlflow.deployments的实践建议：

1. 只想调用已有目标：get_deploy_client(target_uri)起步，参考 interface.py 的示例；Databricks 用户看 databricks/init.py，OpenAI 用户看 openai/init.py，本地 Gateway 用户看 mlflow/init.py。
2. 想管理预测响应：使用PredictionsResponse的get_predictions（dataframe/ndarray 两种格式）、to_json、from_json，实现见init.py。
3. 想写自己的部署插件：以BaseDeploymentClient为基类，实现 4.1 节的三要素（唯一子类 +run_local+target_help），以mlflow.deployments为 entry points 组名注册，校验规则见 plugin_manager.py。

相关测试可进一步参考 tests/deployments 目录与 tests/gateway，其中包含对上述客户端行为与插件校验逻辑的覆盖用例，可作为理解各方法真实语义的补充证据。

【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow