Flussi di lavoro con AG-UI

I flussi di lavoro orchestrano più agenti in un grafico di esecuzione definito. In .NET si espone un flusso di lavoro su AG-UI esattamente come si espone qualsiasi agente: convertirlo in un AIAgent oggetto con AsAIAgent() ed eseguirne il mapping con MapAGUIServer. Non è disponibile alcuna API server specifica del flusso di lavoro da apprendere.

using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.Agents.AI.Workflows;
using OpenAI.Chat;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddAGUIServer();

string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"]
    ?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");

ChatClient chatClient = new AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName);

AIAgent researcher = chatClient.AsAIAgent(
    name: "researcher", instructions: "Research the user's topic and write a short, factual brief.");
AIAgent reporter = chatClient.AsAIAgent(
    name: "reporter", instructions: "Summarize the researcher's brief into a single clear paragraph.");

// A workflow-as-agent is just an AIAgent. Map it like any other agent.
AIAgent workflowAgent = AgentWorkflowBuilder.BuildSequential(researcher, reporter).AsAIAgent();

WebApplication app = builder.Build();
app.MapAGUIServer("/", workflowAgent);

await app.RunAsync();

Avvertimento

DefaultAzureCredential è utile per lo sviluppo, ma richiede un'attenta considerazione nell'ambiente di produzione. Nell'ambiente di produzione prendere in considerazione l'uso di credenziali specifiche ,ad esempio ManagedIdentityCredential, per evitare problemi di latenza, probe di credenziali indesiderate e potenziali rischi per la sicurezza dai meccanismi di fallback.

Il lato client è invariato. Connettersi con AGUIChatClient come in Introduzione. I flussi di output di ogni agente come normali AG-UI eventi di testo e chiamata agli strumenti e in ogni aggiornamento identificano l'agente AuthorName nel flusso di lavoro prodotto.

Annotazioni

L'integrazione .NET trasmette l'output dell'agente di un flusso di lavoro (chiamate di testo e strumenti) tramite l'interfaccia utente del gruppo di disponibilità, ma non gli eventi AG-UI specifici del flusso di lavoro visualizzati nella versione Python di questo articolo: rilevamento dei passaggi (STEP_STARTED / STEP_FINISHED), snapshot di attività e interrupt a livello di flusso di lavoro. Questi eventi sono ancora in continua evoluzione per .NET (rilevati da agent-framework#2494).

Passaggi successivi

Questa esercitazione illustra come esporre i flussi di lavoro di Agent Framework tramite un endpoint AG-UI. I flussi di lavoro orchestrano più agenti e strumenti in un grafo di esecuzione definito, e l'integrazione AG-UI trasmette eventi avanzati dei flussi di lavoro — come il tracciamento dei passi, gli snapshot delle attività, le interruzioni e gli eventi personalizzati — ai client web in tempo reale.

Prerequisiti

Prima di iniziare, assicurati di avere:

  • Python 3.10 o versione successiva
  • agent-framework-ag-ui e agent-framework-foundry installati
  • Familiarità con l'esercitazione introduttiva
  • Conoscenza di base dei concetti relativi al flusso di lavoro di Agent Framework

Quando usare flussi di lavoro con AG-UI

Usare un flusso di lavoro anziché un singolo agente quando necessario:

  • Orchestrazione multi-agente: Smistare le attività fra agenti specializzati (ad esempio, valutazione → rimborso → ordine)
  • Passaggi di esecuzione strutturati: tenere traccia dello stato di avanzamento nelle fasi definite con STEP_STARTED / STEP_FINISHED eventi
  • Interrompere/riprendere i flussi: sospendere l'esecuzione per raccogliere l'input umano o le approvazioni, quindi riprendere
  • Streaming di eventi personalizzati: generare eventi specifici del dominio (request_info, status, workflow_output) al client

Avvolgimento di un flusso di lavoro con AgentFrameworkWorkflow

AgentFrameworkWorkflow è un wrapper leggero che adatta un elemento nativo Workflow al protocollo AG-UI. È possibile fornire un'istanza del flusso di lavoro predefinita o una factory che crea un nuovo flusso di lavoro per ogni thread.

Istanza diretta

Usare un'istanza diretta quando un singolo oggetto di flusso di lavoro può gestire in modo sicuro tutte le richieste, ad esempio pipeline stateless:

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 con ambito thread

Usare workflow_factory quando ogni thread di conversazione necessita del proprio stato del flusso di lavoro. La fabbrica riceve thread_id e restituisce un nuovo 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.",
)

Importante

È necessario passare oworkflowoppureworkflow_factory, non entrambi. Il wrapper genera un oggetto ValueError se vengono forniti entrambi.

Registrazione dell'endpoint

Registrare il flusso di lavoro con add_agent_framework_fastapi_endpoint lo stesso modo in cui si registra un singolo agente:

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

