KI-Workflows mit dem Microsoft Agent Framework orchestrieren

Inhalt

Workflows im Microsoft Agent Framework sind graphbasierte Prozesse, die Entwickler vorab festlegen: eine feste Folge von Schritten (Executors), die über Kanten verbunden sind und in einer vorhersehbaren Reihenfolge laufen. Sie eignen sich statt eines gesprächsorientierten ChatAgent immer dann, wenn bestimmte Schritte garantiert laufen müssen, Freigaben an den richtigen Stellen passieren sollen und Sie hinterher nachvollziehen müssen, was passiert ist. In diesem Leitfaden bauen wir in Python einen IT-Ticket-Workflow mit vier Schritten, starten ihn über die Konsole und in der DevUI und ergänzen dann bedingte Kanten, die Tickets mit hoher Priorität an einen Alarm weiterleiten.

In unserem vorherigen Beitrag haben wir mit dem Microsoft Agent Framework einen praxistauglichen IT-Helpdesk-Agenten gebaut. Der ChatAgent sprach mit den Benutzern, prüfte über Tools den Status von Systemen und verwaltete den Zustand des Gesprächs. Er verstand, was der Benutzer wollte, und reagierte sofort mit den passenden Schritten.

Echte Geschäftsprozesse sind selten so einfach. Was passiert, wenn sich ein Prozess über mehrere Tage zieht, etwa weil eine Führungskraft eine neue Softwarelizenz freigeben muss? Wie gehen Sie mit einer Aufgabe um, die anhalten, auf eine menschliche Entscheidung warten und dann ohne Kontextverlust weiterlaufen muss? Und was ist mit einer mehrstufigen Datenpipeline, in der jeder Schritt erfolgreich sein muss, bevor der nächste beginnt, und die gesamte Abfolge prüfbar sein soll?

Hier stößt der gesprächsorientierte ChatAgent an seine Grenzen. Ein ChatAgent bearbeitet spontane Anfragen dynamisch und entscheidet anhand der Eingabe, was er tut. Ein Workflow legt dagegen eine feste, graphbasierte Abfolge von Schritten fest. Workflows sind für komplexe, lange laufende Prozesse gedacht, bei denen Zuverlässigkeit und Nachvollziehbarkeit Pflicht sind.

Dieser Beitrag behandelt die Orchestrierungs-Engine selbst: wie Sie Workflows mit linearen und bedingten Pfaden bauen, ausführen und untersuchen. Dauerhaftigkeit und das Human-in-the-Loop-Muster, die KI im Unternehmen etwas komplexer, aber vertrauenswürdig machen, sind hier nicht Thema.

ChatAgent oder Workflow: das richtige Werkzeug wählen

Bevor Sie Code schreiben, steht im Microsoft Agent Framework eine grundlegende Architekturentscheidung an: Wann nehmen Sie einen gesprächsorientierten ChatAgent, und wann bauen Sie einen strukturierten Workflow? Die beiden sind nicht austauschbar, denn jeder löst eine andere Art von Problemen.

ChatAgent: der Gesprächspartner

Wie in unserem vorherigen Leitfaden gezeigt, ist der ChatAgent stark bei offenen, dynamischen Interaktionen. Sein Verhalten steuert das Large Language Model in seinem Kern. Sie geben ihm Anweisungen, eine Reihe von Tools und einen Prompt des Benutzers, und der Agent überlegt, wie er am besten vorgeht. Sein Weg steht nicht vorher fest, er ergibt sich aus dem Dialog.

Ein ChatAgent passt zu gesprächsbasierten Oberflächen für Benutzer (Chatbots, Copilots), zu spontanen Aufgaben, bei denen die Absicht des Benutzers den Ablauf bestimmt, und zu Fällen, in denen Flexibilität und Sprachverständnis wichtiger sind als ein starrer, wiederholbarer Ablauf.

Angetrieben wird er von den Überlegungen des LLM, Runde für Runde: Es plant, ruft ein Tool auf, betrachtet das Ergebnis und plant in einer Schleife neu. Der Zustand des Agenten liegt im AgentThread, der den Gesprächsverlauf enthält. Für den Kontext einer einzelnen, zusammenhängenden Interaktion ist das ideal. Ohne zusätzlichen Entwicklungsaufwand ist der AgentThread aber nicht dafür gedacht, Systemneustarts dauerhaft zu überstehen. Die größte Stärke eines ChatAgent ist seine Flexibilität: Er passt sich unerwarteten Anfragen an und führt komplexe Gespräche ohne Drehbuch.

Workflow: der strukturierte Orchestrator

Ein Workflow ist dagegen ein deterministischer, graphbasierter Prozess, den der Entwickler festlegt. Ist ein Agent eine kluge Mitarbeiterin, der man ein Ziel gibt, dann ist der Workflow der dokumentierte Geschäftsprozess, an den sie sich halten muss.

