Erstellen und ausführen Sie dynamische Workflows mit Azure Functions gehosteten Fähigkeiten

In diesem Artikel aktivieren Sie dynamische Workflows für eine von Azure Functions gehostete Fähigkeit, erstellen workflow-sichere Python-Tools und führen einen Workflow lokal aus. Sie können das Standard-Azure Storage-Backend oder das empfohlene Durable Task Scheduler (DTS)-Backend verwenden.

Voraussetzungen

Bevor Sie beginnen, benötigen Sie Folgendes:

• Ein vorhandenes Azure Functions hosted skills-Projekt, das Python verwendet und über main.agent.md verfügt. Falls du noch keine hast, sieh dir Erstelle eine ereignisgesteuerte KI-App mit in Azure Functions gehosteten Skills an.
• Das Paket azurefunctions-agents-runtime in requirements.txt.
• Azure Functions Core Tools.
• Ein Modellanbieter, der für die gehostete Skills-Laufzeit konfiguriert ist.

Du brauchst außerdem ein Speicher-Backend für die Workflow-Laufzeit. Wählen Sie das Backend, das zu Ihrer Umgebung passt:

Für ein vollständiges Projekt, das diesen Workflow demonstriert, siehe das Beispiel für die Workflow-Vorfalltriage.

Aktiviere Workflows auf der gehosteten Skill

Füge workflows.enabled: true zum Frontmatter-Bereich der Datei .agent.md hinzu, in der du dynamische Workflows verwenden möchtest. Das folgende Beispiel zeigt eine gehostete Skill mit aktivierten Workflows:

---
name: Incident Triage Assistant
description: Investigates incidents by gathering evidence from multiple sources in parallel, correlating findings, and producing a written report.
builtin_endpoints: true
workflows:
  enabled: true
---

You are an incident-triage assistant. When a user describes a production incident, gather evidence from logs, metrics, and deployment history. When the work involves multiple evidence sources or a settling delay, run it as a workflow and summarize the final result for the user.

Wenn Sie true auf workflows.enabled setzen, fügt die Laufzeit die im Überblick über dynamische Workflows beschriebenen Workflow-Management-Tools hinzu.

Erstellen Sie workflow-sichere Tools

Dynamische Workflows dürfen nur Werkzeuge aufrufen, die diese Anforderungen erfüllen:

• Laufen Sie synchron (nicht asynchron).
• Akzeptiere ein einziges dict Argument.
• Gib einen JSON-serialisierbaren Wert zurück.
• idempotent sein, denn ein Worker-Ausfall kann dazu führen, dass das Tool mehrmals ausgeführt wird.

Markiere eine Funktion als workflowsicher, indem du sie mit @workflow_tool dekorierst und im Ordner tools/ ablegst. Wenn du die Funktion auch als normales Chat-Tool zur Verfügung stellen möchtest, füge den @tool Dekorateur neben @workflow_toolhinzu.

Das folgende Beispiel definiert drei workflow-sichere Tools. Zwei Werkzeuge sammeln Beweise, und ein Werkzeug fasst die Ergebnisse zusammen:

from typing import Any

from azure_functions_agents import workflow_tool


@workflow_tool(description="Fetch recent log lines for a service.")
def fetch_logs(args: dict[str, Any]) -> dict[str, Any]:
    service = args["service"]
    return {
        "service": service,
        "lines": [
            f"[ERROR] {service}: upstream timeout",
            f"[WARN] {service}: latency above SLO",
        ],
        "errors": 1,
        "warnings": 1,
    }


@workflow_tool(description="Fetch recent service metrics.")
def fetch_metrics(args: dict[str, Any]) -> dict[str, Any]:
    service = args["service"]
    return {
        "service": service,
        "cpu_p99": 86.2,
        "latency_p99_ms": 1420,
        "saturation": "high",
    }


@workflow_tool(description="Summarize logs and metrics into an incident finding.")
def summarize_findings(args: dict[str, Any]) -> dict[str, Any]:
    logs = args["logs"]
    metrics = args["metrics"]
    return {
        "service": logs["service"],
        "likely_cause": "resource pressure is correlated with elevated latency",
        "evidence": [
            f"{logs['errors']} error log entries found",
            f"p99 latency is {metrics['latency_p99_ms']} ms",
        ],
        "recommended_action": "scale out the service and review recent dependency timeouts",
    }

Verstehen Sie den Workflow-Plan

Das Modell erstellt den Plan, der an start_workflowübergeben wird. Der folgende vereinfachte Plan sammelt Protokolle und Metriken parallel, wartet 30 Sekunden und fasst dann die Ergebnisse zusammen:

{
  "tasks": [
    {
      "id": "logs",
      "type": "tool",
      "tool": "fetch_logs",
      "args": {
        "service": "orders-api"
      }
    },
    {
      "id": "metrics",
      "type": "tool",
      "tool": "fetch_metrics",
      "args": {
        "service": "orders-api"
      }
    },
    {
      "id": "settle",
      "type": "wait",
      "duration": "PT30S",
      "depends_on": [
        "logs",
        "metrics"
      ]
    },
    {
      "id": "summary",
      "type": "tool",
      "tool": "summarize_findings",
      "args": {
        "logs": "${logs.result}",
        "metrics": "${metrics.result}"
      },
      "depends_on": [
        "settle"
      ]
    }
  ]
}

Im generierten Plan wird die vollständige JSON-Ausgabe einer vorgelagerten Aufgabe referenziert. ${task_id.result} Zum Beispiel übergibt ${logs.result} den gesamten Rückgabewert der logs Aufgabe an summarize_findings. Das Modell kann auch auf ein bestimmtes Feld mit ${task_id.result.path.to.field}verweisen.

Lokal ausführen

Starte den Functions-Host und führe einen Workflow aus, indem du das von dir gewählte Speicher-Backend in den Voraussetzungen nutzt.

Verwandte Inhalte

• Überblick über dynamische Arbeitsabläufe
• Überwachen und Behandeln von Problemen von gehosteten Skills
• Beispiel für die Workflow-Vorfall-Triage
• Muster von Workflow-Subagenten
• Entwickeln mit dauerhaftem Aufgabenplaner
• Übersicht über dauerhafte Funktionen