Microsoft Agent Framework Workflows Orchestrations - Gruppenchat

Gruppenchat-Orchestrierung modelliert eine gemeinsame Unterhaltung zwischen mehreren Agents, koordiniert von einem Vorgesetzten, der die Auswahl und den Unterhaltungsfluss von Sprechern bestimmt. Dieses Muster eignet sich ideal für Szenarien, die iterative Verfeinerung, kollaborative Problemlösung oder multispektivische Analysen erfordern.

Unterschiede zwischen Gruppenchat und anderen Mustern

Die Gruppenchat-Orchestrierung weist im Vergleich zu anderen Multi-Agent-Mustern unterschiedliche Merkmale auf:

• Zentrale Koordination: Im Gegensatz zu Übergabemustern, bei denen Agents die Kontrolle direkt übertragen, verwendet gruppenchat einen Vorgesetzten, um zu koordinieren, wer als Nächstes spricht
• Iterative Verfeinerung: Agenten können die Antworten der anderen in mehreren Runden überprüfen und darauf aufbauen.
• Flexible Sprecherauswahl: Der Vorgesetzte kann verschiedene Strategien (Roundrobin, promptbasierte, benutzerdefinierte Logik) verwenden, um Lautsprecher auszuwählen.
• Freigegebener Kontext: Alle Agenten sehen den vollständigen Unterhaltungsverlauf, wodurch kollaborative Verfeinerung ermöglicht wird.

Sie lernen Folgendes

• So erstellen Sie spezielle Agents für die Gruppenzusammenarbeit
• Konfigurieren von Strategien für die Sprecherauswahl
• Erstellen von Workflows mit iterativer Agent-Verfeinerung
• Anpassen des Unterhaltungsflusses mit benutzerdefinierten Managern

Einrichten des Azure OpenAI-Clients

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

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

Definieren Sie Ihre Agenten

Erstellen Sie spezielle Agents für verschiedene Rollen im Gruppengespräch.

// Create a copywriter agent
ChatClientAgent writer = new(client,
    "You are a creative copywriter. Generate catchy slogans and marketing copy. Be concise and impactful.",
    "CopyWriter",
    "A creative copywriter agent");

// Create a reviewer agent
ChatClientAgent reviewer = new(client,
    "You are a marketing reviewer. Evaluate slogans for clarity, impact, and brand alignment. " +
    "Provide constructive feedback or approval.",
    "Reviewer",
    "A marketing review agent");

Konfigurieren des Gruppenchats mit Round-Robin Manager

Erstellen Sie den Gruppenchatworkflow mithilfe von AgentWorkflowBuilder:

// Build group chat with round-robin speaker selection
// The manager factory receives the list of agents and returns a configured manager
var workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(agents => 
        new RoundRobinGroupChatManager(agents) 
        { 
            MaximumIterationCount = 5  // Maximum number of turns
        })
    .AddParticipants(writer, reviewer)
    .Build();

Ausführen des Gruppen-Chat-Workflows

Führen Sie den Workflow aus, und beobachten Sie die iterative Unterhaltung:

// Start the group chat
var messages = new List<ChatMessage> { 
    new(ChatRole.User, "Create a slogan for an eco-friendly electric vehicle.") 
};

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

await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is AgentRunUpdateEvent update)
    {
        // Process streaming agent responses
        AgentRunResponse response = update.AsResponse();
        foreach (ChatMessage message in response.Messages)
        {
            Console.WriteLine($"[{update.ExecutorId}]: {message.Text}");
        }
    }
    else if (evt is WorkflowOutputEvent output)
    {
        // Workflow completed
        var conversationHistory = output.As<List<ChatMessage>>();
        Console.WriteLine("\n=== Final Conversation ===");
        foreach (var message in conversationHistory)
        {
            Console.WriteLine($"{message.AuthorName}: {message.Text}");
        }
        break;
    }
}

Beispielinteraktion

[CopyWriter]: "Green Dreams, Zero Emissions" - Drive the future with style and sustainability.

[Reviewer]: The slogan is good, but "Green Dreams" might be a bit abstract. Consider something 
more direct like "Pure Power, Zero Impact" to emphasize both performance and environmental benefit.

[CopyWriter]: "Pure Power, Zero Impact" - Experience electric excellence without compromise.

[Reviewer]: Excellent! This slogan is clear, impactful, and directly communicates the key benefits. 
The tagline reinforces the message perfectly. Approved for use.

[CopyWriter]: Thank you! The final slogan is: "Pure Power, Zero Impact" - Experience electric 
excellence without compromise.

Einrichten des Chatclients

from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

# Initialize the Azure OpenAI chat client
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())

Definieren Sie Ihre Agenten

Erstellen Sie spezialisierte Agents mit unterschiedlichen Rollen:

from agent_framework import ChatAgent

# Create a researcher agent
researcher = ChatAgent(
    name="Researcher",
    description="Collects relevant background information.",
    instructions="Gather concise facts that help answer the question. Be brief and factual.",
    chat_client=chat_client,
)

