Fluxos de trabalho com AG-UI

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 instalado
• Familiaridade com o tutorial de Início
• 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 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.",
)

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:

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 RUN_FINISHED evento cujo interrupts campo contém uma lista de objetos de pedido 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 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 o resume payload contendo as respostas do utilizador codificadas pelo ID da interrupção:

{
  "threadId": "abc123",
  "messages": [],
  "resume": {
    "interrupts": [
      {
        "id": "request-id-1",
        "value": "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.

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.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 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          interrupts=[{id: "...", value: {function_approval_request}}]

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

Passos seguintes

Recursos adicionais

• Visão geral do AG-UI
• Começar
• Agent Framework Workflows
• Repositório GitHub do Agent Framework