实用笔记：使用 Azure ML 构建并部署自定义模型，随后进行连接配置

本指南将逐步构建从原材料到可运行系统的完整流程，内容包括：使用 Azure ML 构建并部署自定义模型，再将其集成到 Foundry Agent 中。重点在于可操作的步骤、明确的检查点以及可直接放入代码仓库的代码，无需猜测其用途。 在概述阶段，应在修改代码之前明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏的状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。

先建立思维模型

在完成“思维模型”第一阶段的工作时，首先写下相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向单一的责任模块，而非复杂的流程链。 缓存系统中稳定的指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。

前提条件

在完成前置条件阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。

pip install azure-ai-ml azure-ai-projects azure-identity

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential

ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="<subscription-id>",
    resource_group_name="<resource-group>",
    workspace_name="<aml-workspace-name>",
)

步骤1：训练模型

在完成第一步的训练阶段时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外费用。 缓存稳定的系统指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。 在完成第一步的训练阶段时，首先需明确相关约定：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。

from azure.ai.ml import command, Input, Output

train_job = command(
    code="./src",
    command="python train.py --data ${{inputs.training_data}} --model_output ${{outputs.model_output}}",
    inputs={"training_data": Input(type="uri_folder", path="azureml://datastores/workspaceblobstore/paths/churn-training/")},
    outputs={"model_output": Output(type="uri_folder")},
    environment="azureml://registries/azureml/environments/sklearn-1.5/labels/latest",
    compute="cpu-cluster",
    display_name="churn-model-training",
)
returned_job = ml_client.jobs.create_or_update(train_job)
ml_client.jobs.stream(returned_job.name)

第二步：注册已训练的模型

将第二步的注册阶段视为可测量的表面来处理时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一的责任主体，而非复杂的流程链。 为每轮及每次会话设定预算额度。智能工具会大量消耗上下文资源；设置上限可避免演示过程变成意外的费用账单。

from azure.ai.ml.entities import Model
from azure.ai.ml.constants import AssetTypes

model = ml_client.models.create_or_update(
    Model(
        path=f"azureml://jobs/{returned_job.name}/outputs/model_output",
        name="churn-classifier",
        type=AssetTypes.MLFLOW_MODEL,
        description="Customer churn classifier, trained on 18 months of account history.",
    )
)

第三步：在托管端点后部署它

第三步的部署阶段若被视为可度量的工作面，效果会最佳。在扩大范围之前，需记录一份理想的运行结果、一个故障案例以及回滚说明。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现悄无声息的半完成状态。 需为每轮及每次会话设定预算额度。智能工具往往会过度扩展上下文，设置上限可避免演示过程变成意外的费用账单。

from azure.ai.ml.entities import ManagedOnlineEndpoint, ManagedOnlineDeployment

endpoint = ManagedOnlineEndpoint(name="churn-endpoint", auth_mode="key")
ml_client.online_endpoints.begin_create_or_update(endpoint).result()
deployment = ManagedOnlineDeployment(
    name="blue",
    endpoint_name="churn-endpoint",
    model=model,
    instance_type="Standard_DS3_v2",
    instance_count=1,
)
ml_client.online_deployments.begin_create_or_update(deployment).result()
endpoint.traffic = {"blue": 100}
ml_client.online_endpoints.begin_create_or_update(endpoint).result()

第四步：在有其他任何操作介入之前确认其正常运行

第4步的确认阶段在被视为可度量的对象时效果最佳。在扩大范围之前，需记录一份理想流程的完整日志、一个失败案例以及回滚说明。在功能结果旁还需记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。应为每轮操作和每次会话设定令牌预算，因为智能代理工具会大量消耗上下文资源，设置上限能防止演示环境变成意外收费的源头。

import json

test_input = {"input_data": {"columns": ["tenure_months", "monthly_spend", "support_tickets"], "data": [[14, 89.50, 3]]}}
response = ml_client.online_endpoints.invoke(
    endpoint_name="churn-endpoint",
    request_file=None,
    deployment_name="blue",
    input_data=json.dumps(test_input),
)
print(response)

第4步的确认阶段在被视为可度量的对象时效果最佳。在扩大范围之前，需记录一份理想流程的完整日志、一个失败案例以及回滚说明。应同时记录正常流程和故障恢复流程的详细信息，重试机制、人工审核环节以及死信处理都属于产品功能的一部分，而非后续需要补充的内容。

第5步：将端点封装为Foundry代理函数工具

在第五步中，应在修改代码之前先定义阶段、输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相比冗长的脚本，更应采用小型且可测试的单元。当某一步骤失败时，故障原因应能明确指向某个具体的责任模块，而非整个复杂的流程。 如果后续步骤是代码或工具调用，相比自由形式的文字描述，带有架构验证的结构化输出更为合适。

