Flujos de trabajo con AG-UI

MAF .NET puede exponer un flujo de trabajo mediante AG-UI convirtiendo el flujo de trabajo en un AIAgent y asignándolo como cualquier otro agente:

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

app.MapAGUIServer("/", workflowAgent);

El punto de conexión transmite en streaming el texto estándar y la salida de las llamadas a herramientas de los agentes que lo componen. AuthorName identifica el agente que generó cada actualización.

MAF .NET no asocia actualmente a AG-UI el comportamiento específico del ciclo de vida del flujo de trabajo. Los clientes no reciben eventos de paso de flujo de trabajo, instantáneas de actividad, interrupciones de flujo de trabajo ni operaciones de reanudación de flujo de trabajo equivalentes a la integración de Python. Encapsular un flujo de trabajo como AIAgent no añade esas asignaciones.

Para obtener el estado de seguimiento de .NET actual, consulte microsoft/agent-framework#2494. Para la construcción y ejecución de flujos de trabajo independientemente de AG-UI, consulte Conceptos de flujo de trabajo de MAF.

Pasos siguientes

Revisión de las consideraciones de producción y seguridad

En este tutorial se muestra cómo exponer flujos de trabajo de Agent Framework a través de un punto de conexión de AG-UI. Los flujos de trabajo organizan varios agentes y herramientas en un gráfico de ejecución definido y la AG-UI la integración transmite eventos de flujo de trabajo enriquecidos ( seguimiento de pasos, instantáneas de actividad, interrupciones y eventos personalizados) a los clientes web en tiempo real.

Prerequisites

Antes de comenzar, asegúrese de que tiene lo siguiente:

Cuándo usar flujos de trabajo con AG-UI

Use un flujo de trabajo en lugar de un único agente cuando necesite:

  • Orquestación multiagente: enrutar tareas entre agentes especializados (por ejemplo, clasificación → reembolso → pedido)
  • Pasos de ejecución estructurados: seguimiento del progreso a través de fases definidas con STEP_STARTED / STEP_FINISHED eventos
  • Flujos de interrupción y reanudación: pausar la ejecución para recopilar aprobaciones o entradas humanas y, a continuación, reanudar
  • Streaming de eventos personalizados: emita eventos específicos del dominio (request_info, status, workflow_output) al cliente.

Encapsulamiento de un flujo de trabajo con AgentFrameworkWorkflow

AgentFrameworkWorkflow es un contenedor ligero que adapta un nativo Workflow al protocolo AG-UI. Puede proporcionar una instancia de flujo de trabajo predefinida o una fábrica que cree un nuevo flujo de trabajo para cada subproceso.

Instancia directa

Use una instancia directa cuando un único objeto de flujo de trabajo pueda atender de forma segura todas las solicitudes (por ejemplo, canalizaciones sin 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 con ámbito de subproceso

Use workflow_factory cuando cada subproceso de conversación necesite su propio estado de flujo de trabajo. La fábrica recibe thread_id y devuelve un nuevo 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.",
)

Important

Debe pasar unoworkflowoworkflow_factory, no ambos. El contenedor lanza ValueError si se proporcionan ambos.

Registro del punto de conexión

Registre el flujo de trabajo con add_agent_framework_fastapi_endpoint de la misma manera que registraría un solo 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",
)

También puede pasar Workflow sin envoltorios directamente: el punto de conexión lo envuelve automáticamente en AgentFrameworkWorkflow.

add_agent_framework_fastapi_endpoint(app, my_workflow, "/workflow")

Eventos de AG-UI Emitidos por los Flujos de Trabajo

Las ejecuciones de flujo de trabajo emiten un conjunto más completo de eventos de AG-UI en comparación con las ejecuciones de agente único:

