Workflows mit AG-UI

In diesem Lernprogramm erfahren Sie, wie Sie Agent Framework-Workflows über einen AG-UI Endpunkt verfügbar machen. Workflows koordinieren mehrere Agents und Tools in einem definierten Ausführungsdiagramm, und die AG-UI Integration streamen umfangreiche Workflowereignisse – Schrittnachverfolgung, Aktivitätsmomentaufnahmen, Unterbrechungen und benutzerdefinierte Ereignisse – in Echtzeit an Webclients.

Voraussetzungen

Bevor Sie beginnen, stellen Sie sicher, dass Sie folgendes haben:

• Python 3.10 oder höher
• agent-framework-ag-ui installiert
• Vertrautheit mit dem Lernprogramm "Erste Schritte "
• Grundlegendes Verständnis des Agentenframeworks Arbeitsabläufe

Wann sollten Workflows mit AG-UI verwendet werden?

Verwenden Sie bei Bedarf einen Workflow anstelle eines einzelnen Agents:

• Multi-Agent-Orchestrierung: Aufgabenrouting zwischen spezialisierten Agenten (z. B. Triage, Rückerstattung, Bestellung)
• Strukturierte Ausführungsschritte: Nachverfolgen des Fortschritts durch definierte Phasen mit STEP_STARTED / STEP_FINISHED Ereignissen
• Unterbrechungs-/Fortsetzungsflüsse: Anhalten der Ausführung, um menschliche Eingaben oder Genehmigungen zu sammeln, und dann fortsetzen
• Benutzerdefiniertes Ereignisstreaming: Emittieren domänenspezifischer Ereignisse (request_info, status, workflow_output) an den Client

Umwicklung eines Workflows mit AgentFrameworkWorkflow

AgentFrameworkWorkflow ist ein leichter Wrapper, der ein systemeigenes Workflow an das AG-UI-Protokoll anpasst. Sie können entweder eine vordefinierte Workflowinstanz oder eine Factory bereitstellen, die einen neuen Workflow pro Thread erstellt.

Direkte Instanz

Verwenden Sie eine direkte Instanz, wenn ein einzelnes Workflowobjekt alle Anforderungen sicher verarbeiten kann (z. B. zustandslose Pipelines):

from agent_framework import Workflow
from agent_framework.ag_ui import AgentFrameworkWorkflow

workflow = build_my_workflow()  # returns a Workflow

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    name="my-workflow",
    description="Single-instance workflow.",
)

Threadbereichs-Fabrik

Verwenden Sie workflow_factory, wenn jeder Unterhaltungsthread einen eigenen Workflowstatus benötigt. Die Fabrik empfängt thread_id und gibt eine frische Workflow zurück:

from agent_framework.ag_ui import AgentFrameworkWorkflow

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda thread_id: build_my_workflow(),
    name="my-workflow",
    description="Thread-scoped workflow.",
)

Registrieren des Endpunkts

Registrieren Sie den Workflow add_agent_framework_fastapi_endpoint auf die gleiche Weise, wie Sie einen einzelnen Agent registrieren würden:

from fastapi import FastAPI
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)

app = FastAPI(title="Workflow AG-UI Server")

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda thread_id: build_my_workflow(),
    name="handoff-demo",
    description="Multi-agent handoff workflow.",
)

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=ag_ui_workflow,
    path="/workflow",
)

Sie können ein Bare Workflow auch direkt übergeben – der Endpunkt fügt es automatisch in AgentFrameworkWorkflow ein:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI Ereignisse, die von Workflows ausgegeben werden

Workflowausführungen geben einen umfangreicheren Satz von AG-UI Ereignissen im Vergleich zu Einzel-Agent-Ausführungen aus:

Clients können Ereignisse verwenden STEP_STARTED / STEP_FINISHED , um Statusanzeigen zu rendern, die zeigen, welcher Agent derzeit aktiv ist.

Unterbrechen und Fortsetzen

Workflows können die Ausführung anhalten, um menschliche Eingaben oder Toolgenehmigungen zu sammeln. Die AG-UI Integration verarbeitet dies über das Interrupt/Resume-Protokoll.

Funktionsweise von Unterbrechungen

1. Während der Ausführung löst der Workflow eine ausstehende Anforderung aus (z. B. eine HandoffAgentUserRequest Aufforderung zu weiteren Details oder ein Tool mit approval_mode="always_require").

2. Die AG-UI-Brücke gibt ein CUSTOM-Ereignis aus, das name="request_info" enthält und die Anforderungsdaten umfasst.

3. Die Ausführung endet mit einem RUN_FINISHED Ereignis, dessen interrupts Feld eine Liste der ausstehenden Anforderungsobjekte enthält:

   {
     "type": "RUN_FINISHED",
     "threadId": "abc123",
     "runId": "run_xyz",
     "interrupts": [
       {
         "id": "request-id-1",
         "value": { "request_type": "HandoffAgentUserRequest", "data": "..." }
       }
     ]
   }
   

