Udostępnij przez

Orkiestracje przepływów pracy programu Microsoft Agent Framework — czat grupowy

Orkiestracja czatów grupowych modeluje współpracę konwersacji między wieloma agentami koordynowanymi przez menedżera, który określa wybór osoby mówiącej i przepływ konwersacji. Ten wzorzec jest idealny w scenariuszach wymagających uściślenia iteracyjnego, wspólnego rozwiązywania problemów lub analizy wielo perspektyw.

Różnice między czatem grupy i innymi wzorcami

Orkiestracja czatu grupowego ma różne cechy w porównaniu do innych wzorców multi-agentowych.

  • Scentralizowana koordynacja: w przeciwieństwie do wzorców przekazywania kontroli, w których agenci bezpośrednio przekazują kontrolę, czat grupowy używa menedżera do koordynowania kolejności osób mówiących
  • Uściślenie iteracyjne: agenty mogą przeglądać i opierać się nawzajem na swoich wzajemnych odpowiedziach w wielu rundach
  • Elastyczny wybór mówcy: Menedżer może używać różnych strategii (rotacyjnej, opartej na podpowiedziach, niestandardowej logiki) do wybierania mówców
  • Kontekst udostępniony: wszyscy agenci widzą pełną historię konwersacji, umożliwiając wspólne uściślenie

Czego nauczysz się

  • Jak utworzyć wyspecjalizowanych agentów na potrzeby współpracy grupowej
  • Jak skonfigurować strategie wyboru osoby mówiącej
  • Jak tworzyć przepływy pracy za pomocą iteracyjnego doskonalenia agenta
  • Jak dostosować przepływ konwersacji za pomocą własnych menedżerów

Konfigurowanie klienta usługi 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();

Definiowanie agentów

Utwórz wyspecjalizowanych agentów dla różnych ról w konwersacji grupowej:

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

Konfigurowanie czatu grupowego za pomocą menedżera Round-Robin

Skompiluj przepływ pracy czatu grupowego przy użyciu polecenia 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();

Uruchamianie przepływu pracy czatu grupowego

Wykonaj przepływ pracy i obserwuj konwersację iteracyjną:

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

Przykładowa interakcja

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

Konfigurowanie klienta czatu

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

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

Definiowanie agentów

Utwórz wyspecjalizowanych agentów z różnymi rolami:

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

Konfigurowanie czatu grupowego przy użyciu prostego selektora

Skompiluj czat grupowy przy użyciu niestandardowej logiki wyboru osoby mówiącej:

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

Konfigurowanie czatu grupowego z menedżerem opartym na agencie

Alternatywnie należy użyć menedżera opartego na agencie do inteligentnego wyboru osoby mówiącej. Menedżer jest pełnoprawnym ChatAgent z dostępem do narzędzi, kontekstu i możliwości obserwacji.

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

Uruchamianie przepływu pracy czatu grupowego

Wykonaj przepływ pracy i zrealizuj zdarzenia procesu:

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

Przykładowa interakcja

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.

Kluczowe pojęcia

  • Scentralizowany menedżer: czat grupowy używa menedżera do koordynowania wyboru i przepływu osoby mówiącej
  • AgentWorkflowBuilder.CreateGroupChatBuilderWith(): Tworzy przepływy pracy z funkcją fabryki menedżera
  • RoundRobinGroupChatManager: Wbudowany menedżer, który zmienia głośniki w sposób okrężny
  • MaximumIterationCount: określa maksymalną liczbę ruchów agenta przed zakończeniem
  • Menedżerowie niestandardowi: Rozszerz RoundRobinGroupChatManager lub implementuj logikę niestandardową
  • Iteracyjne uściślenie: Agenci przeglądają i ulepszają nawzajem swoje wkłady
  • Kontekst udostępniony: wszyscy uczestnicy widzą pełną historię konwersacji
  • Strategie elastycznego menedżera: wybór między prostymi selektorami, menedżerami opartymi na agentach lub logiką niestandardową
  • GroupChatBuilder: tworzy przepływy pracy z konfigurowalnym wyborem osoby mówiącej
  • set_select_speakers_func(): Definiowanie niestandardowych funkcji języka Python na potrzeby wyboru osoby mówiącej
  • set_manager(): Używanie menedżera opartego na agencie do inteligentnej koordynacji osoby mówiącej
  • GroupChatStateSnapshot: zapewnia stan konwersacji do podejmowania decyzji dotyczących wyboru
  • Iteracyjna współpraca: Agenci opierają się na wkładach wzajemnych
  • Przesyłanie strumieniowe zdarzeń: przetwarzanie AgentRunUpdateEvent i WorkflowOutputEvent w czasie rzeczywistym
  • list[ChatMessage] Output: Wszystkie aranżacje zwracają listę wiadomości czatu

Zaawansowane: wybór niestandardowego głośnika

Logikę niestandardowego menedżera można zaimplementować, tworząc niestandardowego menedżera czatów grupy:

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

Możesz zaimplementować zaawansowaną logikę wyboru na podstawie stanu konwersacji:

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

Kiedy używać czatu grupowego

Orkiestracja czatów grupowych jest idealna dla:

  • Iteracyjne doprecyzowanie: wiele rund przeglądu i ulepszania
  • Wspólne rozwiązywanie problemów: Agenci z uzupełniającą wiedzą, współpracują ze sobą
  • Tworzenie zawartości: przepływy pracy pisarza i recenzenta przy tworzeniu dokumentów
  • Analiza z wieloma perspektywami: pobieranie różnych punktów widzenia na tych samych danych wejściowych
  • Kontrola jakości: zautomatyzowane procesy przeglądu i zatwierdzania

Rozważ alternatywy, gdy:

  • Potrzebujesz ścisłego przetwarzania sekwencyjnego (użyj orkiestracji sekwencyjnej)
  • Agenci powinni działać całkowicie niezależnie (użyj orkiestracji współbieżnej)
  • Wymagane są bezpośrednie przekazywanie pomiędzy agentami (użyj koordynacji przekazywania)
  • Wymagane jest złożone planowanie dynamiczne (użyj orkiestracji magentycznej)

Dalsze kroki

Orkiestracja handoff

  • Last updated on