Der Kontrollfluss ist explizit: eine Reihe von Knoten (Executors), die über Kanten verbunden sind. Innerhalb eines Knotens kann ein LLM eine anspruchsvolle Aufgabe übernehmen, die Richtung des Gesamtprozesses bestimmt es aber nicht. Aus unserer Sicht ermöglicht erst diese explizite Struktur die besten Enterprise-Funktionen des Frameworks.

Workflows passen zu etablierten Geschäftsprozessen (Spesenfreigaben, Onboarding neuer Benutzer), zu lange laufenden Aufgaben, die anhalten und weiterlaufen müssen (warten auf menschliche Eingaben, ein stundenlanger Datenjob), und zu Fällen, in denen Nachvollziehbarkeit, Zuverlässigkeit und ein vorhersehbarer Ablauf nicht verhandelbar sind.

Angetrieben wird ein Workflow von einem Ausführungsgraphen, den der Entwickler definiert. Der Ablauf wandert anhand vorab festgelegter Verbindungen und Bedingungen von einem Executor zum nächsten. Workflows sind auf Dauerhaftigkeit ausgelegt: Sie unterstützen Checkpointing, ihr Zustand lässt sich also jederzeit serialisieren und dauerhaft speichern (etwa in einer Datenbank). So überstehen sie Neustarts und können beliebig lange auf externe Ereignisse warten. Die größte Stärke eines Workflows ist seine Zuverlässigkeit: Der Prozess ist vorhersehbar, nachvollziehbar und übersteht Unterbrechungen.

Agenten in Workflows einbetten

Sie müssen sich nicht für eines von beiden entscheiden. Die ausgefeiltesten Lösungen kombinieren oft beide Muster: Ein Workflow steuert den übergeordneten Prozess, und einer seiner Knoten ist ein ChatAgent, der eine bestimmte, agentenartige Teilaufgabe übernimmt.

Nehmen wir eine Pipeline zur Dokumentenverarbeitung:

1. Workflow-Knoten 1 (Executor): Eine einfache Funktion holt ein neues PDF aus einem SharePoint-Ordner. Dieser Schritt ist deterministisch und zuverlässig.
2. Workflow-Knoten 2 (Agent): Das PDF geht an einen ChatAgent mit der Anweisung: "Du bist Rechtsanalyst. Lies dieses Dokument, fasse die wichtigsten Klauseln zusammen und extrahiere die Namen aller beteiligten Parteien in ein JSON-Objekt." Dieser Schritt nutzt die Denkfähigkeit des LLM für eine komplexe, unstrukturierte Aufgabe.
3. Workflow-Knoten 3 (Executor): Die strukturierte JSON-Ausgabe des Agenten wird in einer Datenbank gespeichert. Auch das ist ein deterministischer, zuverlässiger Schritt.

Eingehende Kundenbestellungen sind ein typischer Kandidat für eine solche Pipeline. Bei Meusburger gehen 96 % der Bestellungen, die per E-Mail, PDF oder Excel eintreffen, ohne Korrektur ins System.

Dieser hybride Ansatz verbindet die nachvollziehbare Orchestrierung eines Workflows mit der flexiblen Denkfähigkeit eines Agenten für klar umrissene Aufgaben. Weitere Muster dieser Art beschreiben wir in unserem Beitrag über zuverlässige KI-Agenten-Systeme.

Eine einfache Faustregel für die Entscheidung:

Nehmen Sie einen ChatAgent, wenn...	Nehmen Sie einen Workflow, wenn...
die wichtigste Schnittstelle ein Gespräch mit einem Benutzer ist.	Sie einen mehrstufigen Geschäftsprozess im Backend automatisieren.
die Abfolge der Schritte unvorhersehbar ist und vom LLM bestimmt wird.	die Abfolge der Schritte bekannt ist und vorhersehbar sein soll.
die Aufgabe relativ kurzlebig ist (Sekunden bis Minuten).	der Prozess lange laufen kann (Stunden oder Tage).
Sie maximale Flexibilität für unterschiedlichste Eingaben brauchen.	Sie maximale Zuverlässigkeit, Nachvollziehbarkeit und Kontrolle brauchen.

Grundbegriffe von Workflows: Executors, Kanten und Kontext

Um einen Workflow zu bauen, konstruieren Sie einen ausführbaren gerichteten azyklischen Graphen (DAG). Das Microsoft Agent Framework bietet dafür klar getrennte Bausteine für Knoten, Kanten und den Datenfluss des Graphen. Wer modulare, wartbare automatisierte Prozesse entwerfen will, muss sie kennen.

Executors: die Knoten des Graphen

Ein Executor ist die grundlegende Recheneinheit eines Workflows: ein einzelner, in sich geschlossener Knoten im Ausführungsgraphen. Jeder Executor sollte genau eine Aufgabe haben: eine Eingabe empfangen, eine bestimmte Operation ausführen und eine Ausgabe erzeugen. So bleibt der Workflow modular und testbar.