È anche possibile passare direttamente un semplice Workflow — l'endpoint lo avvolge automaticamente in AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

eventi AG-UI generati dai flussi di lavoro

Le esecuzioni del flusso di lavoro generano un set più completo di eventi AG-UI rispetto alle esecuzioni con agente singolo:

Evento Quando viene emesso Description
RUN_STARTED Inizio del processo Contrassegna l'inizio dell'esecuzione del flusso di lavoro
STEP_STARTED Inizia un executor o un superstep step_name identifica l'agente o il passaggio (ad esempio, "triage_agent")
TEXT_MESSAGE_* L'agente produce testo Eventi standard di streaming di testo
TOOL_CALL_* Agent richiama uno strumento Eventi di chiamata agli strumenti standard
STEP_FINISHED Completa un executor o un superstep Chiude il passaggio per il rilevamento dello stato dell'interfaccia utente
CUSTOM (status) Modifiche dello stato del flusso di lavoro Contiene {"state": "<value>"} nel valore dell'evento
CUSTOM (request_info) Il flusso di lavoro richiede l'input umano Contiene il payload della richiesta affinché il client possa visualizzare un prompt
CUSTOM (workflow_output) Il flusso di lavoro produce risultati Generato sia per gli eventi "output" (terminale) sia per gli eventi "intermediate" del flusso di lavoro. Gli output del terminale riportano la risposta finale; quelli intermedi vengono visualizzati sotto forma di contenuto text_reasoning quando il flusso di lavoro viene eseguito dietro as_agent().
RUN_FINISHED Esecuzione completata Include outcome.type == "interrupt" e outcome.interrupts quando il flusso di lavoro è in attesa di un input

I client possono usare gli eventi STEP_STARTED / STEP_FINISHED per mostrare gli indicatori di avanzamento che mostrano quale agente è attualmente attivo.

Interrompi e riprendi

I flussi di lavoro possono sospendere l'esecuzione per raccogliere l'input umano o le approvazioni degli strumenti. L'integrazione AG-UI gestisce questa operazione tramite il protocollo interrupt/resume.

Funzionamento degli interrupt

  1. Durante l'esecuzione, il flusso di lavoro genera una richiesta in sospeso( ad esempio, una HandoffAgentUserRequest richiesta di altri dettagli o uno strumento con approval_mode="always_require").

  2. Il bridge AG-UI genera un CUSTOM evento contenente name="request_info" i dati della richiesta.

  3. L'esecuzione si conclude con un evento RUN_FINISHED il cui campo outcome.interrupts contiene le richieste in sospeso:

    {
  "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. Il client esegue il rendering dell'interfaccia utente per consentire all'utente di rispondere (un input di testo, un pulsante di approvazione e così via).

Come funziona il curriculum

Il client invia una nuova richiesta con una matrice canonica resume . Ogni voce identifica l'interruzione e riporta la risposta dell'utente:

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

Il server converte il payload resume in risposte del flusso di lavoro e continua l'esecuzione da dove è stata sospesa. Per annullare invece l'esecuzione interrotta, impostare status su "cancelled" e omettere payload.

Esempio completo: Flusso di lavoro di trasferimento multi-agente

Questo esempio mostra un flusso di lavoro di supporto clienti con tre agenti che si consegnano l'uno all'altro, usano strumenti che richiedono l'approvazione e richiedono l'input umano quando necessario.

Definire gli agenti e gli strumenti

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

Costruire il flusso di lavoro

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

Creare l'app 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)

Sequenza di eventi

Un'interazione a più turni tipica produce eventi come:

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

Il client può quindi visualizzare una finestra di dialogo di approvazione e riprendere con la decisione dell'utente.

Ricezione di oggetti prop inoltrati

AG-UI client (ad esempio CopilotKit) possono includere un forwarded_props campo (o forwardedProps) nel payload di input. L'integrazione AG-UI trasmette automaticamente queste proprietà al metodo del flusso di lavoro di run tramite l'argomento keyword 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.
        ...

Dettagli chiave:

  • Sia forwarded_props che forwardedProps vengono accettati nel payload di input; internamente vengono normalizzati in forwarded_props.
  • Se workflow.run() non accetta function_invocation_kwargs (o **kwargs), le proprietà vengono eliminate automaticamente. I flussi di lavoro esistenti non sono interessati.
  • Le proprietà inoltrate vengono archiviate nei metadati della sessione, ma sono filtrate dai metadati associati a LLM, quindi non si infiltrano nelle richieste client di chat.

Passaggi successivi

Human-in-the-Loop

Risorse aggiuntive

Go può esporre i flussi di lavoro ad AG-UI racchiudendo un workflow.Workflow come agente con workflow/agentworkflow, quindi ospitando l’agente con 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{}))

Suggerimento

Consulta il workflow come esempio di agente e il server di esempio AG-UI per esempi completi e pronti all'esecuzione.

  • Last updated on