Fluxos de trabalho com AG-UI

Observação

O suporte ao fluxo de trabalho para a integração .NET AG-UI será disponibilizado em breve.

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

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:

Event	Quando emitido	Descrição
`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	Contém os dados finais ou intermédios de saída
`RUN_FINISHED`	Corrida concluída	Pode incluir `interrupts` se o fluxo de trabalho estiver à 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

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").
A ponte AG-UI emite um evento CUSTOM com name="request_info" que contém os dados do pedido.

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

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

Humano no Ciclo de Decisão

Recursos adicionais

Comentários

Esta página foi útil?

Last updated on 2026-04-09

Partilhar via

Fluxos de trabalho com AG-UI

Pré-requisitos

Quando usar fluxos de trabalho com AG-UI

Envolver um Fluxo de Trabalho com o AgentFrameworkWorkflow

Instância direta

Fábrica com escopo de rosca

Registo do Ponto Final

AG-UI Eventos Emitidos pelos Fluxos de Trabalho

Interromper e Retomar

Como funcionam as interrupções

Como funciona o currículo

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

Defina os agentes e ferramentas

Criar o fluxo de trabalho

Criar a aplicação FastAPI

Sequência de eventos

Passos seguintes

Recursos adicionais

Comentários

Recursos adicionais