Werkstromen met AG-UI

MAF-.NET kan een werkstroom beschikbaar maken via AG-UI door de werkstroom te converteren naar een AIAgent werkstroom en deze toe tewijsen zoals elke andere agent:

AIAgent workflowAgent = AgentWorkflowBuilder
    .BuildSequential(researcher, reporter)
    .AsAIAgent();

app.MapAGUIServer("/", workflowAgent);

Het eindpunt streamt de standaardtekst en de uitvoer van toolaanroepen van de onderliggende agents. AuthorName identificeert de agent die elke update heeft geproduceerd.

MAF .NET vertaalt workflowspecifiek levenscyclusgedrag momenteel niet naar AG-UI. Clients ontvangen geen workflowstapgebeurtenissen, activiteitssnapshots, workflowonderbrekingen of workflowhervattingsbewerkingen zoals in de Python-integratie. Het verpakken van een werkstroom als een AIAgent voegt die toewijzingen niet toe.

Zie microsoft/agent-framework#2494 voor de huidige .NET-traceringsstatus. Zie de concepten van MAF-werkstromen voor werkstroomconstructie en -uitvoering onafhankelijk van AG-UI.

Volgende stappen 

In deze zelfstudie leert u hoe u Agent Framework-werkstromen beschikbaar maakt via een AG-UI-eindpunt. Werkstromen organiseren meerdere agents en hulpprogramma's in een gedefinieerde uitvoeringsgrafiek, en de AG-UI-integratie streamt uitgebreide werkstroomevenementen, staptracering, momentopnamen van activiteiten, interrupts en aangepaste gebeurtenissen, naar webclients in realtime.

Prerequisites

Voordat u begint, moet u ervoor zorgen dat u het volgende heeft:

• Python 3.10 of hoger
• agent-framework-ag-ui en agent-framework-foundry geïnstalleerd
• Bekendheid met de Aan de slag handleiding
• Basiskennis van werkstroomconcepten van Agent Framework

Wanneer werkstromen gebruiken met AG-UI

Gebruik een werkstroom in plaats van één agent wanneer u het volgende nodig hebt:

• Orkestratie met meerdere agenten: routeren van taken tussen gespecialiseerde agenten (bijvoorbeeld triage → restitutie → bestelling)
• Stappen voor gestructureerde uitvoering: voortgang bijhouden door gedefinieerde fasen met STEP_STARTED / STEP_FINISHED gebeurtenissen
• Stroom onderbreken/hervatten: uitvoering pauzeren om menselijke invoer of goedkeuringen te verzamelen en vervolgens hervatten
• Aangepaste gebeurtenisstreaming: domeinspecifieke gebeurtenissen (request_info, status, workflow_output) verzenden naar de client

Een werkstroom verpakken met AgentFrameworkWorkflow

AgentFrameworkWorkflow is een lichtgewicht wrapper die een systeemeigen Workflow aanpast aan het AG-UI-protocol. U kunt een vooraf gebouwd werkstroomexemplaar of een fabriek specificeren waarmee een nieuwe werkstroom per thread wordt gemaakt.

Directe instantie

Gebruik een directe instantie wanneer één workflow-object veilig alle verzoeken kan verwerken (bijvoorbeeld stateless 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.",
)

Factory met threadbereik

Gebruik workflow_factory wanneer elke gespreksthread een eigen werkstroomstatus nodig heeft. De fabriek ontvangt de thread_id en retourneert een nieuw 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.",
)

Het eindpunt registreren

Registreer de werkstroom op add_agent_framework_fastapi_endpoint dezelfde manier als u één agent zou registreren:

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

U kunt ook direct een kale Workflow doorgeven — het eindpunt omhult het automatisch met AgentFrameworkWorkflow.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI gebeurtenissen die worden uitgezonden door werkstromen

Werkstroomuitvoeringen verzenden een uitgebreidere set AG-UI gebeurtenissen in vergelijking met uitvoeringen met één agent:

