Orchestrazioni dei flussi di lavoro di Microsoft Agent Framework - Chat di gruppo

Configurare il client OpenAI di Azure

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

Definire gli agenti

Creare agenti specializzati per ruoli diversi nella conversazione di gruppo:

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

Configurare la chat di gruppo con Round-Robin Orchestrator

Creare il flusso di lavoro di chat di gruppo usando 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();

Eseguire il flusso di lavoro di Chat di gruppo

Eseguire il flusso di lavoro e osservare la conversazione iterativa:

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

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

await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
{
    if (evt is AgentResponseUpdateEvent update)
    {
        // Process streaming agent responses
        AgentResponse 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;
    }
}

Interazione di esempio

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

Configurare il client chat

import os

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

# Initialize the Azure OpenAI client
client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

Definire gli agenti

Creare agenti specializzati con ruoli distinti:

from agent_framework import Agent

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

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

Configurare La chat di gruppo con selettore semplice

Creare una chat di gruppo con logica di selezione voce personalizzata:

from agent_framework.orchestrations import GroupChatBuilder, GroupChatState

def round_robin_selector(state: GroupChatState) -> str:
    """A round-robin selector function that picks the next speaker based on the current round index."""

    participant_names = list(state.participants.keys())
    return participant_names[state.current_round % len(participant_names)]


# Build the group chat workflow
workflow = GroupChatBuilder(
    participants=[researcher, writer],
    termination_condition=lambda conversation: len(conversation) >= 4,
    intermediate_output_from=[researcher, writer],
    selection_func=round_robin_selector,
).build()

Configurare la chat di gruppo con orchestratore basato su agenti

In alternativa, usa un orchestratore basato su agenti per la selezione intelligente degli altoparlanti. L'orchestratore è completo Agent con accesso a strumenti, contesto e osservabilità:

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

