Flux de travail avec AG-UI

Le .NET MAF peut exposer un flux de travail via AG-UI en convertissant le flux de travail en un AIAgent et en le mappant comme n’importe quel autre agent :

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

app.MapAGUIServer("/", workflowAgent);

Le point de terminaison d’API transmet en flux le texte standard et les sorties des appels d’outil des agents qui le composent. AuthorName identifie l’agent qui a produit chaque mise à jour.

MAF .NET n’associe actuellement pas à AG-UI le comportement du cycle de vie spécifique au workflow. Les clients ne reçoivent pas d'événements d'étape de flux de travail, d'instantanés d'activité, d'interruptions de flux de travail ou d'opérations de reprise de flux de travail équivalentes à l'intégration Python. Encapsuler un flux de travail en AIAgent n’ajoute pas ces correspondances.

Pour connaître l’état actuel du suivi des .NET, consultez microsoft/agent-framework#2494. Pour la construction et l’exécution de flux de travail indépendants d’AG-UI, consultez les concepts de workflow MAF.

Étapes suivantes

Ce tutoriel vous montre comment exposer des flux de travail Agent Framework via un point de terminaison AG-UI. Les flux de travail orchestrent plusieurs agents et outils dans un graphique d'exécution défini, et l'intégration AG-UI diffuse des événements de flux de travail enrichis — suivi des étapes, instantanés d’activité, interruptions et événements personnalisés — aux clients web en temps réel.

Prerequisites

Avant de commencer, assurez-vous d’avoir :

• Python 3.10 ou version ultérieure
• agent-framework-ag-ui et agent-framework-foundry installés
• Connaissance du didacticiel de prise en main
• Compréhension de base des concepts de flux de travail agent Framework

Quand utiliser des flux de travail avec AG-UI

Utilisez un flux de travail au lieu d’un seul agent lorsque vous avez besoin de :

• Orchestration multi-agent : acheminer les tâches entre les agents spécialisés (par exemple, triage → remboursement → commande)
• Étapes d’exécution structurées : suivre la progression à travers des étapes définies avec STEP_STARTED / STEP_FINISHED des événements
• Interruption/reprise des flux : suspendre l’exécution pour collecter des entrées ou approbations humaines, puis reprendre
• Diffusion en continu d’événements personnalisés : émettre des événements spécifiques au domaine (request_info, status, workflow_output) au client

Encapsulation d’un workflow avec AgentFrameworkWorkflow

AgentFrameworkWorkflow est un wrapper léger qui adapte un natif Workflow au protocole AG-UI. Vous pouvez fournir soit une instance de workflow préconstruite, soit une fabrique qui crée un nouveau workflow par thread.

Instance directe

Utilisez une instance directe lorsqu’un seul objet de flux de travail peut traiter en toute sécurité toutes les requêtes (par exemple, pipelines sans état) :

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

Fabrique à portée de thread

Utilisez workflow_factory quand chaque thread de conversation a besoin de son propre état de flux de travail. La fabrique reçoit le thread_id et retourne une nouvelle 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.",
)

Enregistrement du point de terminaison

Enregistrez le flux de travail de add_agent_framework_fastapi_endpoint de la même façon que vous enregistreriez un seul 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",
)

Vous pouvez également passer directement un Workflow nu — le point de terminaison l'encapsule automatiquement dans un AgentFrameworkWorkflow :

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

événements AG-UI émis par les flux de travail

Les exécutions de flux de travail émettent un ensemble plus riche d’événements AG-UI par rapport aux exécutions à agent unique :

Les clients peuvent utiliser STEP_STARTED / STEP_FINISHED des événements pour afficher les indicateurs de progression montrant quel agent est actuellement actif.

Interruption et reprise

Les flux de travail peuvent suspendre l’exécution pour collecter des entrées humaines ou des approbations d’outils. L’intégration AG-UI gère cela via le protocole d’interruption/reprise.

Fonctionnement des interruptions

1. Pendant l’exécution, le flux de travail déclenche une demande en attente (par exemple, une HandoffAgentUserRequest demande de détails supplémentaires ou un outil avec approval_mode="always_require").

2. Le pont AG-UI émet un CUSTOM événement avec name="request_info" contenant les données de la requête.

