Flussi di lavoro con AG-UI

MAF .NET può esporre un flusso di lavoro tramite AG-UI convertendo il flusso di lavoro in un AIAgent e mappandolo come qualsiasi altro agente:

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

app.MapAGUIServer("/", workflowAgent);

L'endpoint trasmette in streaming il testo standard e l'output delle chiamate degli strumenti degli agenti che lo compongono. AuthorName identifica l'agente che ha prodotto ogni aggiornamento.

MAF .NET attualmente non mappa in AG-UI il comportamento del ciclo di vita specifico del workflow. I client non ricevono eventi delle fasi del flusso di lavoro, istantanee delle attività, interruzioni del flusso di lavoro o operazioni di ripresa del flusso di lavoro analoghe a quelle dell'integrazione Python. Racchiudere un flusso di lavoro come AIAgent non aggiunge tali mappature.

Per lo stato attuale del monitoraggio di .NET, vedere microsoft/agent-framework#2494. Per la costruzione e l'esecuzione di workflow indipendenti da AG-UI, vedere Concetti dei workflow MAF.

Passaggi successivi

Esaminare le considerazioni sulla produzione e sulla sicurezza

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.

Prerequisites

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 Attività iniziali
  • 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 del flusso di lavoro può gestire in modo sicuro tutte le richieste, ad esempio pipeline senza stato:

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 workflowoppureworkflow_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:

Event Quando viene emesso Descrizione
RUN_STARTED Inizio dell'esecuzione Contrassegna l'inizio dell'esecuzione del flusso di lavoro
STEP_STARTED Un executor o un superstep inizia l'esecuzione 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 Un executor o un superstep completa l'esecuzione 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 recupero

Il client invia una nuova richiesta con una matrice canonica resume . Ogni voce identifica l'interruzione e fornisce 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 del recupero 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.

Salvare e riprendere i checkpoint del flusso di lavoro

Configurare checkpoint_storage su AgentFrameworkWorkflow per salvare lo stato del flusso di lavoro sottostante alla fine di ogni passaggio superiore. È invece possibile passare lo stesso argomento a add_agent_framework_fastapi_endpoint quando si registra un flusso di lavoro. L'archiviazione deve essere disponibile per il wrapper o l'endpoint AG-UI per riprendere l'esecuzione da un checkpoint tramite AG-UI.

L'esempio seguente usa l'archiviazione in memoria per un flusso di lavoro di breve durata:

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()riceve il payload della richiesta AG-UI, quindi un client fornisce l'ID del checkpoint tramite proprietà inoltrate anziché un argomento Python checkpoint_id. Una ripresa basata solo su checkpoint non include un nuovo messaggio dell'utente:

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

L'adattatore ripristina lo stato del flusso di lavoro salvato e continua l'esecuzione. Se il checkpoint contiene un interrupt in sospeso, includere sia l'ID del checkpoint che il payload canonico resume nella stessa richiesta. L'adattatore ripristina il checkpoint prima di inviare la risposta all'interruzione.

L'adattatore associa ogni nuovo checkpoint allo Snapshot Scope della richiesta e all'elemento fornito dal client threadId. Rifiuta una richiesta di ripresa quando uno dei due valori non corrisponde. I checkpoint scritti prima dell'introduzione dei metadati di proprietà rimangono ripristinabili per la compatibilità.

Questo controllo non sostituisce l'autorizzazione degli endpoint né l'archiviazione protetta dei checkpoint. Per altre informazioni, vedere considerazioni sulla sicurezza .

InMemoryCheckpointStorage non sopravvive ai riavvii del processo. Per le opzioni di archiviazione persistente e la selezione dei checkpoint, vedere Checkpoints.

Checkpoint del workflow e snapshot dei thread AG-UI

I checkpoint del flusso di lavoro e gli snapshot dei thread AG-UI conservano dati diversi:

Meccanismo di persistenza Negozi Purpose
Punto di controllo del flusso di lavoro di Agent Framework Executor e stato di esecuzione, comprese le richieste in sospeso Riprendere l'esecuzione del flusso di lavoro dallo stato di runtime salvato
snapshot del thread AG-UI Output del protocollo ripetibile, ad esempio i messaggi, lo stato condiviso e l’ultimo interrupt Riattivare il thread visibile al client

È possibile configurare entrambi i meccanismi. Un checkpoint del flusso di lavoro non sostituisce uno snapshot del thread AG-UI e uno snapshot del thread AG-UI non contiene lo stato dell'executor necessario per riprendere l'esecuzione del flusso di lavoro.

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 proprietà inoltrate

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 run del flusso di lavoro tramite l'argomento di parola chiave 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.
  • Tra le proprietà inoltrate, checkpoint_id e checkpointId sono riservati alla ripresa del checkpoint del flusso di lavoro.
  • 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 all'LLM, quindi non vengono perse nelle richieste client di chat.

Passaggi successivi

Intervento umano nel 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{}))

Tip

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

  • Last updated on