Das Framework kennt zwei Arten, Executors zu implementieren:

1. Klassenbasierte Executors: Für Komponenten mit komplexer Logik, internem Zustand oder umfangreicher Konfiguration definieren Sie eine Klasse, die von agent_framework.Executor erbt. Dieser objektorientierte Ansatz passt zu anspruchsvoller Geschäftslogik.
2. Funktionsbasierte Executors: Für zustandslose Transformationen mit nur einem Zweck definieren Sie eine async-Funktion und versehen sie mit dem Decorator @executor. Dieses schlanke Muster passt zu einfachen Knoten für Datenzuordnung, Filterung oder Weiterleitung.

Handler (@handler): der Einstiegspunkt der Ausführung

In einem klassenbasierten Executor markiert der Decorator @handler die async-Methode, die die Workflow-Engine aufruft. Diese Methode ist der Einstiegspunkt für die Logik des Knotens.

Bei funktionsbasierten Executors ist die Funktion selbst implizit der Handler, der Decorator @executor genügt also. @handler legt die Ausführungslogik innerhalb einer Klasse explizit fest.

WorkflowBuilder und Kanten: die Topologie des Graphen festlegen

Eine Sammlung von Executors ist noch kein Workflow, solange ihre Struktur nicht festgelegt ist. Der WorkflowBuilder ist die Fluent-API, mit der Sie die Topologie des Graphen definieren: die gerichteten Kanten, die Kontrollfluss und Daten zwischen den Executors bestimmen.

Die WorkflowBuilder-API bietet deklarative Methoden, um den Graphen aufzubauen:

• set_start_executor(executor): Legt den Einstiegsknoten des Workflows fest.
• add_edge(source_executor, destination_executor): Erzeugt eine gerichtete Kante. Sendet der source_executor eine Nachricht, leitet die Workflow-Engine sie als Eingabe an den destination_executor weiter.
• add_edge(source_executor, destination_executor, condition=...): Erzeugt Verzweigungen im Graphen. Die Funktion condition erhält die Nachricht und entscheidet, ob die Kante genommen wird. Daten werden so anhand ihres Inhalts weitergeleitet. Das brauchen Sie für Geschäftsregeln und bedingte Logik.

Dieser Ansatz trennt die Orchestrierungslogik (die Struktur des Graphen) von der Geschäftslogik (der Implementierung der Executors), und das hält das System wartbar.

WorkflowContext: die Schnittstelle zur Workflow-Engine

Executors sprechen nicht direkt miteinander. Sie sind vollständig entkoppelt und kommunizieren über die Workflow-Engine mit dem Objekt WorkflowContext (ctx), das jedem Handler als Argument übergeben wird.

WorkflowContext bietet zwei Methoden, um den Datenfluss zu steuern:

1. ctx.send_message(data): Gibt eine Ausgabe des aktuellen Executors aus. Die Engine fängt diesen Aufruf ab und leitet die Nutzdaten data an alle nachgelagerten Knoten weiter, die über ausgehende Kanten verbunden sind. Das ist der übliche Weg, Daten zwischen Knoten weiterzugeben.
2. ctx.yield_output(data): Damit veröffentlichen Endknoten ein Endergebnis für den gesamten Workflow. Alle Daten, die an yield_output gehen, werden gesammelt und an den externen Client zurückgegeben, der den Workflow gestartet hat. Ein Workflow kann mehrere Endknoten haben und mehrere Ausgaben liefern.

Praxis Teil 1: ein mehrstufiger Workflow für IT-Anfragen

Workflows versteht man am besten, indem man einen baut. Wir erstellen einen mehrstufigen Workflow, der die erste Bearbeitung, die Anreicherung und die Protokollierung einer IT-Supportanfrage automatisiert.

Der Workflow ist eine Pipeline mit vier Schritten:

1. Kategorisieren: Ein LLM analysiert die rohe Anfrage eines Benutzers und extrahiert strukturierte Angaben wie Problembereich und Priorität.
2. Anreichern: Der Workflow nimmt die ermittelte Kategorie und reichert das Ticket an, indem er in einer vordefinierten Wissensbasis das zuständige IT-Supportteam nachschlägt.
3. Formatieren: Er fasst alle Informationen in einem sauberen JSON-Objekt zusammen, das direkt an eine API gehen kann.
4. Ticket anlegen: Zum Schluss simuliert er einen Aufruf an die API eines externen Ticketsystems und gibt eine abschließende Bestätigung aus.

Zuerst bauen wir den Workflow und starten ihn über die Kommandozeile, um den Prozess von Anfang bis Ende zu sehen. Danach nutzen wir die eingebaute Dev Web UI des Frameworks als grafische Oberfläche zum Debuggen und Ausprobieren.

Voraussetzungen

