Microsoft Agent Framework Workflow-Orchestrierungen - Sequenzielle

In sequenzieller Orchestrierung werden Agents in einer Pipeline organisiert, in der jeder Agent die Aufgabe wiederum verarbeitet und die Ausgabe an den nächsten Agent in der Sequenz übergibt.

Einrichten des Azure OpenAI-Clients

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// 1) Set up the Azure OpenAI client
var endpoint = Environment.GetEnvironmentVariable("AZURE_OPENAI_ENDPOINT") ??
    throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
var deploymentName = Environment.GetEnvironmentVariable("AZURE_OPENAI_DEPLOYMENT_NAME") ?? "gpt-4o-mini";
var client = new AzureOpenAIClient(new Uri(endpoint), new AzureCliCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

Erstellen Sie spezielle Agents, die nacheinander arbeiten:

// 2) Helper method to create translation agents
static ChatClientAgent GetTranslationAgent(string targetLanguage, IChatClient chatClient) =>
    new(chatClient,
        $"You are a translation assistant who only responds in {targetLanguage}. Respond to any " +
        $"input by outputting the name of the input language and then translating the input to {targetLanguage}.");

// Create translation agents for sequential processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Einrichten der sequenziellen Orchestrierung

Erstellen Sie den Workflow mithilfe von AgentWorkflowBuilder:

// 3) Build sequential workflow
var workflow = AgentWorkflowBuilder.BuildSequential(translationAgents);

Ausführen des sequenziellen Workflows

Führen Sie den Workflow aus, und verarbeiten Sie die Ereignisse:

// 4) Run the workflow
var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, world!") };

StreamingRun run = await InProcessExecution.StreamAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

List<ChatMessage> result = new();
await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is AgentRunUpdateEvent e)
    {
        Console.WriteLine($"{e.ExecutorId}: {e.Data}");
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = (List<ChatMessage>)outputEvt.Data!;
        break;
    }
}

// Display final result
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Content}");
}

Beispielausgabe

French_Translation: User: Hello, world!
French_Translation: Assistant: English detected. Bonjour, le monde !
Spanish_Translation: Assistant: French detected. ¡Hola, mundo!
English_Translation: Assistant: Spanish detected. Hello, world!

Wichtige Konzepte

• Sequenzielle Verarbeitung: Jeder Agent verarbeitet die Ausgabe des vorherigen Agents in der Reihenfolge
• AgentWorkflowBuilder.BuildSequential(): Erstellt einen Pipelineworkflow aus einer Sammlung von Agents.
• ChatClientAgent: Stellt einen Agent dar, der von einem Chatclient mit bestimmten Anweisungen unterstützt wird.
• StreamingRun: Bietet Echtzeitausführung mit Ereignisstreamingfunktionen
• Ereignisbehandlung: Überwachungen des Agentenfortschritts durch AgentRunUpdateEvent und Erfassung des Abschlusses durch WorkflowOutputEvent

Bei der sequenziellen Orchestrierung verarbeitet jeder Agent die Aufgabe nacheinander, wobei die Ausgabe von einem zum nächsten fließt. Beginnen wir mit der Definition von Agents für einen zweistufigen Prozess:

from agent_framework.azure import AzureChatClient
from azure.identity import AzureCliCredential

# 1) Create agents using AzureChatClient
chat_client = AzureChatClient(credential=AzureCliCredential())

writer = chat_client.create_agent(
    instructions=(
        "You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."
    ),
    name="writer",
)

reviewer = chat_client.create_agent(
    instructions=(
        "You are a thoughtful reviewer. Give brief feedback on the previous assistant message."
    ),
    name="reviewer",
)

Einrichten der sequenziellen Orchestrierung

Die SequentialBuilder Klasse erstellt eine Pipeline, in der Agents Aufgaben in der Reihenfolge verarbeiten. Jeder Agent sieht den vollständigen Unterhaltungsverlauf und fügt seine Antwort hinzu:

from agent_framework import SequentialBuilder

# 2) Build sequential workflow: writer -> reviewer
workflow = SequentialBuilder().participants([writer, reviewer]).build()

Ausführen des sequenziellen Workflows

