Fluxos de trabalho com AG-UI

O MAF .NET pode expor um fluxo de trabalho através de AG-UI convertendo o fluxo de trabalho para um AIAgent e mapeando-o como qualquer outro agente:

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

app.MapAGUIServer("/", workflowAgent);

O endpoint transmite em fluxo o texto padrão e o resultado das chamadas de ferramenta dos agentes que o constituem. AuthorName Identifica o agente que produziu cada atualização.

O MAF .NET de momento não mapeia o comportamento específico do ciclo de vida do fluxo de trabalho para a AG-UI. Os clientes não recebem eventos das etapas do fluxo de trabalho, instantâneos de atividades, interrupções do fluxo de trabalho ou operações de retomação do fluxo de trabalho equivalentes às da integração em Python. Encapsular um fluxo de trabalho como AIAgent não adiciona esses mapeamentos.

Para o estado atual do rastreio .NET, consulte microsoft/agent-framework #2494. Para construção e execução de workflows independentes do AG-UI, veja conceitos de workflow MAF.

Passos seguintes

Revisão das considerações de produção e segurança

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:

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
REASONING_* O fluxo de trabalho emite texto de um executor configurado em intermediate_output_from Transmite texto intermédio como bloco de raciocínio. O alias de evento obsoleto "data" segue o mesmo caminho.
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) A saída do fluxo de trabalho não pode ser convertida em conteúdo de mensagem Contém a saída serializada para renderização personalizada do cliente.
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. A integração fecha blocos de raciocínio aberto e texto antes de um evento terminal ou pedido de entrada humana, pelo que os clientes recebem uma sequência completa de eventos.

Quando um fluxo de trabalho em Python falha, RUN_ERROR usa a mensagem Workflow execution failed. pública genérica mais um código de erro. Um executor_failed evento também expõe a mensagem genérica e o tipo de erro. Detalhes internos de exceções e rastreamentos permanecem nos registos do servidor.

Interrupção e retomada

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.

Persistir e retomar os pontos de verificação do fluxo de trabalho

Configure checkpoint_storage em AgentFrameworkWorkflow para guardar o estado subjacente do fluxo de trabalho no final de cada superetapa. Em vez disso, pode passar o mesmo argumento ao add_agent_framework_fastapi_endpoint quando registar um fluxo de trabalho. Se o fluxo de trabalho subjacente foi criado com armazenamento de pontos de verificação, o adaptador pode usar esse construtor ou o armazenamento em tempo de execução diretamente, pelo que não é necessário duplicar a configuração no encapsulador ou no ponto terminal.

O exemplo seguinte utiliza armazenamento em memória para um fluxo de trabalho de curta duração:

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

Quando uma execução é pausada e está disponível um ponto de controlo de pausa, cada interrupção no evento RUN_FINISHED inclui o ID desse ponto de controlo em metadata.agent_framework.checkpoint_id. Use esse valor emitido para retomar a pausa exata entre instâncias da aplicação sem uma pesquisa separada de ponto de verificação mais recente.

AgentFrameworkWorkflow.run() recebe a carga útil do pedido AG-UI, pelo que o cliente fornece o ID do ponto de controlo através de propriedades reencaminhadas, em vez de um argumento Python checkpoint_id. Uma retoma apenas de um ponto de controlo não inclui uma nova mensagem do utilizador:

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

O adaptador restaura o estado do fluxo de trabalho guardado e continua a execução. Se o checkpoint contiver uma interrupção pendente, inclua tanto o ID do checkpoint como a carga canónica resume no mesmo pedido. O adaptador restaura o ponto de verificação antes de emitir a resposta à interrupção. Use o valor de metadata.agent_framework.checkpoint_id como forwardedProps.checkpointId.

O adaptador associa cada novo checkpoint ao Snapshot Scope da solicitação e ao threadId fornecido pelo cliente. Rejeita um pedido de currículo quando nenhum dos valores corresponde. Os checkpoints escritos antes da introdução dos metadados de propriedade continuam a ser retomáveis para compatibilidade.

Esta verificação não substitui a autorização do terminal nem o armazenamento protegido do ponto de verificação. Para obter mais informações, consulte Considerações de segurança.

InMemoryCheckpointStorage Não sobrevive a reinícios de processos. Para opções de armazenamento duráveis e seleção de pontos de controlo, consulte Pontos de controlo.

Pontos de controlo do fluxo de trabalho e instantâneos de threads do AG-UI

Os checkpoints de fluxo de trabalho e os snapshots AG-UI thread mantêm dados diferentes:

Mecanismo de persistência Lojas Purpose
Ponto de verificação do fluxo de trabalho do Agent Framework Estado do executor e de execução, incluindo pedidos pendentes Retomar a execução do fluxo de trabalho a partir do estado de execução guardado
captura da thread AG-UI Resultado do protocolo reproduzível, como mensagens, estado partilhado e a interrupção mais recente Reidrate o fio visível pelo cliente

Podes configurar ambos os mecanismos. Um checkpoint de workflow não substitui um snapshot de thread AG-UI, e esse snapshot de thread AG-UI não contém o estado do executor necessário para retomar a execução do workflow.

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.
  • Nas propriedades reencaminhadas, checkpoint_id e checkpointId estão reservados para a retoma de um ponto de controlo do fluxo de trabalho.
  • 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.

  • Última atualização a