Microsoft Agent Framework-Workflows-Orchestrierungen – gleichzeitig

Bei gleichzeitiger Orchestrierung arbeiten mehrere Agents gleichzeitig und unabhängig an derselben Aufgabe und bieten unterschiedliche Perspektiven für die gleiche Eingabe.

Einrichten des Azure OpenAI-Clients

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.Projects;
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 AIProjectClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetProjectOpenAIClient()
    .GetProjectResponsesClient()
    .AsIChatClient(deploymentName);

Definieren Sie Ihre Agenten

Erstellen Sie mehrere spezialisierte Agents, die gleichzeitig an derselben Aufgabe 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 concurrent processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Einrichten der gleichzeitigen Orchestrierung

Erstellen Sie den Workflow mithilfe von AgentWorkflowBuilder, um Agents parallel auszuführen.

// 3) Build concurrent workflow
var workflow = AgentWorkflowBuilder.BuildConcurrent(translationAgents);

Ausführen des gleichzeitigen Workflows und Sammeln von Ergebnissen

Führen Sie den Workflow aus und verarbeiten Sie Ereignisse von allen Agenten, die gleichzeitig laufen:

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

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

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

// Display aggregated results from all agents
Console.WriteLine("===== Final Aggregated Results =====");
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

Beispielausgabe

French_Agent: English detected. Bonjour, le monde !
Spanish_Agent: English detected. ¡Hola, mundo!
English_Agent: English detected. Hello, world!

===== Final Aggregated Results =====
User: Hello, world!
Assistant: English detected. Bonjour, le monde !
Assistant: English detected. ¡Hola, mundo!
Assistant: English detected. Hello, world!

Wichtige Konzepte

• Parallele Ausführung: Alle Agents verarbeiten die Eingabe gleichzeitig und unabhängig
• AgentWorkflowBuilder.BuildConcurrent(): Erstellt einen gleichzeitigen Workflow aus einer Sammlung von Agents.
• Automatische Aggregation: Ergebnisse aller Agents werden automatisch im Endergebnis erfasst.
• Ereignisstreaming: Echtzeitüberwachung des Agentfortschritts durch AgentResponseUpdateEvent
• Vielfältige Perspektiven: Jeder Agent bringt sein einzigartiges Know-how zum gleichen Problem

Agents sind spezialisierte Entitäten, die Aufgaben verarbeiten können. Der folgende Code definiert drei Agenten: ein Forschungsexperte, ein Marketingexperte und ein Rechtsexperte.

import os

from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

# 1) Create three domain agents using FoundryChatClient
chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher = chat_client.as_agent(
    instructions=(
        "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
        " opportunities, and risks."
    ),
    name="researcher",
)

marketer = chat_client.as_agent(
    instructions=(
        "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
        " aligned to the prompt."
    ),
    name="marketer",
)

legal = chat_client.as_agent(
    instructions=(
        "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns"
        " based on the prompt."
    ),
    name="legal",
)

Einrichten der gleichzeitigen Orchestrierung

Mit der ConcurrentBuilder Klasse können Sie einen Workflow erstellen, um mehrere Agents parallel auszuführen. Sie übergeben die Liste der Agenten als Teilnehmer.

from agent_framework.orchestrations import ConcurrentBuilder

# 2) Build a concurrent workflow
# Participants are either Agents (type of SupportsAgentRun) or Executors
workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

Ausführen des gleichzeitigen Workflows und Sammeln der Ergebnisse

Der Standardaggregator erzeugt eine einzelne AgentResponse mit einer Assistentennachricht pro Teilnehmer:

from agent_framework import AgentResponse

# 3) Run with a single prompt and print the aggregated agent responses
events = await workflow.run("We are launching a new budget-friendly electric bike for urban commuters.")
outputs = events.get_outputs()

if outputs:
    print("===== Final Aggregated Results =====")
    final: AgentResponse = outputs[0]
    for msg in final.messages:
        name = msg.author_name or "assistant"
        print(f"{'-' * 60}\n\n[{name}]:\n{msg.text}")

Beispielausgabe

===== Final Aggregated Results =====
------------------------------------------------------------

[researcher]:
**Insights:**

- **Target Demographic:** Urban commuters seeking affordable, eco-friendly transport;
    likely to include students, young professionals, and price-sensitive urban residents.
- **Market Trends:** E-bike sales are growing globally, with increasing urbanization,
    higher fuel costs, and sustainability concerns driving adoption.
...
------------------------------------------------------------

[marketer]:
**Value Proposition:**
"Empowering your city commute: Our new electric bike combines affordability, reliability, and
    sustainable design—helping you conquer urban journeys without breaking the bank."
