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 diffuse le texte standard et la sortie d’appel d’outil des agents constituants. AuthorName identifie l’agent qui a produit chaque mise à jour.

MAF .NET ne mappe pas actuellement le comportement de cycle de vie spécifique au flux de travail à l'interface utilisateur ag. 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. L’habillage d’un flux de travail en tant que AIAgent mappage n’ajoute pas ces mappages.

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épendamment de l’interface utilisateur du groupe de disponibilité, consultez les concepts du flux de travail MAF.

Étapes suivantes

Examiner les considérations relatives à la production et à la sécurité

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

Important

Vous devez choisir entre l’un ou l’autreworkflow, pas les deux. Le wrapper déclenche un ValueError si les deux sont fournis.

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 :

Événement	En cas d’émission	Description
`RUN_STARTED`	Début de l'exécution	Marque le début de l’exécution du flux de travail
`STEP_STARTED`	Un exécuteur ou un superstep commence	`step_name` identifie l’agent ou l’étape (par exemple, `"triage_agent"`)
`TEXT_MESSAGE_*`	L’agent produit du texte	Événements standard de texte en streaming
`TOOL_CALL_*`	L’agent appelle un outil	Événements d’appel d’outil standard
`STEP_FINISHED`	Un exécuteur ou un superstep se termine	Ferme l’étape pour le suivi de la progression de l’interface utilisateur
`CUSTOM` (`status`)	Modifications de l’état du flux de travail	Contient `{"state": "<value>"}` dans la valeur de l’événement
`CUSTOM` (`request_info`)	Requêtes de flux de travail nécessitant une intervention humaine	Contient la charge de demande pour que le client affiche une invite
`CUSTOM` (`workflow_output`)	Le flux de travail produit une sortie	Émis pour les événements `"output"` (terminaux) et les événements `"intermediate"` de flux de travail. Les sorties du terminal correspondent à la réponse finale ; les sorties intermédiaires apparaissent comme contenu `text_reasoning` lorsque le flux de travail s’exécute derrière `as_agent()`.
`RUN_FINISHED`	Le processus d'exécution est terminé	Inclut `outcome.type == "interrupt"` et `outcome.interrupts` quand le flux de travail attend l’entrée

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

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").
Le pont AG-UI émet un CUSTOM événement avec name="request_info" contenant les données de la requête.

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

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.

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

Human-in-the-Loop

Ressources additionnelles

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{}))

Tip

Consultez le flux de travail en tant qu’exemple d’agent et l’exemple de serveurAG-UI pour obtenir des exemples exécutables complets.

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-08-14

Flux de travail avec AG-UI

Étapes suivantes

Prerequisites

Quand utiliser des flux de travail avec AG-UI

Encapsulation d’un workflow avec AgentFrameworkWorkflow

Instance directe

Fabrique à portée de thread

Enregistrement du point de terminaison

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

Interruption et reprise

Fonctionnement des interruptions

Fonctionnement de la fonctionnalité de reprise

Exemple complet : flux de travail de transfert multi-agent

Définir les agents et les outils

Générer le workflow

Créer l’application FastAPI

Séquence d’événements

Réception d’accessoires transférés

Étapes suivantes

Ressources additionnelles

Commentaires

Ressources supplémentaires