Bevor Sie beginnen, stellen Sie sicher, dass Ihre Umgebung alle nötigen Komponenten enthält, auch die für die Weboberfläche.

1. Installieren Sie die nötigen Pakete:

   1pip install agent-framework --pre
   

2. Setzen Sie die Umgebungsvariablen für Azure OpenAI: Der Workflow verwendet den AzureOpenAIResponsesClient. Folgende Variablen müssen gesetzt sein:

   • AZURE_OPENAI_ENDPOINT
   • AZURE_OPENAI_CHAT_DEPLOYMENT_NAME
   • AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME
   • AZURE_OPENAI_API_KEY

Schritt 1: Datenmodelle und Executors definieren

Der Workflow gibt streng typisierte Daten mit Pydantic-Modellen von Knoten zu Knoten weiter. Wir definieren vier Executors, einen für jeden Schritt der Pipeline.

Legen Sie eine Datei executors.py an.

1import asyncio
2import time
3from pydantic import BaseModel, Field
4from typing import cast
5from typing_extensions import Never
6
7from agent_framework import Executor, WorkflowContext, executor, handler
8from agent_framework.azure import AzureOpenAIResponsesClient
9
10import os
11
12os.environ["AZURE_OPENAI_ENDPOINT"] = "https://<your-name>.openai.azure.com"
13os.environ["AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"] = "gpt-5"
14os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"] = "gpt-5"
15os.environ["AZURE_OPENAI_API_KEY"] = "<your-key>"
16
17
18# --- Datenmodelle ---
19class CategorizedTicket(BaseModel):
20    category: str = Field(
21        description="The general category of the IT issue, e.g., 'Hardware', 'Software', 'Network', 'Access Request'."
22    )
23    priority: str = Field(
24        description="The estimated priority, e.g., 'Low', 'Medium', 'High'."
25    )
26    summary: str = Field(description="A one-sentence summary of the user's issue.")
27
28
29class EnrichedTicket(CategorizedTicket):
30    assigned_team: str = Field(description="The IT team responsible for this category.")
31
32
33# --- Executors ---
34
35
36# Executor 1: Rohe Anfrage mit einem LLM kategorisieren.
37class CategorizeRequestExecutor(Executor):
38    def __init__(self, id: str):
39        super().__init__(id=id)
40        self.responses_client = AzureOpenAIResponsesClient()
41
42    @handler
43    async def categorize(
44        self, user_request: str, ctx: WorkflowContext[CategorizedTicket]
45    ):
46        """Takes raw text, extracts details, and sends a CategorizedTicket."""
47        print(f"--- Node Log (Categorize): Analyzing request... ---")
48        prompt = f'Analyze the IT support request and extract details.\nRequest: "{user_request}"'
49        response = await self.responses_client.get_response(
50            prompt, response_format=CategorizedTicket
51        )
52        if response.value:
53            categorized_ticket = cast(CategorizedTicket, response.value)
54            await ctx.send_message(categorized_ticket)
55        else:
56            raise ValueError("No structured response received from LLM")
57
58
59# Executor 2: Ticket mit interner Geschäftslogik anreichern.
60@executor(id="enrich_ticket_executor")
61async def enrich_ticket(
62    ticket: CategorizedTicket, ctx: WorkflowContext[EnrichedTicket]
63):
64    """Receives a ticket, assigns a team, and forwards an EnrichedTicket."""
65    print(
66        f"--- Node Log (Enrich): Assigning team for category '{ticket.category}'... ---"
67    )
68    team_assignments = {
69        "Hardware": "Desktop Support L2",
70        "Software": "Application Support",
71        "Network": "Network Operations",
72        "Access Request": "Identity & Access Management",
73        "Default": "General Helpdesk L1",
74    }
75    assigned_team = team_assignments.get(ticket.category, team_assignments["Default"])
76    enriched_ticket = EnrichedTicket(**ticket.model_dump(), assigned_team=assigned_team)
77    await ctx.send_message(enriched_ticket)
78
79
80# Executor 3: Angereicherte Daten als JSON-String für eine API formatieren.
81@executor(id="format_ticket_executor")
82async def format_ticket_as_json(ticket: EnrichedTicket, ctx: WorkflowContext[str]):
83    """
84    Receives an EnrichedTicket and sends a JSON string to the next node.
85    Note: The context type is WorkflowContext[str] because it passes data on,
86    it is no longer a terminal node.
87    """
88    print(f"--- Node Log (Format): Formatting ticket into JSON... ---")
89    json_output = ticket.model_dump_json(indent=2)
90    await ctx.send_message(json_output)
91
92
93# Executor 4: Anlage des Tickets in einem externen System simulieren.
94@executor(id="create_ticket_executor")
95async def create_ticket_in_system(ticket_json: str, ctx: WorkflowContext[Never, str]):
96    """
97    Receives the ticket JSON, simulates an API call, and yields the final result.
98    This is the new terminal node of our workflow.
99    """
100    print(f"--- Node Log (Create Ticket): Calling ticketing system API... ---")
101    # Latenz des API-Aufrufs simulieren
102    await asyncio.sleep(1)
103    ticket_id = f"IT-{int(time.time())}"
104    print(f"--- Node Log (Create Ticket): Successfully created ticket {ticket_id} ---")
105
106    # Endergebnis für den gesamten Workflow ausgeben.
107    await ctx.yield_output(
108        f"Successfully created ticket {ticket_id} with details:\n{ticket_json}"
109    )