...
------------------------------------------------------------

[legal]:
**Constraints, Disclaimers, & Policy Concerns for Launching a Budget-Friendly Electric Bike for Urban Commuters:**

**1. Regulatory Compliance**
- Verify that the electric bike meets all applicable federal, state, and local regulations
    regarding e-bike classification, speed limits, power output, and safety features.

Fortgeschritten: Benutzerdefinierte Agent-Executoren

Parallele Orchestrierung unterstützt benutzerdefinierte Executoren, die Agents mit zusätzlicher Logik versehen. Dies ist nützlich, wenn Sie mehr Kontrolle darüber benötigen, wie Agents initialisiert werden und wie sie Anforderungen verarbeiten:

Definieren von benutzerdefinierten Agent-Executoren

from agent_framework import (
    AgentExecutorRequest,
    AgentExecutorResponse,
    Agent,
    Executor,
    WorkflowContext,
    handler,
)

class ResearcherExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "researcher"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
                " opportunities, and risks."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

class MarketerExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "marketer"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
                " aligned to the prompt."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

Erstellen eines Workflows mit benutzerdefinierten Executors

chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher = ResearcherExec(chat_client)
marketer = MarketerExec(chat_client)
legal = LegalExec(chat_client)

workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

Erweitert: Benutzerdefinierter Aggregator

Standardmäßig aggregiert die gleichzeitige Orchestrierung alle Agentenantworten in einer einzigen AgentResponse, mit einer Assistentennachricht pro Teilnehmer. Sie können dieses Verhalten mit einem benutzerdefinierten Aggregator überschreiben, der die Ergebnisse auf eine bestimmte Weise verarbeitet:

Definieren eines benutzerdefinierten Aggregators

from agent_framework import AgentExecutorResponse

# Create a summarizer agent for the aggregator
summarizer_agent = chat_client.as_agent(
    instructions=(
        "You are a helpful assistant that consolidates multiple domain expert outputs "
        "into one cohesive, concise summary with clear takeaways. Keep it under 200 words."
    ),
    name="summarizer",
)

# Define a custom aggregator callback
async def summarize_results(results: list[AgentExecutorResponse]) -> str:
    # Extract one final assistant message per agent
    expert_sections: list[str] = []
    for r in results:
        try:
            messages = getattr(r.agent_response, "messages", [])
            final_text = messages[-1].text if messages and hasattr(messages[-1], "text") else "(no content)"
            expert_sections.append(f"{r.executor_id}:\n{final_text}")
        except Exception as e:
            expert_sections.append(f"{r.executor_id}: (error: {type(e).__name__}: {e})")

    # Ask the model to synthesize a concise summary of the experts' outputs
    prompt = "\n\n".join(expert_sections)
    response = await summarizer_agent.run(prompt)
    # Return the model's final assistant text as the completion result
    return response.messages[-1].text if response.messages else ""

Erstellen eines Workflows mit benutzerdefiniertem Aggregator

workflow = (
    ConcurrentBuilder(participants=[researcher, marketer, legal])
    .with_aggregator(summarize_results)
    .build()
)

output = None
async for event in workflow.run("We are launching a new budget-friendly electric bike for urban commuters.", stream=True):
    if event.type == "output":
        output = event.data

if output:
    print("===== Final Consolidated Output =====")
    print(output)

Beispielausgabe mit benutzerdefiniertem Aggregator

===== Final Consolidated Output =====
Urban e-bike demand is rising rapidly due to eco-awareness, urban congestion, and high fuel costs,
with market growth projected at a ~10% CAGR through 2030. Key customer concerns are affordability,
easy maintenance, convenient charging, compact design, and theft protection. Differentiation opportunities
include integrating smart features (GPS, app connectivity), offering subscription or leasing options, and
developing portable, space-saving designs. Partnering with local governments and bike shops can boost visibility.

Risks include price wars eroding margins, regulatory hurdles, battery quality concerns, and heightened expectations
for after-sales support. Accurate, substantiated product claims and transparent marketing (with range disclaimers)
are essential. All e-bikes must comply with local and federal regulations on speed, wattage, safety certification,
and labeling. Clear warranty, safety instructions (especially regarding batteries), and inclusive, accessible
marketing are required. For connected features, data privacy policies and user consents are mandatory.

Effective messaging should target young professionals, students, eco-conscious commuters, and first-time buyers,
emphasizing affordability, convenience, and sustainability. Slogan suggestion: "Charge Ahead—City Commutes Made
Affordable." Legal review in each target market, compliance vetting, and robust customer support policies are
critical before launch.

Zwischenausgaben

