Orchestrations de flux de travail Microsoft Agent Framework - Conversation de groupe

L’orchestration de conversation de groupe modélise une conversation collaborative entre plusieurs agents, coordonnée par un responsable qui détermine la sélection de l’orateur et le flux de conversation. Ce modèle est idéal pour les scénarios nécessitant un affinement itératif, une résolution de problèmes collaborative ou une analyse multi-perspective.

Différences entre la conversation de groupe et d’autres modèles

L’orchestration de conversation de groupe présente des caractéristiques distinctes par rapport à d’autres modèles multi-agents :

• Coordination centralisée : contrairement aux modèles de transfert où les agents transfèrent directement le contrôle, la conversation de groupe utilise un responsable pour coordonner qui parle ensuite
• Affinement itératif : les agents peuvent examiner et s’appuyer sur les réponses des uns des autres dans plusieurs rondes
• Sélection flexible d'intervenants : le responsable peut utiliser différentes stratégies (tour de rôle, basé sur les invites, logique personnalisée) pour sélectionner des intervenants
• Contexte partagé : tous les agents voient l’historique complet des conversations, ce qui permet l’affinement collaboratif

Ce que vous allez apprendre

• Comment créer des agents spécialisés pour la collaboration de groupe
• Comment configurer des stratégies de sélection d’orateur
• Création de flux de travail avec affinement de l’agent itératif
• Comment personnaliser le flux de conversation avec des gestionnaires personnalisés

Configurer le client Azure OpenAI

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

Définir vos agents

Créez des agents spécialisés pour différents rôles dans la conversation de groupe :

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

Configurer la conversation de groupe avec Round-Robin Manager

Créez le flux de travail de conversation de groupe à l’aide de 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();

Exécuter le flux de travail de conversation de groupe

Exécutez le flux de travail et observez la conversation itérative :

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

Exemple d’interaction

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

Configurer le client de conversation

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

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

Définir vos agents

Créez des agents spécialisés avec des rôles distincts :

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

Configurer la conversation de groupe avec un sélecteur simple

Créez une conversation de groupe avec une logique de sélection personnalisée de l’orateur :

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

Configurer la conversation de groupe avec Agent-Based Manager

Vous pouvez également utiliser un gestionnaire basé sur un agent pour la sélection intelligente de l’orateur. Le gestionnaire est entièrement ChatAgent avec accès aux outils, au contexte et à la surveillance des systèmes :

# 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()
)

Exécuter le flux de travail de conversation de groupe

Exécutez les événements de flux de travail et de processus :

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.")

Exemple d’interaction

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.

Concepts clés

• Gestionnaire centralisé : la conversation de groupe utilise un responsable pour coordonner la sélection et le flux de l’orateur
• AgentWorkflowBuilder.CreateGroupChatBuilderWith() : crée des flux de travail avec une fonction de fabrique de gestion
• RoundRobinGroupChatManager : gestionnaire intégré qui alterne les intervenants en tour de rôle
• MaximumIterationCount : contrôle le nombre maximal de tours d’agent avant l’arrêt
• Gestionnaires personnalisés : étendre RoundRobinGroupChatManager ou implémenter une logique personnalisée
• Affinement itératif : les agents examinent et améliorent les contributions des uns et des autres
• Contexte partagé : tous les participants voient l’historique complet des conversations

• Stratégies de gestionnaire flexible : choisir entre des sélecteurs simples, des gestionnaires basés sur des agents ou une logique personnalisée
• GroupChatBuilder : crée des flux de travail avec la sélection configurable de l’orateur
• set_select_speakers_func() : définir des fonctions Python personnalisées pour la sélection de l’orateur
• set_manager() : utiliser un gestionnaire basé sur un agent pour la coordination intelligente de l’orateur
• GroupChatStateSnapshot : fournit un état de conversation pour les décisions de sélection
• Collaboration itérative : les agents s’appuient sur les contributions des uns et des autres
• Streaming d’événements : traiter AgentRunUpdateEvent et WorkflowOutputEvent en temps réel
• list[ChatMessage] Sortie : toutes les orchestrations retournent une liste de messages de conversation

Avancé : Sélection personnalisée de l’orateur

Vous pouvez implémenter une logique de gestionnaire personnalisé en créant un gestionnaire de conversation de groupe personnalisé :

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

Vous pouvez implémenter une logique de sélection sophistiquée en fonction de l’état de conversation :

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

Quand utiliser la conversation de groupe

L’orchestration de conversation de groupe est idéale pour :

• Affinement itératif : séries multiples d’examen et d’amélioration
• Résolution des problèmes collaboratifs : Agents disposant d’une expertise complémentaire travaillant ensemble
• Création de contenu : workflows d'écrivain et de relecteur pour la création de documents
• Analyse multi-perspective : obtention de points de vue variés sur la même entrée
• Assurance qualité : processus d’évaluation et d’approbation automatisés

Tenez compte des alternatives quand :

• Vous avez besoin d’un traitement séquentiel strict (utiliser l’orchestration séquentielle)
• Les agents doivent fonctionner complètement indépendamment (utiliser l’orchestration simultanée)
• Les transferts directs d'agent à agent sont nécessaires (utiliser l'orchestration du transfert)
• Une planification dynamique complexe est requise (utiliser l’orchestration Magentic)

Étapes suivantes

Orchestration de passation