import os
import requests
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, Tool, FunctionTool
from azure.identity import DefaultAzureCredential

def get_churn_score(tenure_months: int, monthly_spend: float, support_tickets: int) -> dict:
    payload = {"input_data": {"columns": ["tenure_months", "monthly_spend", "support_tickets"], "data": [[tenure_months, monthly_spend, support_tickets]]}}
    resp = requests.post(
        "https://churn-endpoint.<region>.inference.ml.azure.com/score",
        headers={"Authorization": f"Bearer {os.environ['AML_ENDPOINT_KEY']}", "Content-Type": "application/json"},
        json=payload,
        timeout=10,
    )
    resp.raise_for_status()
    return {"churn_probability": resp.json()[0]}
func_tool = FunctionTool(
    name="get_churn_score",
    description="Predict churn probability for a customer given tenure, spend, and support ticket history.",
    parameters={
        "type": "object",
        "properties": {
            "tenure_months": {"type": "integer", "description": "How many months the customer has been active."},
            "monthly_spend": {"type": "number", "description": "Average monthly spend in dollars."},
            "support_tickets": {"type": "integer", "description": "Number of support tickets in the last 90 days."},
        },
        "required": ["tenure_months", "monthly_spend", "support_tickets"],
        "additionalProperties": False,
    },
    strict=True,
)
project = AIProjectClient(endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], credential=DefaultAzureCredential())
tools: list[Tool] = [func_tool]
agent = project.agents.create_version(
    agent_name="retention-agent",
    definition=PromptAgentDefinition(
        model="gpt-4.1-mini",
        instructions="Help the team assess churn risk. Call get_churn_score whenever specific customer numbers are provided.",
        tools=tools,
    ),
)

第六步：端到端运行

在第六步的运行阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 当下一步是代码编写或工具调用时，优先采用具有架构验证的结构化输出，而非自由形式的文字描述。

import json
from openai.types.responses.response_input_param import FunctionCallOutput

openai_client = project.get_openai_client()
conversation = openai_client.conversations.create()
response = openai_client.responses.create(
    input="A customer's been with us 14 months, spends about $90/month, and filed 3 tickets recently. Churn risk?",
    conversation=conversation.id,
    extra_body={"agent_reference": {"name": agent.name, "type": "agent_reference"}},
)
for item in response.output:
    if item.type == "function_call" and item.name == "get_churn_score":
        result = get_churn_score(**json.loads(item.arguments))
        follow_up = openai_client.responses.create(
            input=[FunctionCallOutput(type="function_call_output", call_id=item.call_id, output=json.dumps(result))],
            conversation=conversation.id,
            extra_body={"agent_reference": {"name": agent.name, "type": "agent_reference"}},
        )
        print(follow_up.output_text)

其在整体架构中的位置

在确定该步骤在流程中的位置时，应在修改代码之前明确输入参数、负责该步骤的人员以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。 当下一步操作为编写代码或调用工具时，应优先使用具有架构验证的结构化输出，而非自由形式的文字描述。 在确定该步骤在流程中的位置时，应在修改代码之前明确输入参数、负责该步骤的人员以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理方式都是产品本身的一部分，而非后续需要补充的内容。

部署前的生产环境考量

在处理部署前的生产环境相关事项时，首先需明确合同条款：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非复杂的流程链。 请缓存稳定的系统指令和工具结构。重复发送相同的开头信息是导致资源浪费的常见原因。

这会给你带来什么

在处理“这会将你带到何处”这一阶段时，首先写下契约内容：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是造成资源浪费的常见原因。

参考资料

在参考阶段工作时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境过渡到共享环境时出现意外账单。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。 在参考阶段工作时，首先需明确合同条款：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。

操作检查清单

将操作检查清单视为可度量的基准，这样会更有效。在扩大范围之前，先记录一份最佳实践案例、一个故障实例以及回滚说明。

将配置与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放，以便操作人员无需查看全部内容即可进行审计。

为每轮操作和每次会话设定预算上限。智能工具往往会大量消耗资源，设置硬性限制可以避免演示过程变成意外账单。

对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。

编写简短的操作手册：说明如何更换密钥、如何清空队列以及如何回滚上一次的数据导入操作。

将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，并拒绝默许的半完成状态。

在升级整个系统之前，先冻结各版本，为关键流程记录标准输出文本，并确认回滚步骤。共享环境需要设置访问速率限制、租户验证机制，以及明确的密钥轮换负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。

关于2117731bf26b的批注：请将服务提供商密钥存放在仓库之外，设定单次会话的令牌上限，并将输出文本与评估用文件一起保存，以便后续更换模型时仍能保持对比性。