Guidelines:
- Start with Researcher to gather information
- Then have Writer synthesize the final answer
- Only finish after both have contributed meaningfully
""",
    client=client,
)

# Build group chat with agent-based orchestrator
workflow = GroupChatBuilder(
    participants=[researcher, writer],
    # Set a hard termination condition: stop after 4 assistant messages
    # The agent orchestrator will intelligently decide when to end before this limit but just in case
    termination_condition=lambda messages: sum(1 for msg in messages if msg.role == "assistant") >= 4,
    orchestrator_agent=orchestrator_agent,
    intermediate_output_from=[researcher, writer],
).build()

Eseguire il flusso di lavoro di Chat di gruppo

Esegui il flusso di lavoro ed elabora gli aggiornamenti in streaming dei partecipanti. L'output del terminale non in streaming è un oggetto AgentResponse; l'output del terminale in streaming viene emesso in blocchi AgentResponseUpdate.

from agent_framework import AgentResponseUpdate, Message

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

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

last_author: str | None = None
# Run the workflow with streaming enabled
stream = workflow.run(task, stream=True)
async for event in stream:
    if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
        # Print streaming agent updates
        author = event.data.author_name
        if author != last_author:
            if last_author is not None:
                print()
            print(f"[{author}]:", end=" ", flush=True)
            last_author = author
        print(event.data.text, end="", flush=True)
result = await stream.get_final_response()
if outputs := result.get_outputs():
    print("\n\n" + "=" * 80)
    print("Final Response:")
    print(outputs[-1])

print("\nWorkflow completed.")

Interazione di esempio

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.

Configurare la configurazione di Foundry

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
}

Definire gli agenti

Creare agenti specializzati con ruoli distinti nella conversazione:

copywriter := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a creative copywriter. Generate catchy slogans and marketing copy. Be concise and impactful.",
        Config:      agent.Config{Name: "CopyWriter"},
    },
)

reviewer := foundryprovider.NewAgent(
    endpoint,
    token,
    foundryprovider.ModelDeployment(model),
    foundryprovider.AgentConfig{
        Instructions: "You are a marketing reviewer. Evaluate slogans for clarity, impact, and brand alignment. Provide constructive feedback or approval.",
        Config:      agent.Config{Name: "Reviewer"},
    },
)

Configurare La chat di gruppo con Round-Robin Manager

Compilare il flusso di lavoro della chat di gruppo con agentworkflow.NewGroupChatWorkflowBuilder. Il builder accetta una factory di manager e gli agenti partecipanti. NewRoundRobinGroupChatManager seleziona ogni agente a turno e si interrompe dopo il numero massimo configurato di turni del partecipante.

managerFactory := func(agents []*agent.Agent) *agentworkflow.GroupChatManager {
    return agentworkflow.NewRoundRobinGroupChatManager(
        agents,
        agentworkflow.RoundRobinGroupChatOptions{MaximumIterationCount: 5},
    )
}

wf, err := agentworkflow.NewGroupChatWorkflowBuilder(managerFactory, copywriter, reviewer).
    WithName("Marketing Review Group Chat").
    WithDescription("A copywriter and reviewer collaborate on marketing copy.").
    Build()
if err != nil {
    return err
}

Eseguire il flusso di lavoro di Chat di gruppo

Eseguire il flusso di lavoro con un messaggio utente e un token di turno. Quando l'emissione di eventi è abilitata, gli aggiornamenti dei partecipanti arrivano come eventi di output intermedi e la trascrizione finale arriva come evento di output del terminale.

run, err := inproc.Default.RunStreaming(ctx, wf, []*message.Message{
    message.NewText("Create a slogan for an eco-friendly electric vehicle."),
})
if err != nil {
    return err
}
defer run.Close(ctx)

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

lastExecutorID := ""
for evt, err := range run.WatchStream(ctx) {
    if err != nil {
        return err
    }

    switch e := evt.(type) {
    case workflow.OutputEvent:
        switch value := e.Output.(type) {
        case *agent.ResponseUpdate:
            if e.ExecutorID != lastExecutorID {
                lastExecutorID = e.ExecutorID
                fmt.Printf("\n[%s]: ", e.ExecutorID)
            }
            fmt.Print(value.String())
        case []*message.Message:
            fmt.Println("\n\n=== Final Conversation ===")
            for _, msg := range value {
                author := msg.AuthorName
                if author == "" {
                    author = string(msg.Role)
                }
                fmt.Printf("%s: %s\n", author, msg.String())
            }
        }
    case workflow.ErrorEvent:
        return e.Error
    case workflow.ExecutorFailedEvent:
        return fmt.Errorf("executor %q failed: %w", e.ExecutorID, e.Error)
    }
}

Interazione di esempio

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

[Reviewer]: This is clear and memorable. It communicates performance and sustainability directly.
Approved.

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

=== Final Conversation ===
user: Create a slogan for an eco-friendly electric vehicle.
CopyWriter: "Pure Power, Zero Impact" - Experience electric performance without compromise.
Reviewer: This is clear and memorable. It communicates performance and sustainability directly. Approved.
CopyWriter: The final slogan is: "Pure Power, Zero Impact" - Experience electric performance without compromise.

• Gestione centralizzata: la chat di gruppo usa un manager per coordinare la selezione e il flusso dell'altoparlante
• AgentWorkflowBuilder.CreateGroupChatBuilderWith(): crea flussi di lavoro con una funzione factory di gestione
• RoundRobinGroupChatManager: gestore predefinito che alterna gli altoparlanti in modalità round robin
• MaximumIterationCount: controlla il numero massimo di turni dell'agente prima della terminazione
• Gestori personalizzati: estendere RoundRobinGroupChatManager o implementare la logica personalizzata
• Perfezionamento iterativo: gli agenti esaminano e migliorano i contributi degli altri
• Contesto condiviso: tutti i partecipanti visualizzano la cronologia completa della conversazione

• Strategie dell'agente di orchestrazione flessibile: scegliere tra selettori semplici, agenti di orchestrazione basati su agenti o logica personalizzata tramite parametri del costruttore (selection_func, orchestrator_agento orchestrator).
• GroupChatBuilder: crea flussi di lavoro con selezione voce configurabile
• GroupChatState: fornisce lo stato della conversazione per le decisioni di selezione
• Collaborazione iterativa: gli agenti si basano sui contributi degli altri
• Output di AgentResponse: l'output del terminale è un AgentResponse contenente il messaggio di completamento dell'orchestratore
• Streaming di eventi: elaborare AgentResponseUpdate gli eventi in tempo reale tramite workflow.run(task, stream=True)
• Output intermedi: passa intermediate_output_from=[participant, ...] per esporre l'output di ciascun partecipante elencato come eventi "intermediate", oltre all'evento terminale "output" dell'orchestratore

• GroupChatWorkflowBuilder: crea un flusso di lavoro di topologia star con un host di chat di gruppo nel centro e gli agenti ospitati come partecipanti
• GroupChatManager: seleziona il partecipante successivo, può aggiornare la cronologia delle trasmissioni e può terminare la conversazione
• NewRoundRobinGroupChatManager: gestore integrato che alterna i partecipanti in ordine ciclico
• RoundRobinGroupChatOptions: configura il numero massimo di turni dei partecipanti e una funzione di terminazione facoltativa
• Eventi di output: per impostazione predefinita, gli output dei partecipanti sono eventi intermedi e l'host della chat di gruppo restituisce la trascrizione del terminale
• Gestori personalizzati: implementare SelectNextAgent e callback facoltative del ciclo di vita per la selezione personalizzata dello speaker o dello stato salvato tramite checkpoint

È possibile implementare la logica di gestione personalizzata creando un gestore chat di gruppo personalizzato:

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

È possibile implementare una logica di selezione sofisticata in base allo stato della conversazione:

def smart_selector(state: GroupChatState) -> str:
    """Select speakers based on conversation content and context."""
    conversation = state.conversation

    last_message = conversation[-1] if conversation else None

    # If no messages yet, start with Researcher
    if not last_message:
        return "Researcher"

    # Check last message content
    last_text = last_message.text.lower()

    # If researcher finished gathering info, switch to writer
    if "i have finished" in last_text and last_message.author_name == "Researcher":
        return "Writer"

    # Else continue with researcher until it indicates completion
    return "Researcher"

workflow = GroupChatBuilder(
    participants=[researcher, writer],
    selection_func=smart_selector,
).build()

Output intermediari

Per impostazione predefinita, solo l'output finale dell'orchestratore appare come evento del flusso di lavoro "output" (terminale). Passa intermediate_output_from con i partecipanti che vuoi designare come sorgenti intermedie, in modo da esporre anche i rispettivi output individuali come eventi "intermediate":

workflow = GroupChatBuilder(
    participants=[researcher, writer],
    termination_condition=lambda conversation: len(conversation) >= 4,
    selection_func=round_robin_selector,
    intermediate_output_from=[researcher, writer],
).build()

Implementa la selezione personalizzata dell'altoparlante restituendo un GroupChatManager dalla manager factory del builder:

type approvalManager struct {
    agents []*agent.Agent
}

func newApprovalManager(agents []*agent.Agent) *agentworkflow.GroupChatManager {
    manager := &approvalManager{agents: agents}
    return &agentworkflow.GroupChatManager{
        SelectNextAgent: manager.selectNextAgent,
        ShouldTerminate: manager.shouldTerminate,
    }
}

func (m *approvalManager) selectNextAgent(_ context.Context, history []*message.Message) (*agent.Agent, error) {
    last := lastAssistantMessage(history)
    if last == nil || last.AuthorName == "Reviewer" {
        return m.agentByName("CopyWriter")
    }
    return m.agentByName("Reviewer")
}

func (m *approvalManager) shouldTerminate(_ context.Context, history []*message.Message, iterationCount int) (bool, error) {
    if iterationCount >= 10 {
        return true, nil
    }
    last := lastAssistantMessage(history)
    return last != nil &&
        last.AuthorName == "Reviewer" &&
        strings.Contains(strings.ToLower(last.String()), "approve"), nil
}

func (m *approvalManager) agentByName(name string) (*agent.Agent, error) {
    for _, currentAgent := range m.agents {
        if currentAgent.Name() == name {
            return currentAgent, nil
        }
    }
    return nil, fmt.Errorf("agent %q is not part of the group chat", name)
}

func lastAssistantMessage(history []*message.Message) *message.Message {
    for i := len(history) - 1; i >= 0; i-- {
        if history[i].Role == message.RoleAssistant {
            return history[i]
        }
    }
    return nil
}

wf, err := agentworkflow.NewGroupChatWorkflowBuilder(newApprovalManager, copywriter, reviewer).
    WithName("Approval Group Chat").
    Build()

GroupChatManager supporta anche i callback UpdateHistory, Reset, OnCheckpoint e OnCheckpointRestored per i manager avanzati che filtrano i messaggi broadcast o mantengono in modo persistente lo stato del manager.

Output intermediari

Per impostazione predefinita, GroupChatWorkflowBuilder genera gli output dei partecipanti come output intermedi del flusso di lavoro e genera la trascrizione cumulativa della conversazione come output finale. Usare OutputEvent.IsIntermediate() per distinguere gli aggiornamenti dei partecipanti dalla trascrizione finale:

if output, ok := evt.(workflow.OutputEvent); ok {
    if output.IsIntermediate() {
        fmt.Printf("intermediate from %s: %v\n", output.ExecutorID, output.Output)
        return nil
    }

    fmt.Printf("terminal output: %v\n", output.Output)
}

La chiamata WithOutputFrom o WithIntermediateOutputFrom sul generatore di chat di gruppo passa alla designazione di output esplicita. Utilizzare questi metodi quando si desidera che gli output dei partecipanti selezionati vengano visualizzati anziché la trascrizione finale predefinita e tutti gli output intermedi dei partecipanti.