Fluxos de trabalho com AG-UI

Este tutorial mostra como expor workflows do Agent Framework por meio de um endpoint AG-UI. Os fluxos de trabalho orquestram vários agentes e ferramentas em um grafo de execução definido e a integração AG-UI transmite eventos avançados de fluxo de trabalho – acompanhamento de etapas, instantâneos de atividade, interrupções e eventos personalizados – para clientes Web em tempo real.

Pré-requisitos

Antes de começar, verifique se você tem:

• Python 3.10 ou posterior
• agent-framework-ag-ui Instalado
• Familiaridade com o tutorial de Introdução
• Compreensão básica dos fluxos de trabalho 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 de vários agentes: rotear tarefas entre agentes especializados (por exemplo, triagem → pedido de reembolso →)
• Etapas de execução estruturadas: acompanhar o progresso por meio de estágios definidos com STEP_STARTED / STEP_FINISHED eventos
• Interrupção/retomada de fluxos: pausar a execução para coletar aprovações ou entradas humanas e, em seguida, retomar
• Streaming de eventos personalizados: emitir eventos específicos do domínio (request_info, status, workflow_output) para o cliente

Encapsulando um fluxo de trabalho com AgentFrameworkWorkflow

AgentFrameworkWorkflow é um wrapper leve que adapta um nativo Workflow ao protocolo AG-UI. Você pode fornecer uma instância de fluxo de trabalho pré-criada ou uma fábrica que cria um novo fluxo de trabalho por thread.

Instância direta

Utilize uma instância direta quando um único objeto de processo de trabalho puder gerenciar com segurança todas as solicitações (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 restrito a thread

Use workflow_factory quando cada tópico de conversa precisar de seu próprio estado de fluxo de trabalho. A fábrica recebe o thread_id e retorna 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.",
)

Registrando o ponto de extremidade

Registre o fluxo de trabalho com add_agent_framework_fastapi_endpoint da mesma forma que você registraria um agente único.

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

Você também pode passar um simples Workflow diretamente - o endpoint o encapsula automaticamente em AgentFrameworkWorkflow:

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

Eventos AG-UI Emitidos por Fluxos de Trabalho

As execuções de fluxo de trabalho emitem um conjunto mais avançado de eventos de AG-UI em comparação com execuções de agente único:

Os clientes podem usar STEP_STARTED / STEP_FINISHED eventos para renderizar indicadores de progresso mostrando qual agente está ativo no momento.

Interromper e retomar

Os fluxos de trabalho podem pausar a execução para coletar entrada de humanos ou aprovações de ferramentas. A integração AG-UI lida com isso por meio do protocolo de interrupção/retomada.

Como as interrupções funcionam

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

2. A ponte AG-UI emite um CUSTOM evento com name="request_info" contendo os dados da solicitação.

3. A execução é concluída com um RUN_FINISHED evento cujo interrupts campo contém uma lista de objetos de solicitação pendentes:

   {
     "type": "RUN_FINISHED",
     "threadId": "abc123",
     "runId": "run_xyz",
     "interrupts": [
       {
         "id": "request-id-1",
         "value": { "request_type": "HandoffAgentUserRequest", "data": "..." }
       }
     ]
   }
   

4. O cliente renderiza a interface do usuário para que o usuário responda (uma entrada de texto, um botão de aprovação etc.).

Como o currículo funciona

O cliente envia uma nova solicitação com o resume conteúdo que contém as respostas do usuário chaveadas pela ID de interrupção:

{
  "threadId": "abc123",
  "messages": [],
  "resume": {
    "interrupts": [
      {
        "id": "request-id-1",
        "value": "User's response text or approval decision"
      }
    ]
  }
}

O servidor converte o conteúdo do currículo em respostas de fluxo de trabalho e continua a execução de onde ele fez uma pausa.

Exemplo completo: fluxo de trabalho de repasse entre múltiplos agentes

Este exemplo mostra um fluxo de trabalho de suporte ao cliente com três agentes que entregam trabalho uns aos outros, usam ferramentas que exigem aprovação e solicitam entrada humana quando necessário.

Definir os agentes e as 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.azure import AzureOpenAIResponsesClient
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 = AzureOpenAIResponsesClient(
        project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
        deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
        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 o aplicativo 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 de 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          interrupts=[{id: "...", value: {function_approval_request}}]

Em seguida, o cliente pode exibir uma caixa de diálogo de aprovação e retomar com a decisão do usuário.

Próximas Etapas 

Recursos adicionais

• Visão geral doAG-UI
• Introdução
• Fluxos de trabalho do Agent Framework
• Repositório do GitHub Agent Framework