Flujos de trabajo de Microsoft Agent Framework: uso de flujos de trabajo como agentes

Requisitos

Para usar un flujo de trabajo como agente, el ejecutor de inicio del flujo de trabajo debe poder manejar IEnumerable<ChatMessage> como entrada. Esto se satisface automáticamente al usar ChatClientAgent u otros ejecutores basados en agente.

Creación de un agente de flujo de trabajo

Use el método de AsAgent() extensión para convertir cualquier flujo de trabajo compatible en un agente:

using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

// First, build your workflow
var workflow = AgentWorkflowBuilder
    .CreateSequentialPipeline(researchAgent, writerAgent, reviewerAgent)
    .Build();

// Convert the workflow to an agent
AIAgent workflowAgent = workflow.AsAgent(
    id: "content-pipeline",
    name: "Content Pipeline Agent",
    description: "A multi-agent workflow that researches, writes, and reviews content"
);

Parámetros de AsAgent

Uso de agentes de flujo de trabajo

Creación de un hilo

Cada conversación con un agente de flujo de trabajo requiere un subproceso para administrar el estado:

// Create a new thread for the conversation
AgentThread thread = workflowAgent.GetNewThread();

Ejecución sin streaming

Para casos de uso sencillos en los que desee la respuesta completa:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

AgentRunResponse response = await workflowAgent.RunAsync(messages, thread);

foreach (ChatMessage message in response.Messages)
{
    Console.WriteLine($"{message.AuthorName}: {message.Text}");
}

Ejecución de streaming

Para las actualizaciones en tiempo real a medida que se ejecuta el flujo de trabajo:

var messages = new List<ChatMessage>
{
    new(ChatRole.User, "Write an article about renewable energy trends in 2025")
};

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Process streaming updates from each agent in the workflow
    if (!string.IsNullOrEmpty(update.Text))
    {
        Console.Write(update.Text);
    }
}

Control de solicitudes de entrada externas

Cuando un flujo de trabajo contiene ejecutores que solicitan entrada externa (mediante RequestInfoExecutor), estas solicitudes se muestran como llamadas de función en la respuesta del agente:

await foreach (AgentRunResponseUpdate update in workflowAgent.RunStreamingAsync(messages, thread))
{
    // Check for function call requests
    foreach (AIContent content in update.Contents)
    {
        if (content is FunctionCallContent functionCall)
        {
            // Handle the external input request
            Console.WriteLine($"Workflow requests input: {functionCall.Name}");
            Console.WriteLine($"Request data: {functionCall.Arguments}");

            // Provide the response in the next message
        }
    }
}

Serialización y reanudación de hilos

Los subprocesos del agente de flujo de trabajo se pueden serializar para la persistencia y reanudarse más adelante:

// Serialize the thread state
JsonElement serializedThread = thread.Serialize();

// Store serializedThread to your persistence layer...

// Later, resume the thread
AgentThread resumedThread = workflowAgent.DeserializeThread(serializedThread);

// Continue the conversation
await foreach (var update in workflowAgent.RunStreamingAsync(newMessages, resumedThread))
{
    Console.Write(update.Text);
}

Puntos de control con agentes de flujo de trabajo

Habilite los puntos de control para conservar el estado del flujo de trabajo en los reinicios del proceso:

// Create a checkpoint manager with your storage backend
var checkpointManager = new CheckpointManager(new FileCheckpointStorage("./checkpoints"));

// Create workflow agent with checkpointing enabled
AIAgent workflowAgent = workflow.AsAgent(
    id: "persistent-workflow",
    name: "Persistent Workflow Agent",
    checkpointManager: checkpointManager
);

Requisitos

Para usar un flujo de trabajo como agente, el ejecutor de inicio del flujo de trabajo debe poder controlarlo list[ChatMessage] como entrada. Esto se cumple automáticamente cuando se usa ChatAgent o AgentExecutor.

Creación de un agente de flujo de trabajo

Llame a as_agent() en cualquier flujo de trabajo compatible para convertirlo en un agente:

from agent_framework import WorkflowBuilder, ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Create your chat client and agents
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

researcher = ChatAgent(
    name="Researcher",
    instructions="Research and gather information on the given topic.",
    chat_client=chat_client,
)

writer = ChatAgent(
    name="Writer", 
    instructions="Write clear, engaging content based on research.",
    chat_client=chat_client,
)

# Build your workflow
workflow = (
    WorkflowBuilder()
    .set_start_executor(researcher)
    .add_edge(researcher, writer)
    .build()
)

# Convert the workflow to an agent
workflow_agent = workflow.as_agent(name="Content Pipeline Agent")

parámetros de as_agent

Uso de agentes de flujo de trabajo

Creación de un hilo