Clients kunnen gebeurtenissen gebruiken STEP_STARTED / STEP_FINISHED om voortgangsindicatoren weer te geven die aangeven welke agent momenteel actief is.

Onderbreken en hervatten

Werkstromen kunnen de uitvoering onderbreken om menselijke invoer of goedkeuringen voor hulpprogramma's te verzamelen. De AG-UI-integratie behandelt dit via het interrupt/resume-protocol.

Hoe onderbrekingen werken

1. Tijdens de uitvoering genereert de werkstroom een aanvraag die in behandeling is (bijvoorbeeld een HandoffAgentUserRequest vraag om meer details of een hulpprogramma met approval_mode="always_require").

2. De AG-UI brug verzendt een CUSTOM gebeurtenis met name="request_info" de aanvraaggegevens.

3. De run eindigt met een RUN_FINISHED-gebeurtenis waarvan het veld outcome.interrupts de openstaande aanvragen bevat:

   {
     "type": "RUN_FINISHED",
     "threadId": "abc123",
     "runId": "run_xyz",
     "outcome": {
       "type": "interrupt",
       "interrupts": [
         {
           "id": "request-id-1",
           "reason": "input_required",
           "message": "Provide the requested information.",
           "responseSchema": { "type": "string" },
           "metadata": {
             "agent_framework": {
               "request_type": "HandoffAgentUserRequest"
             }
           }
         }
       ]
     }
   }
   

4. De client geeft de gebruikersinterface weer om te reageren (een tekstinvoer, een goedkeuringsknop, enzovoort).

Hoe cv werkt

De client verzendt een nieuwe aanvraag met een canonieke resume matrix. Elke vermelding identificeert de interrupt en levert het antwoord van de gebruiker:

{
  "threadId": "abc123",
  "messages": [],
  "resume": [
    {
      "interruptId": "request-id-1",
      "status": "resolved",
      "payload": "User's response text or approval decision"
    }
  ]
}

De server converteert de nettolading van het cv naar werkstroomantwoorden en gaat verder met de uitvoering van de plaats waar deze is onderbroken. Als u in plaats daarvan de onderbroken uitvoering wilt annuleren, stelt u status in op "cancelled" en laat u payload weg.

Controlepunten voor werkstromen behouden en hervatten

Configureer checkpoint_storage aan AgentFrameworkWorkflow om de onderliggende werkstroomstatus aan het einde van elke superstep op te slaan. U kunt in plaats daarvan hetzelfde argument add_agent_framework_fastapi_endpoint doorgeven wanneer u een werkstroom registreert. De opslag moet beschikbaar zijn voor de AG-UI wrapper of eindpunt om een controlepunt te hervatten via AG-UI.

In het volgende voorbeeld wordt opslag in het geheugen gebruikt voor een werkstroom met korte levensduur:

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

app = FastAPI()
checkpoint_storage = InMemoryCheckpointStorage()
workflow = build_my_workflow()

ag_ui_workflow = AgentFrameworkWorkflow(
    workflow=workflow,
    checkpoint_storage=checkpoint_storage,
)
add_agent_framework_fastapi_endpoint(
    app,
    ag_ui_workflow,
    "/workflow",
)

AgentFrameworkWorkflow.run() ontvangt de payload van de AG-UI-aanvraag, zodat een client de checkpoint-ID via meegegeven eigenschappen doorgeeft in plaats van via een Python-checkpoint_idargument. Een cv met alleen controlepunten bevat geen nieuw gebruikersbericht:

{
  "threadId": "abc123",
  "messages": [],
  "forwardedProps": {
    "checkpointId": "checkpoint-id-from-your-storage"
  }
}

