Orquestrações de fluxos de trabalho do Microsoft Agent Framework - Chat de Grupo

Configurar o Cliente OpenAI do 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);

Defina seus agentes

Crie agentes especializados para diferentes funções na conversa em grupo:

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

Configurar o Chat de Grupo com Round-Robin Orchestrator

Crie o fluxo de trabalho de chat de grupo 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();

Executar o fluxo de trabalho do Chat de Grupo

Execute o fluxo de trabalho e observe a conversa 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;
    }
}

Interação de amostra

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

Configurar o cliente de 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(),
)

Defina seus agentes

Crie agentes especializados com funções distintas:

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

Configurar o Chat de Grupo com o Seletor Simples

Crie um bate-papo em grupo com lógica de seleção de alto-falante personalizada:

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

Configurar o Chat de Grupo com Orquestrador Baseado em Agentes

Em alternativa, utilize um orquestrador baseado em agentes para a seleção inteligente de altifalantes. O orquestrador completo tem acesso a ferramentas, contexto e observabilidade:

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

Executar o fluxo de trabalho do Chat de Grupo

Execute o fluxo de trabalho e processe as atualizações dos participantes em streaming. A saída do terminal sem transmissão é uma AgentResponse; a saída do terminal com transmissão é emitida em blocos 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.")

Interação de amostra

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.

Configurar a configuração do 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
}

Defina seus agentes

Crie agentes especializados com papéis distintos na conversa:

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

Configurar o Chat de Grupo com o Round-Robin Manager

Constrói o fluxo de trabalho do grupo de chat com agentworkflow.NewGroupChatWorkflowBuilder. O construtor leva uma fábrica gestora e os agentes participantes. NewRoundRobinGroupChatManager seleciona cada agente à vez e termina após o número máximo configurado de turnos dos participantes.

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
}

Executar o fluxo de trabalho do Chat de Grupo

Executa o fluxo de trabalho com uma mensagem de utilizador e um token de turno. Quando a emissão de eventos está ativada, as atualizações dos participantes chegam como eventos intermédios de saída e a transcrição final chega como um evento de saída terminal.

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

Interação de amostra

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

• Gerente centralizado: o bate-papo em grupo usa um gerente para coordenar a seleção e o fluxo de alto-falantes
• AgentWorkflowBuilder.CreateGroupChatBuilderWith(): cria fluxos de trabalho com uma função de fábrica do gerenciador
• RoundRobinGroupChatManager: Gestor integrado que alterna oradores de forma rotativa
• MaximumIterationCount: Controla o número máximo de turnos do agente antes do término
• Gerenciadores personalizados: Estenda RoundRobinGroupChatManager ou implemente a lógica personalizada
• Refinamento iterativo: os agentes analisam e melhoram as contribuições uns dos outros
• Contexto compartilhado: todos os participantes veem o histórico completo da conversa

• Estratégias Flexíveis para Orquestradores: Escolha entre seletores simples, orquestradores baseados em agentes ou lógica personalizada através de parâmetros do construtor (selection_func, orchestrator_agent, ou orchestrator).
• GroupChatBuilder: Cria fluxos de trabalho com seleção configurável de alto-falantes
• GroupChatState: Fornece estado de conversa para decisões de seleção
• Colaboração iterativa: os agentes aproveitam as contribuições uns dos outros
• Saída do AgentResponse: A saída do terminal é um AgentResponse que contém a mensagem de conclusão do orquestrador
• Streaming de Eventos: Processar eventos AgentResponseUpdate em tempo real através de workflow.run(task, stream=True)
• Saídas Intermédias: Passe intermediate_output_from=[participant, ...] para expor a saída de cada participante indicado em eventos "intermediate", além do evento terminal "output" do orquestrador

• GroupChatWorkflowBuilder: Cria um fluxo de trabalho de topologia em estrelas com um anfitrião de chat de grupo no centro e agentes alojados como participantes
• GroupChatManager: Seleciona o próximo participante, pode atualizar o histórico de transmissão e terminar a conversa
• NewRoundRobinGroupChatManager: Gestor integrado que alterna os participantes em alternância circular
• RoundRobinGroupChatOptions: Configura o número máximo de turnos dos participantes e uma função opcional de terminação
• Eventos de Saída: Por defeito, as saídas dos participantes são eventos intermédios e o anfitrião do grupo de chat fornece a transcrição do terminal
• Gestores personalizados: Implemente SelectNextAgent e chamadas de retorno opcionais do ciclo de vida para seleção personalizada de altifalantes ou estado com ponto de verificação

Você pode implementar a lógica do gerenciador personalizado criando um gerenciador de chat de grupo personalizado:

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

Você pode implementar uma lógica de seleção sofisticada com base no estado da conversa:

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

Saídas Intermédias

Por defeito, apenas a saída final do orquestrador é apresentada como um evento de fluxo de trabalho "output" (terminal). Transmita intermediate_output_from com os participantes que pretende designar como fontes intermédias para também expor as respetivas saídas individuais como eventos "intermediate":

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

Implemente a seleção personalizada de orador devolvendo um GroupChatManager da fábrica do gestor do 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 também suporta UpdateHistory, Reset, OnCheckpoint e OnCheckpointRestored funções de retorno para gestores avançados que filtram mensagens difundidas ou persistem o estado pertencente ao gestor.

Saídas Intermédias

Por defeito, GroupChatWorkflowBuilder emite os resultados dos participantes como saídas intermédias do fluxo de trabalho e emite a transcrição acumulada da conversa como saída final. Use OutputEvent.IsIntermediate() para distinguir as atualizações dos participantes da transcrição final:

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

Chamar WithOutputFrom ou WithIntermediateOutputFrom no construtor de chat de grupo faz com que este passe para a designação explícita da saída. Use estes métodos quando quiser resultados selecionados dos participantes em vez da transcrição final padrão mais todos os resultados intermédios dos participantes.