Cada conversación con un agente de flujo de trabajo requiere un subproceso para administrar el estado:

# Create a new thread for the conversation
thread = workflow_agent.get_new_thread()

Ejecución sin streaming

Para casos de uso sencillos en los que desee la respuesta completa:

from agent_framework import ChatMessage, Role

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

response = await workflow_agent.run(messages, thread=thread)

for message in response.messages:
    print(f"{message.author_name}: {message.text}")

Ejecución de streaming

Para las actualizaciones en tiempo real a medida que se ejecuta el flujo de trabajo:

messages = [ChatMessage(role=Role.USER, content="Write an article about AI trends")]

async for update in workflow_agent.run_stream(messages, thread=thread):
    # Process streaming updates from each agent in the workflow
    if update.text:
        print(update.text, end="", flush=True)

Control de solicitudes de entrada externas

Cuando un flujo de trabajo contiene ejecutores que solicitan entrada externa (mediante RequestInfoExecutor), estas solicitudes se muestran como llamadas de función. El agente de flujo de trabajo realiza un seguimiento de las solicitudes pendientes y espera respuestas antes de continuar:

from agent_framework import (
    FunctionCallContent,
    FunctionApprovalRequestContent,
    FunctionApprovalResponseContent,
)

async for update in workflow_agent.run_stream(messages, thread=thread):
    for content in update.contents:
        if isinstance(content, FunctionApprovalRequestContent):
            # The workflow is requesting external input
            request_id = content.id
            function_call = content.function_call

            print(f"Workflow requests input: {function_call.name}")
            print(f"Request data: {function_call.arguments}")

            # Store the request_id to provide a response later

# Check for pending requests
if workflow_agent.pending_requests:
    print(f"Pending requests: {list(workflow_agent.pending_requests.keys())}")

Proporcionar respuestas a solicitudes pendientes

Para continuar la ejecución del flujo de trabajo después de una solicitud de entrada externa:

# Create a response for the pending request
response_content = FunctionApprovalResponseContent(
    id=request_id,
    function_call=function_call,
    approved=True,
)

response_message = ChatMessage(
    role=Role.USER,
    contents=[response_content],
)

# Continue the workflow with the response
async for update in workflow_agent.run_stream([response_message], thread=thread):
    if update.text:
        print(update.text, end="", flush=True)

Ejemplo completo

Este es un ejemplo completo que muestra un agente de flujo de trabajo con salida de streaming:

import asyncio
from agent_framework import (
    ChatAgent,
    ChatMessage,
    Role,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework._workflows import SequentialBuilder
from azure.identity import AzureCliCredential


async def main():
    # Set up the chat client
    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

    # Create specialized agents
    researcher = ChatAgent(
        name="Researcher",
        instructions="Research the given topic and provide key facts.",
        chat_client=chat_client,
    )

    writer = ChatAgent(
        name="Writer",
        instructions="Write engaging content based on the research provided.",
        chat_client=chat_client,
    )

    reviewer = ChatAgent(
        name="Reviewer",
        instructions="Review the content and provide a final polished version.",
        chat_client=chat_client,
    )

    # Build a sequential workflow
    workflow = (
        SequentialBuilder()
        .add_agents([researcher, writer, reviewer])
        .build()
    )

    # Convert to a workflow agent
    workflow_agent = workflow.as_agent(name="Content Creation Pipeline")

    # Create a thread and run the workflow
    thread = workflow_agent.get_new_thread()
    messages = [ChatMessage(role=Role.USER, content="Write about quantum computing")]

    print("Starting workflow...")
    print("=" * 60)

    current_author = None
    async for update in workflow_agent.run_stream(messages, thread=thread):
        # Show when different agents are responding
        if update.author_name and update.author_name != current_author:
            if current_author:
                print("\n" + "-" * 40)
            print(f"\n[{update.author_name}]:")
            current_author = update.author_name

        if update.text:
            print(update.text, end="", flush=True)

    print("\n" + "=" * 60)
    print("Workflow completed!")


if __name__ == "__main__":
    asyncio.run(main())

Comprensión de la conversión de eventos

Cuando un flujo de trabajo se ejecuta como agente, los eventos de flujo de trabajo se convierten en respuestas del agente. El tipo de respuesta depende del método que use:

• run(): devuelve un objeto AgentRunResponse que contiene el resultado completo una vez finalizado el flujo de trabajo.
• run_stream(): produce AgentRunResponseUpdate objetos a medida que se ejecuta el flujo de trabajo, lo que proporciona actualizaciones en tiempo real.

Durante la ejecución, los eventos de flujo de trabajo internos se asignan a las respuestas del agente de la siguiente manera:

Esta conversión le permite usar la interfaz del agente estándar mientras sigue teniendo acceso a información detallada del flujo de trabajo cuando sea necesario.