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__可以看出模块导出的全部公开符号符号类型作用get_deploy_client函数获取指定目标target的部署客户端实例run_local函数将模型在本地部署起来用于测试BaseDeploymentClient类部署客户端的抽象基类定义标准 API 契约DatabricksDeploymentClient类Databricks Serving Endpoints 内置客户端OpenAIDeploymentClient类OpenAI / Azure OpenAI 内置客户端MlflowDeploymentClient类MLflow AI Gateway本地部署服务内置客户端DatabricksEndpoint类字典风格的 Databricks Serving 端点对象PredictionsResponse类评分请求如/invocationsREST 调用的响应封装get_deployments_target/set_deployments_target函数读取 / 设置全局部署目标值得注意的实现细节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否则抛出MlflowExceptionINVALID_PARAMETER_VALUE。get_deployments_target()的取值优先级先返回模块内_deployments_target即代码中设置过的值若未设置则回退读取环境变量MLFLOW_DEPLOYMENTS_TARGET见 mlflow/environment_variables.py 中定义两者皆无则抛MlflowException提示设置方式。2.2 环境变量方式MLFLOW_DEPLOYMENTS_TARGETexport MLFLOW_DEPLOYMENTS_TARGEThttp://localhost:50002.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_uriNone)定义在 mlflow/deployments/interface.py返回BaseDeploymentClient的子类实例暴露用于部署模型的标准 API。其目标解析优先级显式传入target_uri参数未传时尝试get_deployments_target()即代码设置的全局目标或MLFLOW_DEPLOYMENTS_TARGET环境变量若两者皆无打印提示日志并返回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( targetredisai, # 部署目标 namespamDetector, # 部署名称 model_uriruns:/someRunId/myModel, # 模型 URI flavorNone, # (可选) 模型 flavor缺省自动选择 configNone, # (可选) 目标特有的配置字典 )按 interface.pyrun_local()直接调用插件的run_local方法。其签名与create_deployment非常相似因为二者逻辑上做的是同类操作。注意本地部署的模型无法被其他部署管理 APIupdate_deployment、delete_deployment等管理它只服务于测试目的。3.2_target_help()目标专属帮助_target_help(target)interface.py返回目标专属的详细文档字符串当用户执行mlflow deployments help -t target-nameCLI 时展示内容包括传给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一个合法的部署插件模块必须实现恰好一个继承自BaseDeploymentClient的客户端类暴露管理部署的主要用户 API模块级run_local(target, name, model_uri, flavorNone, configNone)函数用于本地测试部署模块级target_help()函数返回描述目标 URI 格式与部署 config 的帮助消息。run_local与target_help在 base.py 中仅作为文档占位存在——直接调用会抛NotImplementedError真正的实现必须在插件模块的顶层命名空间中以plugin_module.run_local/plugin_module.target_help方式可调用。4.2 部署管理抽象方法必须实现以下方法均为abc.abstractmethod插件必须实现方法签名要点语义create_deployment(name, model_uri, flavorNone, configNone, endpointNone)部署模型。默认阻塞直到部署完成可进行推理同名冲突抛MlflowException或远程部署的HTTPError返回包含name键的 dictupdate_deployment(name, model_uriNone, flavorNone, configNone, endpointNone)更新部署。可更新模型 URI、flavor此时必须同时给出 model_uri及目标专属属性默认阻塞至更新完成delete_deployment(name, configNone, endpointNone)删除部署应幂等对不存在的部署重试也不应失败list_deployments(endpointNone)返回不分页的部署列表每个 dict 保证含name键插件也可返回带deployments字段及next_page_token的分页字典get_deployment(name, endpointNone)返回描述指定部署的 dict不存在时抛MlflowException或HTTPErrorpredict(deployment_nameNone, inputsNone, endpointNone)用指定部署对输入做推理输入/输出类型与mlflow pyfunc predict一致返回PredictionsResponse实例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 targetbase.py。端点Endpoint管理族create_endpoint、update_endpoint、delete_endpoint、list_endpoints、get_endpoint——分别对应端点的创建阻塞至可用、返回含name的 dict、更新、幂等删除、列出与查询。基类默认抛出 Method is unimplemented in base clientbase.py。这些方法的公共参数约定name为部署唯一名config为目标专属配置字典endpoint为可选的端点参数并非所有目标都支持。五、PredictionsResponse统一预测响应封装PredictionsResponse(dict)定义于 mlflow/deployments/init.py以dict为基类封装发送给 MLflow Model Server 的/invocations端点的评分请求所返回的预测与元数据。5.1get_predictions(predictions_formatdataframe, dtypeNone)按指定格式取出预测结果predictions_formatdataframe默认返回pandas.DataFrame。内部逻辑init.py若predictions是字符串包装为单行 DataFrame若是 dict 且所有值均非一维 list-like按单行多列构造index[0]否则直接由pd.DataFrame(datapredictions)构造。predictions_formatndarray返回np.array(self[predictions], dtype)dtype为可选的 NumPy 数据类型。其他格式抛MlflowExceptionINVALID_PARAMETER_VALUE。5.2to_json(pathNone)返回 JSON 字符串表示若指定path则把 JSON 写入该文件路径并返回Noneinit.py。5.3from_json(json_str)类方法从 JSON 字符串构造PredictionsResponseJSON 解析失败 → 抛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: alicecompany.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 NotImplementedErrordatabricks/init.py。这是因为 Databricks 的模型服务以端点Serving Endpoint为管理单元而非独立部署。因此该客户端的主打能力集中在端点管理与查询上。6.3 推理predict与predict_streamclient get_deploy_client(databricks) # 普通推理POST /api/2.0/serving-endpoints/{endpoint}/invocations response client.predict( endpointchat, inputs{messages: [{role: user, content: Hello!}]}, ) # 返回 OpenAI 兼容格式的 dict含 id/object/created/model/choices/usage 等字段 # 流式推理请求体自动追加 streamTrue chunk_iter client.predict_stream( endpointdatabricks-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.pypredict通过_call_endpoint发 POST 请求路由为{endpoint}/invocations超时由环境变量MLFLOW_DEPLOYMENT_PREDICT_TIMEOUT单请求与MLFLOW_DEPLOYMENT_PREDICT_TOTAL_TIMEOUT总重试时间控制predict_stream在请求体中注入streamTrue逐行解析响应——每行须为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(nameNone, configNone, route_optimizedFalse)创建 Serving Endpoint。推荐把name与route_optimized全部放进config字典新风格直接作为 API 请求体旧的独立参数风格会触发UserWarning弃用提示。示例外部模型 gpt-4endpoint 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.5DatabricksEndpointDatabricksEndpoint(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( endpointgpt-4o-mini, inputs{messages: [{role: user, content: Hello!}]}, )注意_check_openai_key()openai/init.py要求环境变量OPENAI_API_KEY必须存在否则抛MlflowExceptionINVALID_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(messagesinputs[messages], modelendpoint).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。八、内置客户端之三MlflowDeploymentClientMLflow AI Gateway用于与MLflow AI Gateway交互——即通过mlflow gateway start --config-path ...启动的本地部署服务。定义于 mlflow/deployments/mlflow/init.py。mlflow gateway start --config-path path/to/config.yamlfrom 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_urlutils.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 注册流程PluginManagerdeveloper_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用parse_target_uri解析目标名并在注册表查找找不到则抛MlflowExceptionRESOURCE_DOES_NOT_EXIST提示安装合适的插件通过 entry pointload()加载插件模块加载失败抛RuntimeError校验插件接口完整性模块必须同时提供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 NAMEVALUE可多次目标专属配置_user_args_to_dict会按首个拆分键值并拒绝重复参数cli.py。-I / --input-path必填predict 类命令输入预测负载文件路径可为 JSONPython dict或 CSVpandas 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的实践建议只想调用已有目标get_deploy_client(target_uri)起步参考 interface.py 的示例Databricks 用户看 databricks/init.pyOpenAI 用户看 openai/init.py本地 Gateway 用户看 mlflow/init.py。想管理预测响应使用PredictionsResponse的get_predictionsdataframe/ndarray 两种格式、to_json、from_json实现见init.py。想写自己的部署插件以BaseDeploymentClient为基类实现 4.1 节的三要素唯一子类 run_localtarget_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
MLflow 部署 API 完全指南：从 get_deploy_client 到自定义部署插件（mlflow.deployments 模块深度解析）
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__可以看出模块导出的全部公开符号符号类型作用get_deploy_client函数获取指定目标target的部署客户端实例run_local函数将模型在本地部署起来用于测试BaseDeploymentClient类部署客户端的抽象基类定义标准 API 契约DatabricksDeploymentClient类Databricks Serving Endpoints 内置客户端OpenAIDeploymentClient类OpenAI / Azure OpenAI 内置客户端MlflowDeploymentClient类MLflow AI Gateway本地部署服务内置客户端DatabricksEndpoint类字典风格的 Databricks Serving 端点对象PredictionsResponse类评分请求如/invocationsREST 调用的响应封装get_deployments_target/set_deployments_target函数读取 / 设置全局部署目标值得注意的实现细节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否则抛出MlflowExceptionINVALID_PARAMETER_VALUE。get_deployments_target()的取值优先级先返回模块内_deployments_target即代码中设置过的值若未设置则回退读取环境变量MLFLOW_DEPLOYMENTS_TARGET见 mlflow/environment_variables.py 中定义两者皆无则抛MlflowException提示设置方式。2.2 环境变量方式MLFLOW_DEPLOYMENTS_TARGETexport MLFLOW_DEPLOYMENTS_TARGEThttp://localhost:50002.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_uriNone)定义在 mlflow/deployments/interface.py返回BaseDeploymentClient的子类实例暴露用于部署模型的标准 API。其目标解析优先级显式传入target_uri参数未传时尝试get_deployments_target()即代码设置的全局目标或MLFLOW_DEPLOYMENTS_TARGET环境变量若两者皆无打印提示日志并返回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( targetredisai, # 部署目标 namespamDetector, # 部署名称 model_uriruns:/someRunId/myModel, # 模型 URI flavorNone, # (可选) 模型 flavor缺省自动选择 configNone, # (可选) 目标特有的配置字典 )按 interface.pyrun_local()直接调用插件的run_local方法。其签名与create_deployment非常相似因为二者逻辑上做的是同类操作。注意本地部署的模型无法被其他部署管理 APIupdate_deployment、delete_deployment等管理它只服务于测试目的。3.2_target_help()目标专属帮助_target_help(target)interface.py返回目标专属的详细文档字符串当用户执行mlflow deployments help -t target-nameCLI 时展示内容包括传给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一个合法的部署插件模块必须实现恰好一个继承自BaseDeploymentClient的客户端类暴露管理部署的主要用户 API模块级run_local(target, name, model_uri, flavorNone, configNone)函数用于本地测试部署模块级target_help()函数返回描述目标 URI 格式与部署 config 的帮助消息。run_local与target_help在 base.py 中仅作为文档占位存在——直接调用会抛NotImplementedError真正的实现必须在插件模块的顶层命名空间中以plugin_module.run_local/plugin_module.target_help方式可调用。4.2 部署管理抽象方法必须实现以下方法均为abc.abstractmethod插件必须实现方法签名要点语义create_deployment(name, model_uri, flavorNone, configNone, endpointNone)部署模型。默认阻塞直到部署完成可进行推理同名冲突抛MlflowException或远程部署的HTTPError返回包含name键的 dictupdate_deployment(name, model_uriNone, flavorNone, configNone, endpointNone)更新部署。可更新模型 URI、flavor此时必须同时给出 model_uri及目标专属属性默认阻塞至更新完成delete_deployment(name, configNone, endpointNone)删除部署应幂等对不存在的部署重试也不应失败list_deployments(endpointNone)返回不分页的部署列表每个 dict 保证含name键插件也可返回带deployments字段及next_page_token的分页字典get_deployment(name, endpointNone)返回描述指定部署的 dict不存在时抛MlflowException或HTTPErrorpredict(deployment_nameNone, inputsNone, endpointNone)用指定部署对输入做推理输入/输出类型与mlflow pyfunc predict一致返回PredictionsResponse实例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 targetbase.py。端点Endpoint管理族create_endpoint、update_endpoint、delete_endpoint、list_endpoints、get_endpoint——分别对应端点的创建阻塞至可用、返回含name的 dict、更新、幂等删除、列出与查询。基类默认抛出 Method is unimplemented in base clientbase.py。这些方法的公共参数约定name为部署唯一名config为目标专属配置字典endpoint为可选的端点参数并非所有目标都支持。五、PredictionsResponse统一预测响应封装PredictionsResponse(dict)定义于 mlflow/deployments/init.py以dict为基类封装发送给 MLflow Model Server 的/invocations端点的评分请求所返回的预测与元数据。5.1get_predictions(predictions_formatdataframe, dtypeNone)按指定格式取出预测结果predictions_formatdataframe默认返回pandas.DataFrame。内部逻辑init.py若predictions是字符串包装为单行 DataFrame若是 dict 且所有值均非一维 list-like按单行多列构造index[0]否则直接由pd.DataFrame(datapredictions)构造。predictions_formatndarray返回np.array(self[predictions], dtype)dtype为可选的 NumPy 数据类型。其他格式抛MlflowExceptionINVALID_PARAMETER_VALUE。5.2to_json(pathNone)返回 JSON 字符串表示若指定path则把 JSON 写入该文件路径并返回Noneinit.py。5.3from_json(json_str)类方法从 JSON 字符串构造PredictionsResponseJSON 解析失败 → 抛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: alicecompany.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 NotImplementedErrordatabricks/init.py。这是因为 Databricks 的模型服务以端点Serving Endpoint为管理单元而非独立部署。因此该客户端的主打能力集中在端点管理与查询上。6.3 推理predict与predict_streamclient get_deploy_client(databricks) # 普通推理POST /api/2.0/serving-endpoints/{endpoint}/invocations response client.predict( endpointchat, inputs{messages: [{role: user, content: Hello!}]}, ) # 返回 OpenAI 兼容格式的 dict含 id/object/created/model/choices/usage 等字段 # 流式推理请求体自动追加 streamTrue chunk_iter client.predict_stream( endpointdatabricks-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.pypredict通过_call_endpoint发 POST 请求路由为{endpoint}/invocations超时由环境变量MLFLOW_DEPLOYMENT_PREDICT_TIMEOUT单请求与MLFLOW_DEPLOYMENT_PREDICT_TOTAL_TIMEOUT总重试时间控制predict_stream在请求体中注入streamTrue逐行解析响应——每行须为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(nameNone, configNone, route_optimizedFalse)创建 Serving Endpoint。推荐把name与route_optimized全部放进config字典新风格直接作为 API 请求体旧的独立参数风格会触发UserWarning弃用提示。示例外部模型 gpt-4endpoint 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.5DatabricksEndpointDatabricksEndpoint(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( endpointgpt-4o-mini, inputs{messages: [{role: user, content: Hello!}]}, )注意_check_openai_key()openai/init.py要求环境变量OPENAI_API_KEY必须存在否则抛MlflowExceptionINVALID_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(messagesinputs[messages], modelendpoint).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。八、内置客户端之三MlflowDeploymentClientMLflow AI Gateway用于与MLflow AI Gateway交互——即通过mlflow gateway start --config-path ...启动的本地部署服务。定义于 mlflow/deployments/mlflow/init.py。mlflow gateway start --config-path path/to/config.yamlfrom 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_urlutils.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 注册流程PluginManagerdeveloper_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用parse_target_uri解析目标名并在注册表查找找不到则抛MlflowExceptionRESOURCE_DOES_NOT_EXIST提示安装合适的插件通过 entry pointload()加载插件模块加载失败抛RuntimeError校验插件接口完整性模块必须同时提供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 NAMEVALUE可多次目标专属配置_user_args_to_dict会按首个拆分键值并拒绝重复参数cli.py。-I / --input-path必填predict 类命令输入预测负载文件路径可为 JSONPython dict或 CSVpandas 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的实践建议只想调用已有目标get_deploy_client(target_uri)起步参考 interface.py 的示例Databricks 用户看 databricks/init.pyOpenAI 用户看 openai/init.py本地 Gateway 用户看 mlflow/init.py。想管理预测响应使用PredictionsResponse的get_predictionsdataframe/ndarray 两种格式、to_json、from_json实现见init.py。想写自己的部署插件以BaseDeploymentClient为基类实现 4.1 节的三要素唯一子类 run_localtarget_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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