Schritt 2: Den Workflow bauen und über die Konsole starten

Mit den vier Executors können wir sie verbinden und den Workflow direkt starten, um die Logik von Anfang bis Ende zu testen.

Legen Sie eine Datei main.py mit folgendem Inhalt an.

1# main.py
2import asyncio
3from agent_framework import WorkflowBuilder
4from executors import (
5    CategorizeRequestExecutor,
6    create_ticket_in_system,
7    enrich_ticket,
8    format_ticket_as_json,
9)
10
11async def run_workflow():
12    # 1. Executors instanziieren.
13    categorize_node = CategorizeRequestExecutor(id="categorize_request_node")
14    enrich_node = enrich_ticket
15    format_node = format_ticket_as_json
16    create_ticket_node = create_ticket_in_system
17
18    # 2. Workflow-Graphen mit vier Schritten aufbauen.
19    workflow = (
20        WorkflowBuilder()
21        .add_edge(categorize_node, enrich_node)
22        .add_edge(enrich_node, format_node)
23        .add_edge(format_node, create_ticket_node)
24        .set_start_executor(categorize_node)
25        .build()
26    )
27
28    # 3. Eingabe festlegen und Workflow starten.
29    user_request = "My VPN client keeps disconnecting every 5 minutes, it's impossible to work."
30    print(f"Starting workflow with input: '{user_request}'\n")
31    events = await workflow.run(user_request)
32
33    # 4. Endergebnis abholen und ausgeben.
34    final_outputs = events.get_outputs()
35    print("\nWorkflow completed. Final output:")
36    for output in final_outputs:
37        print(output)
38
39    print(f"\nFinal state: {events.get_final_state()}")
40
41if __name__ == "__main__":
42    asyncio.run(run_workflow())

Starten Sie das Skript im Terminal mit python main.py. Sie sehen nacheinander die Log-Ausgaben aller vier Knoten, ein Zeichen, dass der Workflow korrekt durchgelaufen ist.

Erwartete Konsolenausgabe:

1Starting workflow with input: 'My VPN client keeps disconnecting every 5 minutes, it's impossible to work.'
2
3--- Node Log (Categorize): Analyzing request... ---
4--- Node Log (Enrich): Assigning team for category 'Network'... ---
5--- Node Log (Format): Formatting ticket into JSON... ---
6--- Node Log (Create Ticket): Calling ticketing system API... ---
7--- Node Log (Create Ticket): Successfully created ticket IT-1732598400 ---
8
9Workflow completed. Final output:
10Successfully created ticket IT-1732598400 with details:
11{
12  "category": "Network",
13  "priority": "High",
14  "summary": "User's VPN client frequently disconnects, disrupting their work.",
15  "assigned_team": "Network Operations"
16}
17
18Final state: WorkflowRunState.IDLE

Schritt 3: Den Workflow in der DevUI visualisieren

Zum Überprüfen reicht die Konsole, aber das Framework bringt auch ein sehr gutes Werkzeug für Entwicklung und Debugging mit: die Dev Web UI. Sie zeigt den Ausführungsgraphen, macht den Datenfluss zwischen den Knoten sichtbar und lässt Sie den Workflow interaktiv ausführen.

Wir passen main.py so an, dass diese Oberfläche startet.

1. Das Workflow-Objekt anlegen: Definieren Sie zuerst ein wiederverwendbares Workflow-Objekt außerhalb des Hauptprogramms.

2. main.py für die Oberfläche anpassen: Ersetzen Sie die Funktion run_workflow und den Hauptblock durch den Code, der die Oberfläche startet.