Standardmäßig erscheint nur die Ausgabe des Aggregators als Workflow-"output"-Ereignis (Terminal). Übergeben Sie intermediate_output_from mit den Teilnehmern, die Sie als Zwischenquellen bestimmen möchten, um ihre individuellen Ausgaben ebenfalls als "intermediate"-Ereignisse auszugeben:

workflow = ConcurrentBuilder(
    participants=[researcher, marketer, legal],
    intermediate_output_from=[researcher, marketer, legal],
).build()

Sie können diese Ereignisse im Streamingmodus in Echtzeit behandeln:

from agent_framework import AgentResponseUpdate

# Track the last author to format streaming output.
last_author: str | None = None

async for event in workflow.run("Analyze our new product launch strategy.", stream=True):
    if event.type == "intermediate" and isinstance(event.data, AgentResponseUpdate):
        update = event.data
        author = update.author_name
        if author != last_author:
            if last_author is not None:
                print()  # Newline between different authors
            print(f"{author}: {update.text}", end="", flush=True)
            last_author = author
        else:
            print(update.text, end="", flush=True)

Wichtige Konzepte

• Parallele Ausführung: Alle Agents arbeiten gleichzeitig und unabhängig an der Aufgabe.
• AgentResponse Output: Der Standardaggregator liefert eine einzelne AgentResponse mit einer Assistentennachricht pro Teilnehmer (keine Benutzeraufforderung enthalten)
• Vielfältige Perspektiven: Jeder Agent bringt sein einzigartiges Know-how zum gleichen Problem
• Flexible Teilnehmer: Sie können Agenten direkt verwenden oder in benutzerdefinierte Executoren umschließen
• Benutzerdefinierte Verarbeitung: Überschreiben des Standardaggregators zum Synthetisieren von Ergebnissen auf domänenspezifische Weise
• Zwischenergebnisse: Übergeben Sie intermediate_output_from=[participant, ...], um die Ausgabe jedes aufgeführten Teilnehmers zusätzlich zu dem abschließenden "intermediate"-Ereignis des Aggregators als "output"-Ereignisse auszugeben

Go unterstützt gleichzeitige Agent-Workflows mit agentworkflow.NewConcurrentWorkflowBuilder. Sie können dasselbe Muster auch manuell mit Fan-out- und Fan-in-Kanten erstellen, wenn Sie ein benutzerdefiniertes Verhalten des Executors benötigen.

Einrichten der Foundry-Konfiguration

Konfigurieren Sie den Endpunkt des Foundry-Projekts, die Bereitstellung des Modells und die Authentifizierung:

endpoint := os.Getenv("FOUNDRY_PROJECT_ENDPOINT")
model := cmp.Or(os.Getenv("FOUNDRY_MODEL"), "gpt-4o-mini")

token, err := azidentity.NewDefaultAzureCredential(nil)
if err != nil {
    return err
}

Definieren Sie Ihre Agenten

Erstellen Sie mehrere spezialisierte Agents, die gleichzeitig an derselben Aufgabe arbeiten:

newTranslationAgent := func(language string) *agent.Agent {
    return foundryprovider.NewAgent(
        endpoint,
        token,
        foundryprovider.ModelDeployment(model),
        foundryprovider.AgentConfig{
            Instructions: fmt.Sprintf(
                "You are a translation assistant who only responds in %s. Respond to any input by outputting the name of the input language and then translating the input to %s.",
                language,
                language,
            ),
            Config: agent.Config{Name: language},
        },
    )
}

agents := []*agent.Agent{
    newTranslationAgent("French"),
    newTranslationAgent("Spanish"),
    newTranslationAgent("English"),
}

Einrichten der gleichzeitigen Orchestrierung

Erstellen Sie den Workflow mit agentworkflow.NewConcurrentWorkflowBuilder:

wf, err := agentworkflow.NewConcurrentWorkflowBuilder(agents...).
    WithName("translation-concurrent").
    Build()
if err != nil {
    return err
}

Ausführen des gleichzeitigen Workflows und Sammeln von Ergebnissen

Führen Sie den Workflow mit einer Benutzernachricht und einem Turntoken aus. Wenn die Ereignisemissionen aktiviert sind, werden Agentupdates als Workflowausgabeereignisse vor der endgültigen aggregierten Ausgabe angezeigt.

run, err := inproc.Default.RunStreaming(ctx, wf, []*message.Message{message.NewText("Hello, world!")})
if err != nil {
    return err
}
defer run.Close(ctx)

emitEvents := true
if err := run.SendMessage(ctx, workflow.TurnToken{EmitEvents: &emitEvents}); err != nil {
    return err
}

