Оркестрации рабочих процессов Microsoft Agent Framework — групповый чат

Оркестрация группового чата моделирует совместную беседу между несколькими агентами, координируемую оркестратором, который определяет выбор говорящего и поток беседы. Этот шаблон идеально подходит для сценариев, требующих итеративного уточнения, совместного решения проблем или анализа с несколькими перспективами.

Внутри группы оркестрация чата собирает агентов в звездной топологии с оркестратором в середине. Оркестратор может реализовать различные стратегии выбора следующего агента, например по кругу, выбора на основе подсказки или с использованием пользовательской логики на основе контекста беседы, что делает его гибким и мощным шаблоном для совместной работы с несколькими агентами.

Различия между групповым чатом и другими шаблонами

Оркестрация группового чата имеет уникальные характеристики по сравнению с другими шаблонами с несколькими агентами:

• Централизованная координация: в отличие от шаблонов передачи управления, в которых агенты напрямую передают управление, групповый чат использует оркестратор для координации того, кто говорит следующим.
• Итеративное уточнение: агенты могут просматривать и развивать ответы друг друга в нескольких раундах
• Гибкий выбор выступающих: оркестратор может использовать различные стратегии (циклический перебор, основанный на подсказках, пользовательская логика) для выбора выступающих.
• Общий контекст: все агенты видят полную историю бесед, что позволяет улучшить совместную работу.

Цели обучения

• Создание специализированных агентов для групповой совместной работы
• Настройка стратегий выбора говорящего
• Создание рабочих процессов с помощью пошаговой оптимизации агентов
• Настройка потока беседы с помощью пользовательских оркестраторов

Настройка клиента 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();

Предупреждение

DefaultAzureCredential удобно для разработки, но требует тщательного рассмотрения в рабочей среде. В рабочей среде рекомендуется использовать определенные учетные данные (например, ManagedIdentityCredential), чтобы избежать проблем с задержкой, непреднамеренной проверки данных аутентификации и потенциальных рисков безопасности из-за резервных механизмов.

Определение агентов

Создайте специализированные агенты для разных ролей в беседе группы:

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

Настройка группового чата с помощью оркестратора Round-Robin

Создание рабочего процесса группового чата с помощью 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();

Запуск рабочего процесса группового чата

Выполните рабочий процесс и просмотрите итеративную беседу:

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

Пример взаимодействия

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

Настройка клиента чата

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

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

Определение агентов

Создайте специализированные агенты с различными ролями:

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

Настройка группового чата с помощью простого селектора

Создание группового чата с логикой выбора настраиваемого докладчика:

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

Настройка группового чата с оркестратором на основе агента

В качестве альтернативы используйте агентский оркестратор для интеллектуального выбора громкоговорителей. Оркестратор — полноценный Agent с доступом к инструментам, контексту и наблюдаемости.

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

Запуск рабочего процесса группового чата

Выполните рабочий процесс и обработайте события.

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

Пример взаимодействия

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.

Основные понятия

• Централизованный менеджер: групповый чат использует диспетчер для координации выбора и потока говорящего
• AgentWorkflowBuilder.CreateGroupChatBuilderWith(): создает рабочие процессы с помощью фабричной функции менеджера
• RoundRobinGroupChatManager: встроенный менеджер, который чередует участников по кругу
• MaximumIterationCount: управляет максимальным числом итераций агента перед завершением.
• Настраиваемые диспетчеры: расширение RoundRobinGroupChatManager или реализация пользовательской логики
• Итеративное уточнение: агенты просматривают и улучшают вклад друг друга
• Общий контекст: все участники видят полную историю бесед

• Гибкие стратегии оркестратора: выберите между простыми селекторами, оркестраторами на основе агента или пользовательской логикой с помощью параметров конструктора (selection_func, orchestrator_agent или orchestrator).
• GroupChatBuilder: создает рабочие процессы с настраиваемым выбором говорящего
• GroupChatState: предоставляет состояние беседы для принятия решений о выборе
• Итеративная совместная работа: агенты опираются на вклад друг друга
• Потоковая передача событий: обработка WorkflowOutputEvent данных AgentResponseUpdate в режиме реального времени
• list[Message] Выходные данные: все оркестрации возвращают список сообщений чата

Дополнительно: выбор настраиваемого говорящего

Вы можете реализовать логику пользовательского диспетчера, создав пользовательский диспетчер групповых чатов:

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

Вы можете реализовать сложную логику выбора на основе состояния беседы:

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

Это важно

При использовании пользовательской BaseGroupChatOrchestrator реализации для расширенных сценариев необходимо задать все свойства, в том числе participant_registry, max_roundsи termination_condition. max_rounds и termination_condition, заданные в построителе, будут игнорироваться.

Синхронизация контекста

Как упоминалось в начале этого руководства, все агенты в групповом чате видят всю историю бесед.

Агенты в Agent Framework используют сеансы агента (AgentSession) для управления контекстом. В оркестрации группового чата агенты не используют один и тот же экземпляр сеанса, но оркестратор гарантирует синхронизацию сеанса каждого агента с полным журналом бесед перед каждым ходом. Чтобы добиться этого, после хода каждого агента оркестратор транслирует ответ всем остальным агентам, гарантируя, что у всех участников имеется актуальный контекст для следующего хода.

Подсказка

Агенты не используют один экземпляр сеанса, так как разные типы агентовAgentSession могут иметь разные реализации абстракции. Совместное использование одного экземпляра сеанса может привести к непоследовательностям в том, как каждый агент обрабатывает и поддерживает контекст.

После трансляции ответа оркестратор затем решает следующего докладчика и отправляет запрос выбранному агенту, который теперь имеет полную историю бесед для создания ответа.

Когда следует использовать групповой чат

Оркестрация группового чата идеально подходит для:

• Итеративное уточнение: несколько раундов обзора и улучшения
• Совместное решение проблем: агенты с дополняющими знаниями, работающие вместе
• Создание содержимого: рабочие процессы авторов и рецензентов для создания документа
• Анализ с несколькими перспективами: получение различных точек зрения для одних и того же входных данных
• Quality Assurance: автоматизированные процессы проверки и утверждения

Рассмотрите варианты, когда:

• Требуется строгая последовательная обработка (используйте последовательную оркестрацию)
• Агенты должны работать полностью независимо (использовать параллельную оркестрацию)
• Необходимы прямые передачи между агентами (используйте оркестрацию Handoff)
• Требуется сложное динамическое планирование (использование магентической оркестрации)

Дальнейшие шаги

Magentic Orchestration