Orchestrace pracovních postupů architektury Microsoft Agent – Skupinový chat

Orchestrace skupinového chatu modeluje konverzaci mezi několika agenty, která koordinuje orchestrátor, který určuje výběr mluvčího a tok konverzace. Tento model je ideální pro scénáře vyžadující iterativní upřesnění, řešení problémů ve spolupráci nebo analýzu s více perspektivami.

Orchestrace skupinového chatu interně sestavuje agenty do hvězdicové topologie s orchestrátorem uprostřed. Orchestrátor může implementovat různé strategie pro výběr toho, který agent mluví dál, jako je kruhové dotazování, výběr na základě výzvy nebo vlastní logika založená na kontextu konverzace, což z něj dělá flexibilní a výkonný vzor pro spolupráci s více agenty.

Rozdíly mezi skupinovými chaty a jinými vzory

Orchestrace skupinového chatu má odlišné charakteristiky v porovnání s jinými vzory více agentů:

• Centralizovaná koordinace: Na rozdíl od vzorů předávání, kdy agenti přímo přenesou řízení, používá skupinový chat orchestrátor ke koordinaci toho, kdo bude mluvit dál.
• Iterativní upřesnění: Agenti mohou přezkoumávat a zlepšovat odpovědi ostatních v několika kolech.
• Flexibilní výběr mluvčího: Orchestrátor může použít různé strategie (kruhové dotazování, založené na příkazovém řádku, vlastní logika) k výběru mluvčích.
• Sdílený kontext: Všichni agenti vidí celou historii konverzací a umožňují upřesnění spolupráce.

Co se naučíte

• Vytvoření specializovaných agentů pro spolupráci skupin
• Konfigurace strategií výběru mluvčího
• Jak vytvářet pracovní postupy se zdokonalováním agenta iterativním způsobem
• Přizpůsobení toku konverzace pomocí vlastních orchestrátorů

Nastavení klienta 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 DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

Výstraha

DefaultAzureCredential je vhodný pro vývoj, ale vyžaduje pečlivé zvážení v produkčním prostředí. V produkčním prostředí zvažte použití konkrétních přihlašovacích údajů (např ManagedIdentityCredential. ) k zabránění problémům s latencí, neúmyslnému testování přihlašovacích údajů a potenciálním bezpečnostním rizikům z náhradních mechanismů.

Definování agentů

Vytvořte specializované agenty pro různé role v konverzaci skupiny:

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

Konfigurace skupinového chatu pomocí nástroje Round-Robin Orchestrator

Vytvoření pracovního postupu skupinového chatu pomocí 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();

Spuštění workflowu skupinového chatu

Spusťte pracovní postup a sledujte iterativní konverzaci:

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

Ukázková interakce

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

Nastavení chatovacího klienta

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

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

Definování agentů

Vytváření specializovaných agentů s odlišnými rolemi:

from agent_framework import Agent

# Create a researcher agent
researcher = Agent(
    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 = Agent(
    name="Writer",
    description="Synthesizes polished answers using gathered information.",
    instructions="Compose clear, structured answers using any notes provided. Be comprehensive.",
    chat_client=chat_client,
)

Konfigurace skupinového chatu pomocí jednoduchého selektoru

Vytvoření skupinového chatu s vlastní logikou výběru mluvčího:

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,
    selection_func=round_robin_selector,
).build()

Konfigurace skupinového chatu pomocí nástroje Agent-Based Orchestrator

Případně můžete použít orchestrátor založený na agentech pro inteligentní výběr mluvčího. Orchestrátor má plný Agent přístup k nástrojům, kontextu a sledování:

# 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
""",
    chat_client=chat_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,
).build()

Spuštění workflowu skupinového chatu

Proveďte pracovní postup a zpracování událostí.

from typing import cast
from agent_framework import AgentResponseUpdate, Role

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

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

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

# Run the workflow
async for event in workflow.run_stream(task):
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        # 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 event.type == "output":
        # Workflow completed - data is a list of Message
        final_conversation = cast(list[Message], 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.")

Ukázková interakce

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.

Klíčové koncepty

• Centralizovaný manažer: Skupinový chat používá manažera ke koordinaci výběru a toku mluvčího.
• AgentWorkflowBuilder.CreateGroupChatBuilderWith(): Vytvoří pracovní postupy pomocí funkce továrny správců.
• RoundRobinGroupChatManager: Předdefinovaný manažer, který střídá mluvčí v kruhovém pořadí.
• MaximumIterationCount: Určuje maximální počet iterací agenta před ukončením.
• Vlastní manažeři: Rozšíření RoundRobinGroupChatManager nebo implementace vlastní logiky
• Iterativní upřesnění: Agenti kontrolují a vylepšují příspěvky ostatních
• Sdílený kontext: Všichni účastníci uvidí celou historii konverzací.

• Flexibilní strategie orchestratoru: Volba mezi jednoduchými selektory, orchestrátory založenými na agentech nebo vlastní logikou prostřednictvím parametrů konstruktoru (selection_func, orchestrator_agentnebo orchestrator).
• GroupChatBuilder: Vytvoří pracovní postupy s konfigurovatelným výběrem mluvčího.
• GroupChatState: Poskytuje stav konverzace pro rozhodnutí o výběru
• Iterativní spolupráce: Agenti staví na příspěvcích ostatních
• Streamování událostí: Zpracování WorkflowOutputEvent dat AgentResponseUpdate v reálném čase
• list[Zpráva] Výstup: Všechny orchestrace vrací seznam zpráv chatu.

Pokročilé: Výběr vlastního mluvčího

Logiku vlastního správce můžete implementovat vytvořením vlastního správce chatu skupiny:

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

Na základě stavu konverzace můžete implementovat sofistikovanou logiku výběru:

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

Důležité

Při použití vlastní implementace BaseGroupChatOrchestrator pro pokročilé scénáře musí být nastaveny všechny vlastnosti, včetně participant_registry, max_roundsa termination_condition. max_rounds a termination_condition nastavení v builderu bude ignorováno.

Synchronizace kontextu

Jak je uvedeno na začátku tohoto průvodce, všichni agenti ve skupinovém chatu uvidí celou historii konverzací.

Agenti v rozhraní Agent Framework spoléhají na relace agenta (AgentSession) ke správě kontextu. V orchestraci skupinového chatu agenti nesdílí stejnou instanci relace, ale orchestrátor zajistí synchronizaci relace jednotlivých agentů s úplnou historií konverzací před každým turnem. Aby toho bylo dosaženo, orchestrátor po tahu každého agenta vysílá odpověď všem ostatním agentům a zajistí, aby všichni účastníci měli nejnovější kontext pro jejich další tah.

Návod

Agenti nesdílejí stejnou instanci relace, protože různé typy agentů mohou mít různé implementace AgentSession abstrakce. Sdílení stejné instance relace může vést ke nekonzistencím v tom, jak každý agent zpracovává a udržuje kontext.

Po vysílání odpovědi orchestrátor rozhodne dalšího mluvčího a odešle žádost vybranému agentovi, který teď má úplnou historii konverzací k vygenerování odpovědi.

Kdy použít skupinovou chat

Orchestrace skupinového chatu je ideální pro:

• Iterativní upřesnění: Více kol kontroly a vylepšení
• Spolupráce při řešení problémů: Specialisté s doplňkovými odbornými znalostmi, kteří spolupracují
• Vytváření obsahu: Pracovní postupy pro autory revidujících pro vytváření dokumentů
• Analýza s více perspektivami: Získání různých hledisek na stejném vstupu
• Kontrola kvality: Automatizované procesy kontroly a schvalování

Zvažte alternativy v případech:

• Potřebujete striktní sekvenční zpracování (použijte sekvenční orchestraci).
• Agenti by měli pracovat zcela nezávisle (používat souběžnou orchestraci).
• Jsou potřeba přímé předání agenta agentovi (použijte orchestraci předání).
• Vyžaduje se komplexní dynamické plánování (použití magentické orchestrace)

Další kroky

Magentická orchestrace