for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }
    if output, ok := evt.(workflow.OutputEvent); ok {
        switch value := output.Output.(type) {
        case *agent.ResponseUpdate:
            fmt.Printf("%s: %s\n", output.ExecutorID, value.String())
        case []*message.Message:
            fmt.Println("===== Final Aggregated Results =====")
            for _, msg := range value {
                fmt.Printf("%s: %s\n", msg.Role, msg.String())
            }
        }
    }
}

Beispielausgabe

French: English detected. Bonjour, le monde !
Spanish: English detected. ¡Hola, mundo!
English: English detected. Hello, world!

===== Final Aggregated Results =====
assistant: English detected. Bonjour, le monde !
assistant: English detected. ¡Hola, mundo!
assistant: English detected. Hello, world!

Fortgeschritten: Benutzerdefinierte Agent-Executoren

Erstellen Sie gleichzeitige Workflows manuell, wenn Sie ein benutzerdefiniertes Executorverhalten benötigen. Ein benutzerdefinierter Executor kann einen Agent anrufen und dann an einem Fanout-/Fan-In-Workflow teilnehmen.

agentExecutor := func(id string, ag *agent.Agent) workflow.ExecutorBinding {
    return workflow.BindNewExecutorFunc(id, func(_ string, executorID string) (*workflow.Executor, error) {
        return workflow.NewExecutor(executorID, func(ctx *workflow.Context, prompt string) (string, error) {
            response, err := ag.RunText(ctx, prompt).Collect()
            if err != nil {
                return "", err
            }
            return response.String(), nil
        }), nil
    })
}

researcher := agentExecutor("researcher", researcherAgent)
marketer := agentExecutor("marketer", marketerAgent)
aggregate := aggregateStrings("ConcurrentAggregationExecutor")

wf, err := workflow.NewBuilder(start).
    AddFanOutEdge(start, []workflow.ExecutorBinding{researcher, marketer}).
    AddFanInBarrierEdge([]workflow.ExecutorBinding{researcher, marketer}, aggregate).
    WithOutputFrom(aggregate).
    Build()

Erweitert: Benutzerdefinierter Aggregator

Verwenden Sie WithAggregator, um das Standardverhalten der Nachrichtenaggregation zu ersetzen:

wf, err := agentworkflow.NewConcurrentWorkflowBuilder(agents...).
    WithName("translation-concurrent").
    WithAggregator(func(_ context.Context, batches [][]*message.Message) []*message.Message {
        results := make([]*message.Message, 0, len(batches))
        for _, batch := range batches {
            if len(batch) > 0 {
                results = append(results, batch[len(batch)-1])
            }
        }
        return results
    }).
    Build()
if err != nil {
    return err
}

Zwischenausgaben

Standardmäßig gibt NewConcurrentWorkflowBuilder Teilnehmer- und Batching-Ausgaben als Zwischenausgaben des Workflows aus und das aggregierte Ergebnis als terminale Ausgabe. Für benutzerdefinierte Executor-Workflows markieren Sie Branch-Executors als zwischengeschaltet und den Aggregator als Endknoten:

wf, err := workflow.NewBuilder(start).
    AddFanOutEdge(start, []workflow.ExecutorBinding{physics, chemistry}).
    AddFanInBarrierEdge([]workflow.ExecutorBinding{physics, chemistry}, aggregate).
    WithIntermediateOutputFrom(physics, chemistry).
    WithOutputFrom(aggregate).
    Build()

Jeder workflow.OutputEvent enthält das ExecutorID, das die Ausgabe erzeugt hat. Verwenden Sie OutputEvent.IsIntermediate(), um die Ausgaben von Zwischenzweigen vom finalen Aggregat zu unterscheiden.

Wichtige Konzepte

• Parallele Ausführung: Alle Agents oder Executoren verarbeiten die Eingabe unabhängig.
• agentworkflow. NewConcurrentWorkflowBuilder(): Erstellt einen gleichzeitigen Workflow aus einer Sammlung von Agents.
• Fan-out-/Fan-in-Kanten: Benutzerdefinierte parallele Workflows verwenden AddFanOutEdge und AddFanInBarrierEdge.
• Nachrichtenaggregation: Der Standardaggregator gibt die letzte Nachricht von jedem Teilnehmer zurück; Benutzerdefinierte Aggregatoren können dieses Verhalten ersetzen.
• Ereignisstreaming: Ausgabeereignisse können einzelne Agentupdates und endgültige aggregierte Ergebnisse anzeigen.
• Zwischenausgaben: WithIntermediateOutputFrom Markiert ausgewählte Ausgaben mit workflow.OutputTagIntermediate.