Event Cuando se emite Description
RUN_STARTED Inicio de la ejecución Marca el inicio de la ejecución del flujo de trabajo
STEP_STARTED Se inicia un ejecutor o superpaso step_name identifica el agente o el paso (por ejemplo, "triage_agent").
TEXT_MESSAGE_* El agente genera texto Eventos estándar de texto en streaming
TOOL_CALL_* El agente invoca una herramienta Eventos de llamada de herramienta estándar
REASONING_* El flujo de trabajo emite texto de un ejecutor configurado en intermediate_output_from Transmite texto intermedio como bloque de razonamiento. El alias de evento obsoleto "data" sigue el mismo camino.
STEP_FINISHED Un ejecutor o superstep finaliza Cierra el paso para el seguimiento del progreso de la interfaz de usuario.
CUSTOM (status) Cambios de estado de flujo de trabajo Contiene {"state": "<value>"} en el valor del evento.
CUSTOM (request_info) Solicitudes de flujo de trabajo que requieren intervención humana Contiene la carga de solicitud para que el cliente muestre una solicitud
CUSTOM (workflow_output) La salida del flujo de trabajo no se puede convertir al contenido del mensaje Contiene la salida serializada para el renderizado personalizado del cliente.
RUN_FINISHED Ejecución completada Incluye outcome.type == "interrupt" y outcome.interrupts cuando el flujo de trabajo está esperando la entrada

Los clientes pueden usar STEP_STARTED / STEP_FINISHED eventos para representar indicadores de progreso que muestran qué agente está activo actualmente. La integración cierra los bloques de texto y razonamiento abiertos antes de un evento terminal o una solicitud de entrada humana, por lo que los clientes reciben una secuencia de eventos completa.

Cuando se produce un error en un flujo de trabajo de Python, RUN_ERROR usa el mensaje Workflow execution failed. público genérico más un código de error. Un executor_failed evento también expone el mensaje genérico y el tipo de error. Los detalles internos de las excepciones y las trazas de error permanecen en los registros del servidor.

Interrupción y reanudación

Los flujos de trabajo pueden pausar la ejecución para recopilar aprobaciones de herramientas o entradas humanas. La integración de AG-UI controla esto a través del protocolo de interrupción y reanudación.