De adapter herstelt de opgeslagen werkstroomstatus en gaat door met de uitvoering. Als het checkpoint een interrupt bevat die nog in behandeling is, neem dan zowel de checkpoint-ID als de canonieke resume payload op in hetzelfde verzoek. De adapter herstelt het controlepunt voordat het interrupt-antwoord wordt geleverd.

InMemoryCheckpointStorage overleeft het opnieuw opstarten van het proces niet. Zie Controlepunten voor duurzame opslagopties en controlepuntselectie.

Momentopnamen van controlepunten in de workflow en van AG-UI-threads

Werkstroomcontrolepunten en AG-UI threadmomentopnamen behouden verschillende gegevens:

U kunt beide mechanismen configureren. Een controlepunt voor een werkstroom vervangt geen momentopname van een AG-UI thread en een momentopname van een AG-UI thread bevat niet de uitvoeringsstatus van de uitvoerder die is vereist om de uitvoering van de werkstroom te hervatten.

Volledig voorbeeld: Handoff-werkstroom voor meerdere agents

In dit voorbeeld ziet u een werkstroom voor klantondersteuning met drie agents die werk aan elkaar overdragen, hulpprogramma's gebruiken waarvoor goedkeuring is vereist en waar nodig menselijke invoer aanvragen.

De agenten en hulpmiddelen definiëren

"""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.foundry import FoundryChatClient
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",
    }

De werkstroom bouwen

def create_handoff_workflow() -> Workflow:
    """Build a handoff workflow with triage, refund, and order agents."""
    client = FoundryChatClient(
        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
        model=os.environ["FOUNDRY_MODEL"],
        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()

De FastAPI-app maken

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)

Gebeurtenisreeks

Een typische interactie met meerdere bochten produceert gebeurtenissen zoals:

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          outcome={type: "interrupt", interrupts: [{id: "...", reason: "tool_call"}]}

De client kan vervolgens een goedkeuringsdialoogvenster weergeven en hervatten met de beslissing van de gebruiker.

Doorgestuurde props ontvangen

AG-UI clients (zoals CopilotKit) kunnen een forwarded_props (of forwardedProps) veld in de invoerlading opnemen. De AG-UI-integratie geeft deze props automatisch door aan de run-methode van de werkstroom via het function_invocation_kwargs-trefwoordargument.

class MyWorkflow(Workflow):
    async def run(
        self,
        *,
        message=None,
        responses=None,
        stream: bool = False,
        function_invocation_kwargs: dict | None = None,
    ):
        forwarded_props = (function_invocation_kwargs or {}).get("forwarded_props", {})
        # Use forwarded_props for custom routing, feature flags, etc.
        ...

Belangrijke details:

• Beide forwarded_props en forwardedProps worden geaccepteerd in de invoerpayload; intern worden ze genormaliseerd tot forwarded_props.
• In doorgestuurde eigenschappen zijn checkpoint_id en checkpointId gereserveerd voor het hervatten vanaf een workflow-controlepunt.
• Als workflow.run()function_invocation_kwargs (of **kwargs) niet accepteert, worden de props ongemerkt verwijderd — bestaande werkstromen worden niet beïnvloed.
• Doorgestuurde props worden ook opgeslagen in sessiemetagegevens, maar worden gefilterd op LLM-gebonden metagegevens, zodat ze niet lekken in chatclientaanvragen.

Volgende stappen 

Aanvullende bronnen

• overzicht vanAG-UI
• Aan de slag
• Werkstromen
• GitHub-opslagplaats voor Agent Framework

Go kan werkstromen beschikbaar maken voor AG-UI door een workflow.Workflow als agent te verpakken met workflow/agentworkflowen die agent vervolgens te hosten met provider/aguiprovider.

workflowAgent, err := agentworkflow.New(wf, agentworkflow.AgentConfig{
    IncludeOutputsInResponse: true,
    Config: agent.Config{
        Name: "WorkflowAgent",
    },
})
if err != nil {
    panic(err)
}

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(workflowAgent, aguiprovider.HandlerConfig{}))