# Create a writer agent
writer = ChatAgent(
    name="Writer",
    description="Synthesizes polished answers using gathered information.",
    instructions="Compose clear, structured answers using any notes provided. Be comprehensive.",
    chat_client=chat_client,
)

Konfigurieren des Gruppenchats mit einfacher Auswahl

Erstellen eines Gruppenchats mit benutzerdefinierter Sprecherauswahllogik:

from agent_framework import GroupChatBuilder, GroupChatStateSnapshot

def select_next_speaker(state: GroupChatStateSnapshot) -> str | None:
    """Alternate between researcher and writer for collaborative refinement.

    Args:
        state: Contains task, participants, conversation, history, and round_index

    Returns:
        Name of next speaker, or None to finish
    """
    round_idx = state["round_index"]
    history = state["history"]

    # Finish after 4 turns (researcher → writer → researcher → writer)
    if round_idx >= 4:
        return None

    # Alternate speakers
    last_speaker = history[-1].speaker if history else None
    if last_speaker == "Researcher":
        return "Writer"
    return "Researcher"

# Build the group chat workflow
workflow = (
    GroupChatBuilder()
    .set_select_speakers_func(select_next_speaker, display_name="Orchestrator")
    .participants([researcher, writer])
    .build()
)

Konfigurieren des Gruppenchats mit Agent-Based Manager

Alternativ können Sie einen agentbasierten Manager für die intelligente Lautsprecherauswahl verwenden. Der Vorgesetzte ist vollständig ausgestattet ChatAgent mit Zugriff auf Tools, Kontext und Beobachtbarkeit.

# Create coordinator agent for speaker selection
coordinator = ChatAgent(
    name="Coordinator",
    description="Coordinates multi-agent collaboration by selecting speakers",
    instructions="""
You coordinate a team conversation to solve the user's task.

Review the conversation history and select the next participant to speak.

Guidelines:
- Start with Researcher to gather information
- Then have Writer synthesize the final answer
- Only finish after both have contributed meaningfully
- Allow for multiple rounds of information gathering if needed
""",
    chat_client=chat_client,
)

# Build group chat with agent-based manager
workflow = (
    GroupChatBuilder()
    .set_manager(coordinator, display_name="Orchestrator")
    .with_termination_condition(lambda messages: sum(1 for msg in messages if msg.role == Role.ASSISTANT) >= 4)
    .participants([researcher, writer])
    .build()
)

Ausführen des Gruppen-Chat-Workflows

Führen Sie die Workflow- und Prozessereignisse aus:

from typing import cast
from agent_framework import AgentRunUpdateEvent, Role, WorkflowOutputEvent

task = "What are the key benefits of async/await in Python?"

print(f"Task: {task}\n")
print("=" * 80)

final_conversation: list[ChatMessage] = []
last_executor_id: str | None = None

# Run the workflow
async for event in workflow.run_stream(task):
    if isinstance(event, AgentRunUpdateEvent):
        # Print streaming agent updates
        eid = event.executor_id
        if eid != last_executor_id:
            if last_executor_id is not None:
                print()
            print(f"[{eid}]:", end=" ", flush=True)
            last_executor_id = eid
        print(event.data, end="", flush=True)
    elif isinstance(event, WorkflowOutputEvent):
        # Workflow completed - data is a list of ChatMessage
        final_conversation = cast(list[ChatMessage], event.data)

if final_conversation:
    print("\n\n" + "=" * 80)
    print("Final Conversation:")
    for msg in final_conversation:
        author = getattr(msg, "author_name", "Unknown")
        text = getattr(msg, "text", str(msg))
        print(f"\n[{author}]\n{text}")
        print("-" * 80)

print("\nWorkflow completed.")

Beispielinteraktion

Task: What are the key benefits of async/await in Python?

================================================================================

[Researcher]: Async/await in Python provides non-blocking I/O operations, enabling 
concurrent execution without threading overhead. Key benefits include improved 
performance for I/O-bound tasks, better resource utilization, and simplified 
concurrent code structure using native coroutines.

[Writer]: The key benefits of async/await in Python are:

1. **Non-blocking Operations**: Allows I/O operations to run concurrently without 
   blocking the main thread, significantly improving performance for network 
   requests, file I/O, and database queries.

2. **Resource Efficiency**: Avoids the overhead of thread creation and context 
   switching, making it more memory-efficient than traditional threading.

3. **Simplified Concurrency**: Provides a clean, synchronous-looking syntax for 
   asynchronous code, making concurrent programs easier to write and maintain.

4. **Scalability**: Enables handling thousands of concurrent connections with 
   minimal resource consumption, ideal for high-performance web servers and APIs.

--------------------------------------------------------------------------------

Workflow completed.

Wichtige Konzepte