Funcionamiento de las interrupciones

  1. Durante la ejecución, el flujo de trabajo genera una solicitud pendiente (por ejemplo, una HandoffAgentUserRequest solicitud que solicita más detalles o una herramienta con approval_mode="always_require").

  2. El puente de AG-UI emite un CUSTOM evento que name="request_info" contiene los datos de solicitud.

  3. La ejecución finaliza con un RUN_FINISHED evento cuyo outcome.interrupts campo contiene las solicitudes pendientes:

    {
  "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. El cliente representa la interfaz de usuario para que el usuario responda (una entrada de texto, un botón de aprobación, etc.).

Cómo funciona resume

El cliente envía una nueva solicitud con una matriz canónica resume . Cada entrada identifica la interrupción y proporciona la respuesta del usuario:

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

El servidor convierte la carga de reanudación en respuestas de flujo de trabajo y continúa la ejecución desde donde se ha pausado. Para cancelar la ejecución interrumpida, establezca status en "cancelled" y omita payload.

Conservar y reanudar los puntos de control de flujo de trabajo

Configure checkpoint_storage en AgentFrameworkWorkflow para guardar el estado subyacente del flujo de trabajo al final de cada superpaso. En su lugar, puede pasar el mismo argumento a add_agent_framework_fastapi_endpoint al registrar un flujo de trabajo. Si el flujo de trabajo subyacente se creó con almacenamiento para puntos de control, el adaptador puede usar directamente ese constructor o el almacenamiento del entorno de ejecución, de modo que no es necesario duplicar la configuración en el envoltorio o el punto de conexión.

En el ejemplo siguiente se usa el almacenamiento en memoria para un flujo de trabajo de corta duración:

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

Cuando una ejecución se pausa y hay disponible un punto de control de pausa, cada interrupción del RUN_FINISHED evento incluye el identificador de punto de control en metadata.agent_framework.checkpoint_id. Utilice ese valor emitido para reanudar exactamente desde el punto de pausa en distintas instancias de la aplicación, sin necesidad de realizar una consulta independiente del último punto de control.

AgentFrameworkWorkflow.run()recibe la carga de solicitud AG-UI, por lo que un cliente proporciona el identificador de punto de control a través de propiedades reenviadas en lugar de un argumento Pythoncheckpoint_id. Una reanudación solo desde un punto de control no incluye un mensaje nuevo del usuario:

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

El adaptador restaura el estado de flujo de trabajo guardado y continúa la ejecución. Si el punto de control contiene una interrupción pendiente, incluya tanto el identificador de punto de control como la carga canónica resume en la misma solicitud. El adaptador restaura el punto de control antes de que entregue la respuesta de interrupción. Use el valor de metadata.agent_framework.checkpoint_id como forwardedProps.checkpointId.

El adaptador vincula cada nuevo punto de comprobación al ámbito de Snapshot de la solicitud y al threadId proporcionado por el cliente. Rechaza una solicitud de reanudación cuando cualquiera de los valores no coincide. Los puntos de control escritos antes de introducir metadatos de propiedad permanecen reanudables por motivos de compatibilidad.

Esta comprobación no reemplaza la autorización del punto de conexión ni el almacenamiento de punto de control protegido. Para obtener más información, consulte Consideraciones de seguridad.

InMemoryCheckpointStorage no sobrevive a los reinicios del proceso. Para obtener opciones de almacenamiento duraderas y selección de puntos de control, consulte Puntos de control.

Puntos de control de flujo de trabajo y instantáneas de subprocesos de AG-UI

Los puntos de control del flujo de trabajo y las instantáneas de hilos de AG-UI conservan datos diferentes:

Mecanismo de persistencia Tiendas propósito
Punto de control de flujo de trabajo de Agent Framework El ejecutor y el estado en tiempo de ejecución, incluidas las solicitudes pendientes Reanudación de la ejecución del flujo de trabajo desde el estado de tiempo de ejecución guardado
instantánea del hilo de AG-UI Salida de protocolo que se puede reproducir, como mensajes, estado compartido y la interrupción más reciente Rehidrata el hilo visible para el cliente

Puede configurar ambos mecanismos. Un punto de control de flujo de trabajo no reemplaza una instantánea de subproceso de AG-UI y una instantánea de subproceso de AG-UI no contiene el estado del ejecutor necesario para reanudar la ejecución del flujo de trabajo.

Ejemplo completo: Flujo de trabajo de entrega multiagente

En este ejemplo se muestra un flujo de trabajo de soporte al cliente con tres agentes que entregan trabajo entre sí, usan herramientas que requieren aprobación y solicitan entradas humanas cuando sea necesario.

Definir los agentes y las herramientas

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

Construir el flujo de trabajo

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

Creación de la aplicación 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)

Secuencia de eventos

Una interacción típica de varios turnos genera 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"}]}

Después, el cliente puede mostrar un cuadro de diálogo de aprobación y reanudarlo con la decisión del usuario.

Recibir propiedades enviadas

clientes AG-UI (como CopilotKit) pueden incluir un campo forwarded_props (o forwardedProps) en la carga de entrada. La integración de AG-UI pasa automáticamente estas propiedades al método run del flujo de trabajo mediante el argumento function_invocation_kwargs de palabra clave.

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.
        ...

Detalles clave:

  • Tanto forwarded_props como forwardedProps se aceptan en la carga de entrada; internamente se normalizan a forwarded_props.
  • Entre las propiedades reenviadas, checkpoint_id y checkpointId están reservadas para la reanudación del punto de control del flujo de trabajo.
  • Si workflow.run() no acepta function_invocation_kwargs (o **kwargs), las props se eliminan sin aviso; los flujos de trabajo existentes quedarán sin afectar.
  • Las propiedades reenviadas también se almacenan en metadatos de sesión, pero se excluyen de los metadatos vinculados a LLM, por lo que no se incluyen en las solicitudes del cliente de chat.

Pasos siguientes

Humano en el bucle

Recursos adicionales

Go puede exponer flujos de trabajo a AG-UI envolviendo un workflow.Workflow como agente con workflow/agentworkflow y, a continuación, alojando ese agente con 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

Consulte el flujo de trabajo como ejemplo de agente y el ejemplo de servidor deAG-UI para ver ejemplos completos ejecutables.

  • Last updated on