4. Der Client rendert die Benutzeroberfläche, auf die der Benutzer reagieren kann (eine Texteingabe, eine Genehmigungsschaltfläche usw.).

Funktionsweise des Lebenslaufs

Der Client sendet eine neue Anforderung mit der resume Nutzlast, die die Antworten des Benutzers enthält, die nach der Interrupt-ID geordnet sind:

{
  "threadId": "abc123",
  "messages": [],
  "resume": {
    "interrupts": [
      {
        "id": "request-id-1",
        "value": "User's response text or approval decision"
      }
    ]
  }
}

Der Server wandelt die Lebenslaufdaten in Workflow-Antworten um und setzt die Ausführung an der Stelle fort, an der sie unterbrochen wurde.

Vollständiges Beispiel: Mehragenten-Übergabe-Workflow

Dieses Beispiel zeigt einen Kundensupport-Workflow mit drei Agenten, die die Arbeit untereinander weitergeben, Tools verwenden, die Genehmigungen erfordern, und bei Bedarf menschliche Eingaben anfragen.

Definieren der Agents und Tools

"""AG-UI workflow server with multi-agent handoff."""

import os

from agent_framework import Agent, Message, Workflow, tool
from agent_framework.ag_ui import (
    AgentFrameworkWorkflow,
    add_agent_framework_fastapi_endpoint,
)
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffBuilder
from azure.identity import AzureCliCredential
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware


@tool(approval_mode="always_require")
def submit_refund(refund_description: str, amount: str, order_id: str) -> str:
    """Capture a refund request for manual review before processing."""
    return f"Refund recorded for order {order_id} (amount: {amount}): {refund_description}"


@tool(approval_mode="always_require")
def submit_replacement(order_id: str, shipping_preference: str, replacement_note: str) -> str:
    """Capture a replacement request for manual review before processing."""
    return f"Replacement recorded for order {order_id} (shipping: {shipping_preference}): {replacement_note}"


@tool(approval_mode="never_require")
def lookup_order_details(order_id: str) -> dict[str, str]:
    """Return order details for a given order ID."""
    return {
        "order_id": order_id,
        "item_name": "Wireless Headphones",
        "amount": "$129.99",
        "status": "delivered",
    }

Erstellen des Workflows

def create_handoff_workflow() -> Workflow:
    """Build a handoff workflow with triage, refund, and order agents."""
    client = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        credential=AzureCliCredential(),
    )

    triage = Agent(id="triage_agent", name="triage_agent", instructions="...", client=client)
    refund = Agent(id="refund_agent", name="refund_agent", instructions="...", client=client,
                   tools=[lookup_order_details, submit_refund])
    order = Agent(id="order_agent", name="order_agent", instructions="...", client=client,
                  tools=[lookup_order_details, submit_replacement])

    def termination_condition(conversation: list[Message]) -> bool:
        for msg in reversed(conversation):
            if msg.role == "assistant" and (msg.text or "").strip().lower().endswith("case complete."):
                return True
        return False

    builder = HandoffBuilder(
        name="support_workflow",
        participants=[triage, refund, order],
        termination_condition=termination_condition,
    )
    builder.add_handoff(triage, [refund], description="Route refund requests.")
    builder.add_handoff(triage, [order], description="Route replacement requests.")
    builder.add_handoff(refund, [order], description="Route to order after refund.")
    builder.add_handoff(order, [triage], description="Route back after completion.")

    return builder.with_start_agent(triage).build()

Erstellen der FastAPI-App

app = FastAPI(title="Workflow AG-UI Demo")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow_factory=lambda _thread_id: create_handoff_workflow(),
    name="support_workflow",
    description="Customer support handoff workflow.",
)

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=ag_ui_workflow,
    path="/support",
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8888)

Ereignissequenz

Eine typische Multi-Turn-Interaktion erzeugt Ereignisse wie:

RUN_STARTED           threadId=abc123
STEP_STARTED          stepName=triage_agent
TEXT_MESSAGE_START     role=assistant
TEXT_MESSAGE_CONTENT   delta="I'll look into your refund..."
TEXT_MESSAGE_END
STEP_FINISHED         stepName=triage_agent
STEP_STARTED          stepName=refund_agent
TOOL_CALL_START       toolCallName=lookup_order_details
TOOL_CALL_ARGS        delta='{"order_id":"12345"}'
TOOL_CALL_END
TOOL_CALL_START       toolCallName=submit_refund
TOOL_CALL_ARGS        delta='{"order_id":"12345","amount":"$129.99",...}'
TOOL_CALL_END
RUN_FINISHED          interrupts=[{id: "...", value: {function_approval_request}}]

Der Client kann dann ein Genehmigungsdialogfeld anzeigen und mit der Entscheidung des Benutzers fortsetzen.

Nächste Schritte

Zusätzliche Ressourcen

• AG-UI Übersicht
• Erste Schritte
• Agent-Framework-Workflows
• Agent Framework GitHub Repository