• Zentraler Manager: Gruppenchat verwendet einen Vorgesetzten, um die Sprecherauswahl und den Ablauf zu koordinieren.
• AgentWorkflowBuilder.CreateGroupChatBuilderWith(): Erstellt Workflows mit einer Manager-Factoryfunktion
• RoundRobinGroupChatManager: Integrierter Manager, der Sprecher im Round-Robin-Modus abwechselt
• MaximumIterationCount: Steuert die maximale Anzahl von Agentendurchläufen vor der Beendigung
• Benutzerdefinierte Manager: Erweitern RoundRobinGroupChatManager oder benutzerdefinierte Logik implementieren
• Iterative Verfeinerung: Agenten überprüfen und verbessern die Beiträge der anderen
• Freigegebener Kontext: Alle Teilnehmer sehen den vollständigen Unterhaltungsverlauf

• Flexible Manager-Strategien: Wählen Zwischen einfachen Selektoren, agentbasierten Managern oder benutzerdefinierter Logik
• GroupChatBuilder: Erstellt Workflows mit konfigurierbarer Sprecherauswahl
• set_select_speakers_func(): Definieren benutzerdefinierter Python-Funktionen für die Sprecherauswahl
• set_manager(): Verwenden eines agentenbasierten Managers für intelligente Sprecherkoordination
• GroupChatStateSnapshot: Stellt den Gesprächszustand für Auswahlentscheidungen bereit.
• Iterative Zusammenarbeit: Agents bauen auf den Beiträgen der anderen auf
• Ereignisstreaming: Verarbeiten AgentRunUpdateEvent und WorkflowOutputEvent in Echtzeit
• list[ChatMessage] Ausgabe: Alle Orchestrierungen geben eine Liste von Chatnachrichten zurück.

Erweitert: Benutzerdefinierte Lautsprecherauswahl

Sie können benutzerdefinierte Managerlogik implementieren, indem Sie einen benutzerdefinierten Gruppenchat-Manager erstellen:

public class ApprovalBasedManager : RoundRobinGroupChatManager
{
    private readonly string _approverName;

    public ApprovalBasedManager(IReadOnlyList<AIAgent> agents, string approverName) 
        : base(agents)
    {
        _approverName = approverName;
    }

    // Override to add custom termination logic
    protected override ValueTask<bool> ShouldTerminateAsync(
        IReadOnlyList<ChatMessage> history, 
        CancellationToken cancellationToken = default)
    {
        var last = history.LastOrDefault();
        bool shouldTerminate = last?.AuthorName == _approverName &&
            last.Text?.Contains("approve", StringComparison.OrdinalIgnoreCase) == true;

        return ValueTask.FromResult(shouldTerminate);
    }
}

// Use custom manager in workflow
var workflow = AgentWorkflowBuilder
    .CreateGroupChatBuilderWith(agents => 
        new ApprovalBasedManager(agents, "Reviewer") 
        { 
            MaximumIterationCount = 10 
        })
    .AddParticipants(writer, reviewer)
    .Build();

Sie können basierend auf dem Gesprächszustand komplexe Auswahllogik implementieren.

def smart_selector(state: GroupChatStateSnapshot) -> str | None:
    """Select speakers based on conversation content and context."""
    round_idx = state["round_index"]
    conversation = state["conversation"]
    history = state["history"]

    # Maximum 10 rounds
    if round_idx >= 10:
        return None

    # First round: always start with researcher
    if round_idx == 0:
        return "Researcher"

    # Check last message content
    last_message = conversation[-1] if conversation else None
    last_text = getattr(last_message, "text", "").lower()

    # If researcher asked a question, let writer respond
    if "?" in last_text and history[-1].speaker == "Researcher":
        return "Writer"

    # If writer provided info, let researcher validate or extend
    if history[-1].speaker == "Writer":
        return "Researcher"

    # Default alternation
    return "Writer" if history[-1].speaker == "Researcher" else "Researcher"

workflow = (
    GroupChatBuilder()
    .set_select_speakers_func(smart_selector, display_name="SmartOrchestrator")
    .participants([researcher, writer])
    .build()
)

Wann sollte man Gruppenchats verwenden?

Gruppenchat-Orchestrierung ist ideal für:

• Iterative Verfeinerung: Mehrere Runden der Überprüfung und Verbesserung
• Kollaborative Problemlösung: Agenten mit komplementärer Expertise zusammenarbeiten
• Inhaltserstellung: Writer-Reviewer-Workflows für die Dokumenterstellung
• Multi-Perspective Analysis: Vielfältige Perspektiven zu derselben Eingabe
• Qualitätssicherung: Automatisierte Überprüfungs- und Genehmigungsprozesse

Erwägen Sie Alternativen, wenn:

• Sie benötigen eine strikte sequenzielle Verarbeitung (sequenzielle Orchestrierung verwenden)
• Agents sollten vollständig unabhängig arbeiten (Concurrent Orchestration verwenden)
• Direkte Agent-zu-Agent-Übergaben sind erforderlich (Handoff-Orchestrierung verwenden)
• Komplexe dynamische Planung ist erforderlich (Magentische Orchestrierung verwenden)

Nächste Schritte

Handoff-Orchestrierung