3. L’exécution se termine par un RUN_FINISHED événement dont outcome.interrupts le champ contient les demandes en attente :

   {
     "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. Le client restitue l’interface utilisateur pour que l’utilisateur réponde (entrée de texte, bouton d’approbation, etc.).

Fonctionnement de la fonctionnalité de reprise

Le client envoie une nouvelle requête avec un tableau canonique resume . Chaque entrée identifie l’interruption et fournit la réponse de l’utilisateur :

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

Le serveur convertit la charge utile de reprise en réponses de flux de travail et continue l’exécution à partir de l’endroit où il a suspendu. Pour annuler plutôt l’exécution interrompue, définissez status sur "cancelled" et omettez payload.

Conserver et reprendre des points de contrôle de flux de travail

Configurez checkpoint_storage sur AgentFrameworkWorkflow pour enregistrer l’état du flux de travail sous-jacent à la fin de chaque super-étape. Vous pouvez à la place passer le même argument add_agent_framework_fastapi_endpoint lorsque vous inscrivez un flux de travail. Le stockage doit être disponible pour le wrapper ou le point de terminaison AG-UI pour reprendre un point de contrôle via AG-UI.

L’exemple suivant utilise le stockage en mémoire pour un flux de travail de courte durée :

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()reçoit la charge utile de la demande AG-UI. Par conséquent, un client fournit l’ID de point de contrôle par le biais de propriétés transférées au lieu d’un argument Pythoncheckpoint_id. Une reprise à partir d’un point de contrôle uniquement n’inclut pas de nouveau message de l’utilisateur :

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

L’adaptateur restaure l’état du flux de travail enregistré et poursuit l’exécution. Si le point de contrôle contient une interruption en attente, incluez à la fois l’ID de point de contrôle et la charge utile canonique resume dans la même requête. L’adaptateur restaure le point de contrôle avant de remettre la réponse d’interruption.

InMemoryCheckpointStorage ne survive pas aux redémarrages de processus. Pour connaître les options de stockage durables et la sélection des points de contrôle, consultez Points de contrôle.

Points de contrôle du workflow et captures d’état des threads d’AG-UI

Les points de contrôle de flux de travail et les captures instantanées de threads AG-UI conservent différentes données :

Vous pouvez configurer les deux mécanismes. Un point de contrôle de flux de travail ne remplace pas un instantané de thread AG-UI, et un instantané de thread de AG-UI ne contient pas l’état d’exécuteur requis pour reprendre l’exécution du flux de travail.

Exemple complet : flux de travail de transfert multi-agent

Cet exemple montre un flux de travail de support client avec trois agents qui se utilisent mutuellement, utilisent des outils nécessitant une approbation et demandent une entrée humaine si nécessaire.

Définir les agents et les outils

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

Générer le workflow

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

Créer l’application FastAPI

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)

Séquence d’événements

Une interaction multitour classique produit des événements tels que :

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

Le client peut ensuite afficher une boîte de dialogue d’approbation et reprendre avec la décision de l’utilisateur.

Réception d’accessoires transférés

AG-UI clients (par exemple, CopilotKit) peuvent inclure un forwarded_props champ (ou forwardedProps) dans la charge utile d’entrée. L'intégration AG-UI transmet automatiquement ces propriétés à la méthode du workflow run via l'argument de mot-clé function_invocation_kwargs.

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.
        ...

Détails clés :

• Les deux forwarded_props et forwardedProps sont acceptés dans la charge utile d’entrée ; en interne, ils sont normalisés en forwarded_props.
• Dans les propriétés transférées, checkpoint_id et checkpointId sont réservées à la reprise du point de contrôle de flux de travail.
• S’il workflow.run() n’accepte pas function_invocation_kwargs (ou **kwargs), les props sont supprimés de manière silencieuse, et les flux de travail existants ne sont pas affectés.
• Les propriétés transférées sont également stockées dans les métadonnées de session, mais sont filtrées des métadonnées associées à l'LLM, de sorte qu'elles ne fuient pas dans les demandes des clients de chat.

Étapes suivantes

Ressources additionnelles

• Vue d’ensemble deAG-UI
• Mise en route
• Flux de travail
• Dépôt GitHub pour Agent Framework

Go peut exposer des flux de travail à AG-UI en encapsulant un workflow.Workflow agent avec workflow/agentworkflow, puis en hébergeant cet agent avec 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{}))