1# main.py
2from agent_framework import WorkflowBuilder
3from agent_framework.devui import serve # <-- Funktion serve importieren
4from executors import (
5    CategorizeRequestExecutor,
6    create_ticket_in_system,
7    enrich_ticket,
8    format_ticket_as_json,
9)
10
11# 1. Executors instanziieren und Workflow-Objekt bauen.
12# Das ist derselbe Teil wie vorher, nur auf oberster Ebene definiert.
13categorize_node = CategorizeRequestExecutor(id="categorize_request_node")
14enrich_node = enrich_ticket
15format_node = format_ticket_as_json
16create_ticket_node = create_ticket_in_system
17
18workflow = (
19    WorkflowBuilder(
20        name="IT Ticket Triage Workflow",
21        description="A four-step workflow to categorize, enrich, format, and create an IT support ticket."
22    )
23    .add_edge(categorize_node, enrich_node)
24    .add_edge(enrich_node, format_node)
25    .add_edge(format_node, create_ticket_node)
26    .set_start_executor(categorize_node)
27    .build()
28)
29
30# 2. Hauptfunktion definieren, die die DevUI startet.
31def main():
32    """Launch the IT Ticket Triage workflow in the Dev Web UI."""
33    print("Starting IT Ticket Triage Workflow...")
34    print("Navigate to http://localhost:8091 in your browser.")
35
36    # `serve` startet einen Webserver und stellt die Oberfläche für die übergebenen Entitäten bereit.
37    serve(entities=[workflow], port=8091, auto_open=True)
38
39if __name__ == "__main__":
40    main()

Starten Sie das geänderte Skript mit python main.py. Im Browser öffnet sich http://localhost:8091.

Zuerst sehen Sie Ihren Workflow als interaktiven Graphen.

DevUI-Graph des IT-Ticket-Workflows mit vier Schritten

Klicken Sie auf "Configure & Run", geben Sie eine Beispielanfrage ein und führen Sie den Workflow aus.

DevUI-Dialog zum Eingeben der Workflow-Eingabe und Starten des Workflows

Während der Ausführung verfolgen Sie, wie die Daten durch die Knoten fließen, sehen Ein- und Ausgaben und prüfen, ob sich jeder Schritt wie erwartet verhält.

Zeitleiste der Ausführung in der DevUI: alle vier Knoten abgeschlossen, mit Endergebnis

Praxis Teil 2: bedingte Kanten für Alarme bei hoher Priorität

Lineare Workflows sind ein guter Anfang, aber echte Geschäftsprozesse brauchen oft Verzweigungen. Eine kritische Supportanfrage soll nicht denselben Weg nehmen wie eine Routineanfrage. Das Microsoft Agent Framework löst das mit bedingten Kanten, die Daten anhand ihres Inhalts weiterleiten.

Als Nächstes erweitern wir den IT-Workflow so, dass er Tickets mit hoher Priorität anders behandelt. Stuft das LLM ein Ticket mit der Priorität "High" ein, verzweigt der Workflow und schickt es an einen eigenen Benachrichtigungsknoten "Send Alert" (den wir simulieren). Tickets mit niedriger und mittlerer Priorität laufen weiter über die normale Ticketanlage.

Damit bekommt der Graph einen Entscheidungspunkt, und die Automatisierung reagiert auf die Dringlichkeit jeder Anfrage.

Schritt 1: Einen Alarm-Executor und Bedingungsfunktionen ergänzen

Wir brauchen einen neuen Executor, der Alarme verschickt, und Funktionen, die die Weiterleitungslogik festlegen. Beides kommt in die bestehende Datei executors.py.

1. Den neuen SendAlertExecutor ergänzen

Das ist ein neuer Endknoten für den Zweig mit hoher Priorität.

2. Die Bedingungsfunktion anlegen

Eine Bedingungsfunktion ist eine einfache boolesche Funktion. Sie prüft die Daten, die über eine Kante fließen, und entscheidet, ob dieser Weg genommen wird.

Ergänzen Sie Ihre Datei executors.py um Folgendes:

1# executors.py
2# (Den bisherigen Code aus dem vorherigen Kapitel beibehalten)
3...
4
5# --- Neuer Executor für den Pfad mit hoher Priorität ---
6
7# Executor 5: Endknoten für den Zweig mit hoher Priorität, der einen Alarm sendet.
8@executor(id="send_alert_executor")
9async def send_alert(ticket: EnrichedTicket, ctx: WorkflowContext[Never, str]):
10    """
11    Simulates sending a high-priority alert to a monitoring channel (e.g., Slack, PagerDuty).
12    This node also yields a final output for its branch.
13    """
14    print(f"--- Node Log (ALERT): Sending HIGH PRIORITY alert for team '{ticket.assigned_team}'... ---")
15    alert_message = f"HIGH PRIORITY ALERT: Ticket with summary '{ticket.summary}' assigned to {ticket.assigned_team}."
16
17    # In einer echten Anwendung stünde hier ein Aufruf der Slack- oder PagerDuty-API.
18    await asyncio.sleep(0.5) # API-Aufruf simulieren
19
20    # Eigene Ausgabe für den Alarm-Pfad liefern.
21    await ctx.yield_output(alert_message)
22
23
24# --- Neue Bedingungsfunktionen für die Weiterleitung ---
25
26def is_high_priority(ticket: EnrichedTicket) -> bool:
27    """Condition function that returns True if the ticket priority is 'High'."""
28    print(f"--- Condition Check: Is priority '{ticket.priority}' high? {ticket.priority.lower() == 'high'} ---")
29    return ticket.priority.lower() == 'high'
30
31def is_normal_priority(ticket: EnrichedTicket) -> bool:
32    """Condition function that returns True if the ticket priority is not 'High'."""
33    print(f"--- Condition Check: Is priority '{ticket.priority}' normal? {ticket.priority.lower() != 'high'} ---")
34    return ticket.priority.lower() != 'high'

