Arbetsflöden med AG-UI

Anmärkning

Arbetsflödesstöd för .NET AG-UI integrering kommer snart.

Den här handledningen visar hur du exponerar arbetsflöden i Agent Framework via en AG-UI-slutpunkt. Arbetsflöden samordnar flera agenter och verktyg i ett definierat körningsdiagram, och AG-UI integrering strömmar omfattande arbetsflödeshändelser – stegspårning, ögonblicksbilder av aktiviteter, avbrott och anpassade händelser – till webbklienter i realtid.

Förutsättningar

Kontrollera att du har följande innan du börjar:

Python 3.10 eller senare
agent-framework-ag-ui installerad
Bekanta dig med guiden Komma igång
Grundläggande förståelse för Agent Framework-arbetsflöden

När du ska använda arbetsflöden med AG-UI

Använd ett arbetsflöde i stället för en enda agent när du behöver:

Orkestrering med flera agenter: Dirigera uppgifter mellan specialiserade agenter (till exempel triage → återbetalning → beställning)
Strukturerade körningssteg: Spåra förloppet genom definierade steg med STEP_STARTED / STEP_FINISHED händelser
Avbryt/återuppta flöden: Pausa körningen för att samla in mänskliga indata eller godkännanden och återuppta sedan
Anpassad händelseströmning: Generera domänspecifika händelser (request_info, status, workflow_output) till klienten

Omsluta ett arbetsflöde med AgentFrameworkWorkflow

AgentFrameworkWorkflow är ett lättviktsomslag som anpassar ett inbyggt Workflow till AG-UI-protokollet. Du kan ange antingen en fördefinierad arbetsflödesinstans eller en fabrik som skapar ett nytt arbetsflöde per tråd.

Direktinstans

Använd en direktinstans när ett enskilt arbetsflödesobjekt på ett säkert sätt kan hantera alla begäranden (till exempel tillståndslösa 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.",
)

Trådomfattningsfabrik

Använd workflow_factory när varje konversationstråd behöver sitt eget arbetsflödestillstånd. Fabriken tar emot thread_id och returnerar en ny Workflow:

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

Viktigt!

Du måste antingen välja workflow, inte båda alternativen. Omslutningen genererar en ValueError om båda anges.

Registrera slutpunkten

Registrera arbetsflödet på add_agent_framework_fastapi_endpoint samma sätt som du registrerar en enda agent:

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

Du kan också skicka ett ospecificerat Workflow direkt – slutpunkten omsluter den automatiskt i AgentFrameworkWorkflow.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI händelser som genereras av arbetsflöden

Arbetsflöden genererar en mer omfattande uppsättning AG-UI-händelser jämfört med enskilda agentkörningar.

Event	När det avges	Beskrivning
`RUN_STARTED`	Körningen börjar	Markerar början av arbetsflödets körning
`STEP_STARTED`	En exekutor eller ett supersteg börjar	`step_name` identifierar agenten eller steget (till exempel `"triage_agent"`)
`TEXT_MESSAGE_*`	Agenten skapar text	Standardhändelser för strömmande text
`TOOL_CALL_*`	Agenten anropar ett verktyg	Anropshändelser för standardverktyg
`STEP_FINISHED`	En utförare eller supersteget har slutförts	Stänger steget för förloppsspårning för användargränssnittet
`CUSTOM` (`status`)	Ändringar i arbetsflödestillstånd	Innehåller `{"state": "<value>"}` i händelsevärdet
`CUSTOM` (`request_info`)	Arbetsflöde begär mänskliga indata	Innehåller begärans nyttolast för att klienten ska kunna visa en uppmaning
`CUSTOM` (`workflow_output`)	Arbetsflödet genererar utdata	Innehåller slutdata eller mellanliggande utdata
`RUN_FINISHED`	Körningen har slutförts	Kan inkludera `interrupts` om arbetsflödet väntar på indata

Klienter kan använda STEP_STARTED / STEP_FINISHED händelser för att återge förloppsindikatorer som visar vilken agent som för närvarande är aktiv.

Avbryt och återuppta

Arbetsflöden kan pausa körningen för att samla in mänskliga indata eller verktygsgodkännanden. Den AG-UI integreringen hanterar detta via avbrotts-/återuppta-protokollet.

Så här fungerar avbrott

Under körningen genererar arbetsflödet en väntande begäran (till exempel en HandoffAgentUserRequest begäran om mer information eller ett verktyg med approval_mode="always_require").
Den AG-UI bryggan genererar en CUSTOM händelse med name="request_info" innehållande begärandedata.

Körningen avslutas med en RUN_FINISHED händelse vars interrupts fält innehåller en lista över väntande begärandeobjekt:

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

Klienten renderar användargränssnittet så att användaren kan svara (en textinmatning, en godkännandeknapp osv.).

Så här fungerar CV

Klienten skickar en ny begäran med resume nyttolasten som innehåller användarens svar som är nyckelade med avbrotts-ID:

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

Servern konverterar nyttolasten för återupptagning till svar från arbetsflödet och återupptar körningen där den pausades.

Fullständigt exempel: Arbetsflöde för överlämning av flera agenter

Det här exemplet visar ett arbetsflöde för kundsupport med tre agenter som lämnar över arbetet till varandra, använder verktyg som kräver godkännande och begär mänskliga indata vid behov.

Definiera agenter och verktyg

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

Skapa arbetsflödet

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

Skapa FastAPI-appen

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)

Händelsesekvens

En typisk interaktion med flera svängar skapar händelser som:

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}}]

Klienten kan sedan visa en dialogruta för godkännande och återuppta med användarens beslut.

Nästa steg

Human-in-the-Loop

Ytterligare resurser

Feedback

Var den här sidan till hjälp?

Last updated on 2026-04-09

Dela via

Arbetsflöden med AG-UI

Förutsättningar

När du ska använda arbetsflöden med AG-UI

Omsluta ett arbetsflöde med AgentFrameworkWorkflow

Direktinstans

Trådomfattningsfabrik

Registrera slutpunkten

AG-UI händelser som genereras av arbetsflöden

Avbryt och återuppta

Så här fungerar avbrott

Så här fungerar CV

Fullständigt exempel: Arbetsflöde för överlämning av flera agenter

Definiera agenter och verktyg

Skapa arbetsflödet

Skapa FastAPI-appen

Händelsesekvens

Nästa steg

Ytterligare resurser

Feedback

Ytterligare resurser