Führen Sie den Workflow aus, und erfassen Sie die endgültige Unterhaltung mit den Beiträgen der einzelnen Agenten:

from agent_framework import ChatMessage, WorkflowOutputEvent

# 3) Run and print final conversation
output_evt: WorkflowOutputEvent | None = None
async for event in workflow.run_stream("Write a tagline for a budget-friendly eBike."):
    if isinstance(event, WorkflowOutputEvent):
        output_evt = event

if output_evt:
    print("===== Final Conversation =====")
    messages: list[ChatMessage] | Any = output_evt.data
    for i, msg in enumerate(messages, start=1):
        name = msg.author_name or ("assistant" if msg.role == Role.ASSISTANT else "user")
        print(f"{'-' * 60}\n{i:02d} [{name}]\n{msg.text}")

Beispielausgabe

===== Final Conversation =====
------------------------------------------------------------
01 [user]
Write a tagline for a budget-friendly eBike.
------------------------------------------------------------
02 [writer]
Ride farther, spend less—your affordable eBike adventure starts here.
------------------------------------------------------------
03 [reviewer]
This tagline clearly communicates affordability and the benefit of extended travel, making it
appealing to budget-conscious consumers. It has a friendly and motivating tone, though it could
be slightly shorter for more punch. Overall, a strong and effective suggestion!

Erweitert: Mischen von Agents mit benutzerdefinierten Executoren

Sequenzielle Orchestrierung unterstützt gemischte Agenten mit benutzerdefinierten Ausführungs-Engines für die spezialisierte Verarbeitung. Dies ist nützlich, wenn Sie benutzerdefinierte Logik benötigen, für die keine LLM erforderlich ist:

Definieren eines benutzerdefinierten Executors

from agent_framework import Executor, WorkflowContext, handler
from agent_framework import ChatMessage, Role

class Summarizer(Executor):
    """Simple summarizer: consumes full conversation and appends an assistant summary."""

    @handler
    async def summarize(
        self,
        conversation: list[ChatMessage],
        ctx: WorkflowContext[list[ChatMessage]]
    ) -> None:
        users = sum(1 for m in conversation if m.role == Role.USER)
        assistants = sum(1 for m in conversation if m.role == Role.ASSISTANT)
        summary = ChatMessage(
            role=Role.ASSISTANT,
            text=f"Summary -> users:{users} assistants:{assistants}"
        )
        await ctx.send_message(list(conversation) + [summary])

Erstellen eines gemischten sequenziellen Workflows

# Create a content agent
content = chat_client.create_agent(
    instructions="Produce a concise paragraph answering the user's request.",
    name="content",
)

# Build sequential workflow: content -> summarizer
summarizer = Summarizer(id="summarizer")
workflow = SequentialBuilder().participants([content, summarizer]).build()

Beispielausgabe mit benutzerdefiniertem Executor

------------------------------------------------------------
01 [user]
Explain the benefits of budget eBikes for commuters.
------------------------------------------------------------
02 [content]
Budget eBikes offer commuters an affordable, eco-friendly alternative to cars and public transport.
Their electric assistance reduces physical strain and allows riders to cover longer distances quickly,
minimizing travel time and fatigue. Budget models are low-cost to maintain and operate, making them accessible
for a wider range of people. Additionally, eBikes help reduce traffic congestion and carbon emissions,
supporting greener urban environments. Overall, budget eBikes provide cost-effective, efficient, and
sustainable transportation for daily commuting needs.
------------------------------------------------------------
03 [assistant]
Summary -> users:1 assistants:1

Wichtige Konzepte

• Freigegebener Kontext: Jeder Teilnehmer empfängt den vollständigen Unterhaltungsverlauf, einschließlich aller vorherigen Nachrichten
• Order Matters: Agenten führen streng in der participants() Liste angegebenen Reihenfolge aus
• Flexible Teilnehmer: Sie können Agents und benutzerdefinierte Executoren in beliebiger Reihenfolge kombinieren.
• Unterhaltungsfluss: Jeder Agent/Ausführender fügt der Unterhaltung etwas hinzu und baut einen vollständigen Dialog auf.