Schritt 2: Den verzweigten Workflow bauen

Jetzt passen wir main.py an, um den neuen, nicht linearen Graphen aufzubauen. Mit add_edge und dem Parameter condition erzeugen wir nach dem enrich_node zwei Zweige:

• Der enrich_node wird zum Entscheidungspunkt.
• Liefert is_high_priority den Wert True, fließen die Daten zum send_alert_executor.
• Liefert is_normal_priority den Wert True, fließen die Daten wie bisher zum format_ticket_executor.

Ändern Sie Ihre main.py entsprechend.

1# main.py
2from agent_framework import WorkflowBuilder
3from agent_framework.devui import serve
4from executors import (
5    CategorizeRequestExecutor,
6    create_ticket_in_system,
7    enrich_ticket,
8    format_ticket_as_json,
9    send_alert,
10    is_high_priority,
11    is_normal_priority,
12)
13
14# 1. Alle Executors instanziieren, auch den neuen.
15categorize_node = CategorizeRequestExecutor(id="categorize_request_node")
16enrich_node = enrich_ticket
17format_node = format_ticket_as_json
18create_ticket_node = create_ticket_in_system
19alert_node = send_alert # Neuer Alarm-Knoten
20
21# 2. Workflow mit bedingten Zweigen bauen.
22workflow = (
23    WorkflowBuilder(
24        name="IT Ticket Triage Workflow with Priority Routing",
25        description="A workflow that routes high-priority tickets to a special alert path."
26    )
27    .add_edge(categorize_node, enrich_node)
28
29    # --- Verzweigung nach dem enrich_node ---
30    # Pfad 1: Tickets mit normaler Priorität durchlaufen die normale Ticketanlage.
31    .add_edge(enrich_node, format_node, condition=is_normal_priority)
32    .add_edge(format_node, create_ticket_node)
33
34    # Pfad 2: Tickets mit hoher Priorität gehen direkt zum Alarm-Knoten.
35    .add_edge(enrich_node, alert_node, condition=is_high_priority)
36
37    .set_start_executor(categorize_node)
38    .build()
39)
40
41# 3. Workflow über die DevUI bereitstellen (dieser Teil bleibt gleich).
42def main():
43    """Launch the branching IT Ticket Triage workflow in the Dev Web UI."""
44    print("Starting IT Ticket Triage Workflow with Priority Routing...")
45    print("Navigate to http://localhost:8091 in your browser.")
46    serve(entities=[workflow], port=8091, auto_open=True)
47
48if __name__ == "__main__":
49    main()

Schritt 3: Beide Zweige in der DevUI testen

Starten Sie das geänderte Skript main.py. Die DevUI zeigt jetzt den verzweigten Graphen.

DevUI-Graph mit getrennten Zweigen für Formatieren und Anlegen sowie für den Alarm

Testfall 1: normale Priorität

Geben Sie I need to reset my password for the HR portal, I forgot the old one. ein und klicken Sie auf "Run Workflow". Das LLM sollte die Anfrage mit der Priorität "Low" oder "Medium" einstufen. Im Graphen der DevUI läuft der Workflow über den oberen Zweig Enrich -> Format -> Create Ticket, und am Ende steht die übliche Bestätigung, dass das Ticket angelegt wurde.

Testfall 2: hohe Priorität

Geben Sie The entire payment processing service is down! We cannot process any customer credit cards! ein und klicken Sie auf "Run Workflow". Das LLM stuft die Anfrage mit "High" ein, und der Workflow läuft über den unteren Zweig Enrich -> Send Alert. Am Ende zeigt die Oberfläche die Alarmmeldung mit hoher Priorität.

Mit bedingten Kanten ist der Workflow keine einfache Pipeline mehr: Er verhält sich je nach Daten unterschiedlich. Dieses Muster brauchen Sie für jede Automatisierung, die mit der Vielfalt echter Anfragen zurechtkommen muss.

Zusammenfassung: von Pipelines zu Prozessen

In diesem Leitfaden sind wir vom gesprächsorientierten ChatAgent zur strukturierten Orchestrierung mit Workflows gekommen. Begonnen haben wir mit einer linearen Pipeline aus vier Executors, die unstrukturierte Eingaben von Benutzern in angereicherte, strukturierte Daten für ein nachgelagertes System verwandelt.

