Ein benutzerdefiniertes Modell in Azure ML trainieren und es von einem Foundry Agent aufrufen
Trainieren, registrieren und bereitstellen Sie ein Churn-Modell auf einem von Azure ML verwalteten Endpunkt, und stellen Sie es anschließend als FunctionTool einem Foundry-Agenten zur Verfügung, inklusive Authentifizierungs- und Produktionshinweisen.
Quellen: Building and Serving a Custom Model with Azure ML, Then Wiring It Into a Foundry Agent
Das Modellkatalog von Foundry ist umfangreich, doch es gibt einige Aufgaben, die darin nicht enthalten sind. Stellen Sie sich vor, man möchte den Kundenabwanderungsrisiko anhand der eigenen Kundendaten eines Unternehmens vorhersagen oder Betrug anhand der eigenen Transaktionshistorie bewerten. Das sind herkömmliche maschinelles-Lernen-Aufgaben mit privaten tabellarischen Daten, und Microsofts Lösung dafür ist Azure Machine Learning – ein eigenständiges Produkt mit einem separaten SDK, einer eigenen Arbeitsumgebung sowie einer eigenen Methode zum Bereitstellen von Modellen.
Der folgende Ablauf umfasst den gesamten Prozess: Ein Modell wird mit dem Azure ML SDK trainiert, registriert, über einen verwalteten Endpunkt bereitgestellt und anschließend diesem Endpunkt als Funktionswerkzeug einem Foundry-Agenten zur Verfügung gestellt. Der Agent kann das benutzerdefinierte Modell dann mitten in einer Konversation aufrufen, genauso wie er jedes andere Werkzeug aufrufen würde.
Wo sich die beiden Plattformen treffen
Die beiden Produkte laufen unter separaten Laufzeiten, gehören aber zum selben Ökosystem, und Microsoft hat daran gearbeitet, die Kluft zwischen ihnen gering zu halten. Wenn man genau weiß, wo diese Kluft liegt, ist es einfacher, den Build nachzuvollziehen. Es sind vier Komponenten beteiligt:
- Befehlsjob. Hier finden die Trainingsläufe statt. Das SDK ist auf ein Skript, ein Rechenziel sowie einige Eingaben ausgerichtet und führt den Code als nachvollziehbarer, trackierter Job aus.
- Modellregister. Ein trainiertes Modell wird zu einem benannten, versionierten Artefakt, das unabhängig vom Job, der es erzeugt hat, existiert. So kann eine Version ohne Wiederholung des Trainingslaufs in die Produktion überführt werden.
- Gemanagter Online-Endpunkt. Dieser stellt Vorhersagen aus einem registrierten Modell bereit. Es handelt sich dabei um echte Infrastruktur, die ständig verfügbar ist oder automatisch skaliert wird, mit eigener URL, Authentifizierung und Kosten. Sie befindet sich innerhalb des Azure ML-Arbeitsbereichs, nicht in Foundry.
FunctionTool. Dies ist die Brücke. Foundry weiß weder etwas davon noch kümmert es ihn, dass die Funktion hinter dem Tool einen Azure ML-Endpunkt aufruft. Für den Agenten handelt es sich dabei um einen Namen, ein Schema und ein Ergebnis.Voraussetzungen
- Ein Azure ML-Arbeitsbereich und ein Foundry-Projekt. Sie können sich in denselben oder unterschiedlichen Ressourcengruppen befinden, da hier keine gemeinsame Infrastruktur erforderlich ist.
- Python 3.9 oder neuer mit beiden SDKs:
pip install azure-ai-ml azure-ai-projects azure-identity
Ein MLClient authentifiziert sich und richtet sich auf den Arbeitsbereich aus:
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>",
)
Training als Befehlsaufgabe
Der Trainingsschritt ist eine Befehlsaufgabe. Er nimmt ein Trainingsskript entgegen – in diesem Fall einen gewöhnlichen scikit-learn-Klassifikator – und führt es auf verwalteten Rechnern aus:
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)
Der Aufruf von ml_client.jobs.stream wartet und gibt Logs aus, bis das Training abgeschlossen ist. Verwenden Sie ihn in jedem Skript, das tatsächlich ausgeführt wird. Eine Aufgabe, die im Hintergrund stumm fehlschlägt, ist keine angenehme Art, herauszufinden, dass ein Pipeline-Workflow gestört ist.
Registrieren des Modells
Durch die Registrierung wird der Ausgabe der Aufgabe ein benannter, versionierter Artefakt umgewandelt, auf das unabhängig zugegriffen werden kann:
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.",
)
)
Die Wahl von MLFLOW_MODEL als Typ ist beabsichtigt. Wenn das Trainingsskript das Modell über das automatische Protokollieren von MLflow protokolliert, können verwaltete Endpunkte es mithilfe eines integrierten Scoring-Containers bereitstellen, ohne ein benutzerdefiniertes score.py-Skript. Dadurch wird Zeit gespart, und das Trainingsskript sollte von Anfang an darauf ausgerichtet sein.
Bereitstellung hinter einem Endpunkt
Ein Endpunkt in Kombination mit einer Bereitstellung stellt das Modell online:
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()
Der Deployment-Name blue ist nicht zwingend erforderlich, aber behalten Sie ihn, da er die Gewohnheit schafft, die bei jeder neuen Modellversion nützlich ist. Erstellen Sie neben blue einen green-Deployment, teilen Sie den Traffic zwischen ihnen auf und wechseln Sie vollständig um, sobald die neue Version Vertrauen gewinnt. blue zu überschreiben und auf das Beste zu hoffen, ist die riskantere Alternative.
Bevor Sie etwas anderes konfigurieren, erhalten Sie direkt eine echte Prognose:
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)
Das Debuggen eines fehlerhaften Endpunkts über den Funktionsaufruf-Zyklus eines Agents ist weitaus schwieriger als das direkte Debuggen, daher sollte man zunächst überprüfen, ob er funktioniert.
Verbindung des Endpunkts mit einem Foundry Agent
Der Agent ruft niemals direkt Azure ML auf. Der Anwendungscode tut dies, wenn der Agent darum bittet. Die Einrichtung deklariert einen FunctionTool, erstellt den Agenten und definiert die Funktion, die die HTTP-Anfrage an den Endpunkt sendet:
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,
),
)
Dann findet ein Dialog statt, der Funktionsaufruf des Modells wird abgefangen und das Ergebnis zurückgegeben:
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)
Das Modell entscheidet, ob eine Frage eine Aufrufung von get_churn_score erfordert. Wenn dem so ist, führt der Anwendungscode den HTTP-Aufruf an Azure ML aus, und die Ausgabe wird in das gleiche Gespräch zurückgesendet, damit das Modell seine Antwort vervollständigen kann. Nichts in der Definition des Agents oder in den Responses-API-Aufrufen hängt davon ab, was die Funktion im Hintergrund tut. Sie könnte eine Datenbank abfragen, eine Lambda-Funktion aufrufen oder auf einen anderen Azure ML-Arbeitsbereich zugreifen, wie es hier der Fall ist.
Zwei Ressourcen, zwei Authentifizierungsgebiete
Ein Punkt bleibt in den SDK-Dokumentationen unerwähnt. Der verwaltete Endpunkt in Azure ML und der Projektendpunkt in Foundry sind zwei Ressourcen hinter zwei getrennten Authentifizierungsmechanismen, und nichts in dieser Konfiguration verbindet sie miteinander. Azure ML verfügt über seine eigene Schlüssel oder verwaltete Identität für den Arbeitsbereich, während Foundry unterschiedliche Anmeldeinformationen verwendet. Die Deklaration eines FunctionTool schafft kein Vertrauensverhältnis zwischen ihnen; sie gibt dem Modell lediglich einen Namen sowie ein Schema, um vom Code etwas anzufragen. Die gesamte Authentifizierung gegenüber Azure ML findet innerhalb Ihrer eigenen Funktion statt, genauso wie bei der Aufrufung jeder anderen externen API.
Das ist ein Vorteil. Die Azure-ML-Seite kann unabhängig von Foundry – sogar von einem anderen Team – verwaltet, gesichert und aktualisiert werden, wobei keiner der Seiten Schreibzugriff auf die Ressourcen des anderen benötigt.
Produktionshinweise
- Eine Funktionsaufruf verfällt nach 10 Minuten. Die Ausgabe muss vor Schließen des Fensters in das Gespräch zurückgesendet werden, andernfalls scheitert der Vorgang. Ist der HTTP-Aufruf zu Azure ML langsam, gilt das als fester Fristtermin, den man berücksichtigen muss.
- Wechseln Sie auf eine verwaltete Identität abseits des Prototyps. Ein in einer Umgebungsvariable gespeicherter Schlüssel ist lokal in Ordnung, stellt aber eine Gefahr dar, wenn nichts überwacht wird. Gewähren Sie der Identität, die den Tool-Code ausführt, die Rolle
AzureML Data Scientistoder eine enger gefasste, auf diesen Endpunkt beschränkte benutzerdefinierte Rolle – nicht auf die gesamte Arbeitsumgebung. - Führen Sie neue Modelle schrittweise ein. Starten Sie die neue Bereitstellung neben der bestehenden und leiten Sie den Verkehr schrittweise um. Das ist die einzige Möglichkeit, ein fehlerhaftes Neutraining frühzeitig zu erkennen, bevor jede echte Anfrage davon abhängt.
strict=True aktiviert. Strikte Schemata verhindern, dass das Modell unvollständige oder fehlerhaft formatierte Argumente an get_churn_score sendet, was sonst innerhalb Ihrer Funktion zu Fehlern führen würde und nicht bereits vor dem Aufruf.get_churn_score-Aufruf an einem überprüfbaren Ort und prüfen Sie regelmäßig die Trainingsdaten – nicht erst, wenn sich die Vorhersagen deutlich verschlechtern.Die richtige Plattform auswählen
Für die meisten Arbeiten mit generativer KI ist Foundrys Katalog das richtige Werkzeug, und Teams ohne eigene Trainingsdaten benötigen möglicherweise nie etwas anderes. Wenn es darum geht, auf der Grundlage eigener Historie Prognosen über eigene Kunden zu erstellen, ist Azure ML die bessere Wahl. Andernfalls wird ein Sprachmodell für eine Aufgabe eingesetzt, für die es ursprünglich nicht konzipiert wurde, oder die Bereitstellungsinfrastruktur muss manuell aufgebaut werden. Der Unterschied ist gering: ein Trainingsjob, ein registriertes Modell, ein verwaltetes Endpunkt sowie eine einzige FunctionTool. Die Plattformen bleiben getrennt und erledigen jeweils das, wofür sie am besten geeignet sind.