在 Azure ML 中训练自定义模型，并从 Foundry Agent 调用该模型

Foundry的模型目录虽然丰富，但仍有一些任务不在其覆盖范围内。试想一下，从企业自身的客户记录中预测客户流失，或根据其交易历史判断欺诈行为——这些都是针对私有表格数据的传统机器学习任务，而微软的解决方案是Azure Machine Learning，这是一款独立的产品，拥有专属的SDK、工作空间以及独特的模型部署方式。

以下的实现流程涵盖了整个路径：使用Azure ML SDK训练模型，将其注册，通过托管端点提供模型服务，然后将该端点作为函数工具暴露给Foundry代理。这样，代理就可以在对话过程中像调用其他工具一样调用这个自定义模型。

两个平台的交汇点

这两种产品运行在不同的运行时环境中，但属于同一个生态系统，微软一直在努力缩小它们之间的差距。明确了解这种差距所在之处有助于更轻松地理解整个构建过程。其中涉及四个部分：

• 命令作业。训练过程在此执行。SDK会指向一个脚本、一个计算目标以及一些输入数据，然后将代码作为可追踪、可复现的作业来执行。
• 模型注册表。训练完成的模型会转化为一个有名称、有版本号的工件，该工件独立于生成它的作业而存在，因此无需重复训练即可将某个版本部署到生产环境。
• 托管在线端点。它负责提供已注册模型的预测结果。这是真正的基础设施，处于持续运行状态或可自动扩展，拥有自己的URL、认证机制和成本结构，且位于Azure ML工作区中，而非Foundry中。

Foundry FunctionTool。 它起到了桥梁作用。Foundry既不知道也不关心该工具背后的函数实际上调用了Azure ML的端点。对代理而言，它只是一个名称、一个结构以及一个结果而已。

前提条件

1. 一个Azure ML工作区和一个Foundry项目。它们可以位于同一个资源组中，也可以位于不同的资源组中，因为这里没有任何要求必须使用共享基础设施。
2. 安装了相应SDK的Python 3.9或更高版本：

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

MLClient用于对工作区进行身份验证并定位目标：

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

训练作为命令任务

训练步骤属于命令任务。它接收一个训练脚本——在本例中为普通的 scikit-learn 分类器——并在受管理的计算环境中执行该脚本：

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)

调用 ml_client.jobs.stream 会持续等待并输出日志，直到训练完成。在真正会运行的脚本中应优先使用此方法。如果任务在后台默默失败，就会让人难以察觉到流水线出现了问题。

注册模型

注册功能会将任务的输出转换为带有名称和版本号的成果，从而可以独立引用：

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.",
    )
)

刻意选择MLFLOW_MODEL作为类型是有原因的。如果训练脚本通过MLflow的自动记录功能来保存模型，管理型端点就可以使用内置的评分容器来部署该模型，而无需额外的自定义score.py推理脚本。这样既能节省时间，也能让训练脚本从一开始就围绕这一机制进行设计。

在端点后部署

通过端点与部署功能，即可将模型上线：

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()

部署名称 blue 并非必须，但建议保留它，因为这有助于在新模型版本首次推出时养成良好习惯。在 blue 旁边创建一个 green 部署，将流量在两者之间分配，等新版本获得足够信任后再完全切换。直接覆盖 blue 并寄希望于最佳结果则是风险较高的选择。

在连接其他任何组件之前，先直接获取真实的预测结果：

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)

通过代理的函数调用循环来调试出问题的端点比直接调试要困难得多，因此请先确认其能够正常工作。

将端点连接到 Foundry 代理

代理本身从不直接调用 Azure ML，只有当代理发起请求时，应用程序代码才会进行调用。配置过程中需要声明一个 FunctionTool，创建代理，并定义用于向端点发送 HTTP 请求的函数：

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)

模型会判断某个问题是否需要调用get_churn_score函数。如果需要，应用程序代码就会向Azure ML发起HTTP请求，返回的结果会传回同一对话中，以便模型完成回答。在代理定义或Responses API调用中，没有任何内容依赖于该函数背后的具体实现方式。它既可以查询数据库，也可以调用Lambda函数，或是连接到其他Azure ML工作区，就像这里所示的那样。

两种资源，两种认证边界

SDK文档中有一个要点被忽略了。Azure ML中的托管端点与Foundry中的项目端点分别是处于两套独立认证体系之后的两个资源，当前架构中没有任何机制将它们连接起来。Azure ML为工作区拥有自己的密钥或托管身份，而Foundry则使用不同的凭证。声明FunctionTool并不会在两者之间建立信任关系，它只是为模型指定名称和结构，以便让代码向相应服务发起请求。所有针对Azure ML的认证操作都在你自己的函数内部完成，就像调用其他外部API时一样。

这其实是一个优势。Azure ML相关部分可以独立于Foundry进行管理、加密和密钥轮换，甚至可以由其他团队负责，而且双方都不需要拥有对方资源的写入权限。

生产环境注意事项

• 函数调用在10分钟后失效。必须在窗口关闭前将输出结果传回对话界面，否则运行将会失败。如果向Azure ML发起的HTTP请求速度较慢，则应将其视为严格的截止时间并提前做好相应安排。
• 在原型之外采用托管身份机制。本地环境中将密钥存储在环境变量中尚可，但在无人监控的情况下则存在安全隐患。应为运行工具代码的身份授予AzureML Data Scientist角色，或创建一个范围仅限于该端点而非整个工作区的更精细的自定义角色。
• 逐步推出新模型。应在现有部署旁边启动新的版本，然后分步将流量迁移过去。这是唯一能在所有实际请求依赖新模型之前发现其问题的方法。

请保持 strict=True 的设置不变。 严格的模式会阻止模型向 get_churn_score 发送不完整或格式错误的参数，否则这些错误会在函数内部引发异常，而不会在调用之前就被发现。

空闲状态也会产生费用。 即使没有流量，全天运行一个实例的托管在线端点也会被计费，而无服务器模型调用则不会。对于流量较低或波动较大的场景，可以考虑使用零扩展或批量处理端点。

数据偏移往往悄无声息。 当实际数据偏离训练数据分布时，当前系统不会自动重新训练模型。应将每次 get_churn_score 调用的输入和输出记录在便于查看的地方，并定期重新检查训练数据，而不仅仅是在预测结果明显变差之后才处理。

选择合适的平台

对于大多数生成式 AI 工作而言，Foundry 的产品系列是理想工具，而那些没有自有训练数据的团队可能永远不需要其他选择。一旦需求转变为基于自身历史数据对客户进行预测，Azure ML 则更为合适。否则，语言模型就会被用于其原本并非为之设计的任务，或者需要手动构建服务基础设施。其实实现起来很简单：一次训练任务、一个已注册的模型、一个托管端点以及一个 FunctionTool 即可。这两个平台保持独立，各自发挥其优势。