Orkestreringar för Microsoft Agent Framework-arbetsflöden – sekventiella

I sekventiell orkestrering organiseras agenter till en pipeline. Varje agent bearbetar uppgiften i sin tur och skickar utdata till nästa agent i sekvensen. Detta är idealiskt för arbetsflöden där varje steg bygger på det föregående, till exempel dokumentgranskning, databearbetningspipelines eller resonemang i flera steg.

Viktigt!

Den fullständiga konversationshistoriken från tidigare agenter skickas till nästa agent i sekvensen. Varje agent kan se alla tidigare meddelanden, vilket möjliggör kontextmedveten bearbetning.

Vad du ska lära dig

• Så här skapar du en sekventiell pipeline med agenter
• Så här kedjar du agenter där var och en bygger på tidigare utdata
• Så här blandar du agenter med anpassade utförare för specialiserade uppgifter
• Så här spårar du konversationsflödet via pipelinen

Definiera dina agenter

I sekventiell orkestrering organiseras agenter i en pipeline där varje agent bearbetar uppgiften i sin tur och skickar utdata till nästa agent i sekvensen.

Konfigurera Azure OpenAI-klienten

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 DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

Varning

DefaultAzureCredential är praktiskt för utveckling men kräver noggrant övervägande i produktion. I produktion bör du överväga att använda en specifik autentiseringsuppgift (t.ex. ManagedIdentityCredential) för att undvika problem med svarstid, oavsiktlig avsökning av autentiseringsuppgifter och potentiella säkerhetsrisker från reservmekanismer.

Skapa specialiserade agenter som fungerar i ordning:

// 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));

Konfigurera Sekventiell Orkestrering

Skapa arbetsflödet med :AgentWorkflowBuilder

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

Kör sekventiellt arbetsflöde

Kör arbetsflödet och bearbeta händelserna:

// 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 AgentResponseUpdateEvent 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}");
}

Exempelutdata

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!

Viktiga begrepp

• Sekventiell bearbetning: Varje agent bearbetar utdata från den tidigare agenten i ordning
• AgentWorkflowBuilder.BuildSequential(): Skapar ett pipelinearbetsflöde från en samling agenter
• ChatClientAgent: Representerar en agent som backas upp av en chattklient med specifika instruktioner
• StreamingRun: Erbjuder realtidskörning och funktioner för händelseströmning
• Händelsehantering: Övervaka agentens förlopp genom AgentResponseUpdateEvent och slutförande via WorkflowOutputEvent

I sekventiell orkestrering bearbetar varje agent uppgiften i sin tur, med utdata som flödar från en till en annan. Börja med att definiera agenter för en process i två steg:

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

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

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

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

Konfigurera Sekventiell Orkestrering

Klassen SequentialBuilder skapar en pipeline där agenter bearbetar uppgifter i ordning. Varje agent ser hela konversationshistoriken och lägger till sitt svar:

from agent_framework.orchestrations import SequentialBuilder

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

Kör sekventiellt arbetsflöde

Kör arbetsflödet och samla in den slutliga konversationen som visar varje agents bidrag.

from agent_framework import Message, WorkflowEvent

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

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

Exempelutdata

===== 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!

Avancerat: Blanda agenter med anpassade exekutorer

Sekventiell orkestrering stöder att kombinera agenter med skräddarsydda utförare för specialiserad bearbetning. Detta är användbart när du behöver anpassad logik som inte kräver en LLM:

Definiera en anpassad exekverare

from agent_framework import Executor, WorkflowContext, handler
from agent_framework import Message

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

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

Skapa ett blandat sekventiellt arbetsflöde

# Create a content agent
content = chat_client.as_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()

Exempelutdata med anpassad exekverare

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

Viktiga begrepp

• Delad kontext: Varje deltagare får hela konversationshistoriken, inklusive alla tidigare meddelanden
• Orderärenden: Agenter körs strikt i den ordning som anges i participants() listan
• Flexibla deltagare: Du kan blanda agenter och anpassade utförare i valfri ordning
• Konversationsflöde: Varje agent/exekutor lägger till i konversationen och skapar en fullständig dialog

Nästa steg

Samtidig orkestrering