Fluxos de trabalho com AG-UI

Os fluxos de trabalho orquestram múltiplos agentes num grafo de execução definido. Em .NET expões um fluxo de trabalho sobre AG-UI exatamente da mesma forma que expões qualquer agente: converte-o para um AIAgent com AsAIAgent() e mapeia-o com MapAGUIServer. Não existe uma API de servidor específica para o fluxo de trabalho para aprender.

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

Warning

DefaultAzureCredential é conveniente para o desenvolvimento, mas requer uma consideração cuidadosa na produção. Em produção, considere usar uma credencial específica (por exemplo, ManagedIdentityCredential) para evitar problemas de latência, sondagens não intencionais de credenciais e potenciais riscos de segurança provenientes de mecanismos de recurso.

O lado do cliente mantém-se inalterado. Conecte-se AGUIChatClient como em Começar. Os fluxos de saída de cada agente são transmitidos normalmente AG-UI mensagens de texto e eventos de chamada de ferramenta, e AuthorName em cada atualização identifica qual agente no fluxo de trabalho o produziu.

Observação

A integração .NET transmite a saída do agente de um fluxo de trabalho (texto e chamadas de ferramentas) através do AG-UI, mas não os eventos de AG-UI específicos do fluxo de trabalho mostrados na versão Python deste artigo: acompanhamento de passos (STEP_STARTED / STEP_FINISHED), snapshots de atividade e interrupções ao nível do workflow. Esses eventos ainda estão a evoluir para o .NET (acompanhado pelo agent-framework#2494).

Passos seguintes

Este tutorial mostra-lhe como expor fluxos de trabalho do Agent Framework através de um endpoint AG-UI. Os fluxos de trabalho orquestram múltiplos agentes e ferramentas num grafo de execução definido, e a integração AG-UI transmite eventos ricos de workflow — rastreamento de passos, instantâneos de atividades, interrupções e eventos personalizados — para clientes web em tempo real.

Pré-requisitos

Antes de começar, certifique-se de que tem:

  • Python 3.10 ou posterior
  • agent-framework-ag-ui e agent-framework-foundry instalado
  • Familiaridade com o tutorial de Início
  • Compreensão básica dos conceitos de workflow do Agent Framework

Quando usar fluxos de trabalho com AG-UI

Use um fluxo de trabalho em vez de um único agente quando precisar:

  • Orquestração multi-agente: Encaminhar tarefas entre agentes especializados (por exemplo, triagem → reembolso → ordem)
  • Etapas estruturadas de execução: Acompanhar o progresso através de fases definidas com STEP_STARTED / STEP_FINISHED eventos
  • Interrupção / retomada de fluxos: Pausar a execução para recolher entradas humanas ou aprovações, e retomar
  • Streaming de eventos personalizados: Emitir eventos específicos do domínio (request_info, status, workflow_output) para o cliente

Envolver um Fluxo de Trabalho com o AgentFrameworkWorkflow

AgentFrameworkWorkflow é um wrapper leve que adapta um nativo Workflow ao protocolo AG-UI. Pode fornecer uma instância de workflow pré-configurada ou uma fábrica que crie um novo workflow por thread.

Instância direta

Use uma instância direta em que um único objeto de workflow possa servir todos os pedidos em segurança (por exemplo, pipelines sem estado):

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

Fábrica com escopo de rosca

Use workflow_factory quando cada thread de conversa precisar do seu próprio estado de fluxo de trabalho. A fábrica recebe o thread_id e devolve um novo 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

Deve passar ou umworkflowou outroworkflow_factory, mas não ambos. O invólucro gera um ValueError se ambos forem fornecidos.

Registo do Ponto Final

Registe o fluxo de trabalho add_agent_framework_fastapi_endpoint da mesma forma que registaria um único 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",
)

Também pode passar diretamente um "bare" Workflow — o endpoint envolve-o automaticamente em AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

AG-UI Eventos Emitidos pelos Fluxos de Trabalho

As execuções de workflow emitem um conjunto mais rico de eventos de AG-UI em comparação com as execuções de agente único:

Evento Quando emitido Description
RUN_STARTED A corrida começa Marca o início da execução do fluxo de trabalho
STEP_STARTED Um executor ou superstep é iniciado step_name identifica o agente ou passo (por exemplo, "triage_agent")
TEXT_MESSAGE_* Agente produz texto Eventos padrão de texto em streaming
TOOL_CALL_* O agente invoca uma ferramenta Eventos padrão de chamada de ferramenta
STEP_FINISHED Um executor ou superstep conclui Fecha o passo para o acompanhamento do progresso da interface de utilizador
CUSTOM (status) Alterações no estado do fluxo de trabalho Contém {"state": "<value>"} no valor do evento
CUSTOM (request_info) O fluxo de trabalho solicita a entrada humana Contém o payload de pedido para o cliente renderizar um prompt
CUSTOM (workflow_output) O fluxo de trabalho produz resultados Emitido tanto para eventos de terminal "output" como para eventos de workflow "intermediate". As saídas do terminal contêm a resposta final; as saídas intermédias aparecem como conteúdo text_reasoning quando o fluxo de trabalho é executado em segundo plano por trás de as_agent().
RUN_FINISHED Corrida concluída Inclui outcome.type == "interrupt" e outcome.interrupts quando o fluxo de trabalho está à espera de entrada

Os clientes podem usar STEP_STARTED / STEP_FINISHED eventos para renderizar indicadores de progresso que mostram qual agente está atualmente ativo.

Interromper e Retomar

Os fluxos de trabalho podem pausar a execução para recolher contributos humanos ou aprovações de ferramentas. A integração AG-UI trata disto através do protocolo de interrupção/reinício.

Como funcionam as interrupções

  1. Durante a execução, o fluxo de trabalho gera um pedido pendente (por exemplo, um HandoffAgentUserRequest a solicitar mais detalhes, ou uma ferramenta com approval_mode="always_require").

  2. A ponte AG-UI emite um evento CUSTOM com name="request_info" que contém os dados do pedido.

  3. A execução termina com um evento RUN_FINISHED em que o campo outcome.interrupts contém as solicitações pendentes:

    {
  "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. O cliente renderiza a interface para o utilizador responder (uma entrada de texto, um botão de aprovação, etc.).

Como funciona o currículo

O cliente envia um novo pedido com um array canónico resume . Cada entrada identifica a interrupção e fornece a resposta do utilizador:

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

O servidor converte a carga útil de retomada em respostas do fluxo de trabalho e continua a execução a partir do ponto em que parou. Para cancelar a execução interrompida, defina status para "cancelled" e omita payload.

Exemplo Completo: Fluxo de Trabalho de Transferência Multi-Agente

Este exemplo mostra um fluxo de trabalho de apoio ao cliente com três agentes que transferem trabalho entre si, utilizam ferramentas que requerem aprovação e solicitam intervenção humana quando necessário.

Defina os agentes e ferramentas

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

Criar o fluxo de trabalho

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

Criar a aplicação 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)

Sequência de eventos

Uma interação típica com vários turnos produz eventos como:

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

O cliente pode então mostrar um diálogo de aprovação e retomar com a decisão do utilizador.

Receção de Propriedades Encaminhadas

clientes de AG-UI (como o CopilotKit) podem incluir um campo `forwarded_props` (ou `forwardedProps`) na entrada de carga útil. A integração AG-UI passa automaticamente estas propriedades para o método do fluxo de trabalho run através do argumento de palavra-chave 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.
        ...

Detalhes principais:

  • Tanto forwarded_props como forwardedProps são aceites na carga útil de entrada; internamente são normalizados para forwarded_props.
  • Se workflow.run() não aceitar function_invocation_kwargs (ou **kwargs), as props são silenciosamente descartadas — os fluxos de trabalho existentes não são afetados.
  • Os props encaminhados também são armazenados nos metadados da sessão, mas são filtrados a partir dos metadados ligados ao LLM, para não serem infiltrados nos pedidos do cliente de chat.

Passos seguintes

Humano no Ciclo de Decisão

Recursos adicionais

Go consegue disponibilizar fluxos de trabalho ao AG-UI ao encapsular um workflow.Workflow como agente com workflow/agentworkflow e, em seguida, alojar esse agente com 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

Veja o workflow como exemplo de agente e o exemplo de servidor AG-UI para exemplos completos e executáveis.

  • Last updated on