Danach kam bedingte Logik dazu. Durch die Verzweigung anhand des Inhalts (in unserem Fall der Ticketpriorität) wurde aus der Pipeline ein Geschäftsprozess, der dringende und routinemäßige Anfragen unterschiedlich behandelt. Explizite, nachvollziehbare und bedingte Ausführungspfade wie diese sind die Grundlage zuverlässiger Automatisierung im Unternehmen.

Die bisher gebauten Workflows sind noch in sich geschlossen und laufen in einem Durchgang von Anfang bis Ende. Die wichtigsten Unternehmensprozesse funktionieren oft anders. Sie müssen ihren Zustand über Systemneustarts hinweg behalten, lange auf externe Ereignisse warten und, vor allem, bei kritischen Entscheidungen menschliches Urteilsvermögen einbeziehen.

Auch für diese Anforderungen des Produktivbetriebs bietet das Framework Bausteine:

• Dauerhaftigkeit und Zustandsverwaltung: Mit Checkpointing werden Workflows dauerhaft. Ein Workflow lässt sich dann anhalten, sein Zustand in einer Datenbank sichern und Stunden oder sogar Tage später fortsetzen. Das schützt ihn vor Systemausfällen.
• Human-in-the-Loop: Mit einem Freigabeschritt hält der Workflow an und wartet auf ein externes Signal, etwa die Entscheidung einer Führungskraft, bevor er weiterläuft. So bauen Sie vertrauenswürdige KI-Systeme, die das Tempo der Automatisierung mit menschlicher Aufsicht verbinden.

Die unten verlinkte offizielle Dokumentation zu Workflows behandelt beides.

Wenn Sie einen solchen Prozess in Ihrem Unternehmen automatisieren möchten, etwa die Verarbeitung eingehender Bestellungen oder Anfragen, unterstützt Sie unser Team für KI-Beratung von der Analyse bis zum Produktivbetrieb.

Offizielle Ressourcen

• Offizielle Dokumentation zu Workflows: Die maßgebliche Quelle für Konzepte, fortgeschrittene Muster und die API-Referenz.
• GitHub-Repository: Quellcode, weitere Beispiele (auch die aus diesem Beitrag) und die Möglichkeit, selbst zum Projekt beizutragen.
• Ankündigung im Blog: Microsofts ursprüngliche Ankündigung mit Hintergründen zu Vision und Strategie hinter dem Framework (auf Englisch).

Sie möchten zuverlässige KI-Agenten-Systeme bauen?

Wir haben auf Basis aktueller Forschung einen ausführlichen Leitfaden (auf Englisch) erstellt, wie Sie robuste KI-Agenten-Systeme bauen, die auch im produktiven Betrieb funktionieren. Der Leitfaden behandelt:

• die 14 systematischen Fehlerquellen in Multi-Agenten-Systemen
• belegte Best Practices für das Design von Agenten
• strukturierte Kommunikationsprotokolle und Prüfmechanismen

Weiterlesen

• KI-Agenten mit dem Microsoft Agent Framework: Der erste Beitrag dieser Serie und der Einstieg in das ChatAgent-Muster und die zentralen Enterprise-Funktionen des Frameworks.
• AI Agents From Scratch (auf Englisch): Die Grundlagen von KI-Agenten, von der Planung bis zur Ausführung von Tools, als Hintergrund für die Agenten-Bausteine, die Sie in Workflows einbetten können.
• Langfuse: Observability für LLM-Anwendungen: Observability ist für Produktivsysteme unverzichtbar. Dieser Beitrag zeigt Werkzeuge und Techniken, um komplexe LLM-Anwendungen und Workflows zu überwachen und zu debuggen.
• RAG mit Azure AI Search: Wie Sie Agenten und Workflows mit internem Wissen versorgen, indem Sie ein Retrieval-Augmented-Generation-System auf Azure bauen.
• KI-Agenten mit n8n: Ein anderer Low-Code-Ansatz für agentische Workflows, nützlich als Vergleich für die Prozessautomatisierung.
• SharePoint-Listen als Wissensquelle in Copilot Studio: Der Low-Code-Weg im Microsoft-Stack: ein Agent in Copilot Studio mit Agent-Flow-Tools, die SharePoint-Daten abfragen.
• LLM-Tracing mit Opik (auf Englisch): Traces, Evaluierungen und Monitoring-Dashboards für LLM-Anwendungen, als Alternative zu Langfuse.
• Inhaltsfilter in Azure OpenAI: Was die Standardfilter in Geschäftsszenarien blockieren und wie Sie sie konfigurieren, bevor Ihre Workflows im Produktivbetrieb daran scheitern.

Über den Autor

Andreas Nigg

Mitgründer, Pondhouse Data

Andreas entwickelt seit mehr als acht Jahren Datenanwendungen. Bei Pondhouse Data konzipiert und baut er KI-Systeme und integriert sie in komplexe, gewachsene Unternehmensumgebungen.

Sprechen Sie mit Andreas über Ihr ProjektAuf LinkedIn folgen