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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

В качестве альтернативы используйте агентский оркестратор для интеллектуального выбора громкоговорителей. Оркестратор — полноценный 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
""",
    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()

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

Запустите рабочий процесс и обработайте обновления участников потоковой передачи. Непотоковый вывод терминала представляет собой AgentResponse; потоковый вывод терминала выдаётся в виде фрагментов 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.")

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

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.

Настройка конфигурации 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
}

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

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

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

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

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

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

Создайте рабочий процесс группового чата с помощью agentworkflow.NewGroupChatWorkflowBuilder. Построитель принимает фабрику менеджеров и участвующих агентов. NewRoundRobinGroupChatManager выбирает каждого агента по очереди и останавливается по достижении настроенного максимального числа ходов участников.

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
}

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

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

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

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

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

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

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

• Гибкие стратегии оркестратора: выберите между простыми селекторами, оркестраторами на основе агента или пользовательской логикой с помощью параметров конструктора (selection_func, orchestrator_agent или orchestrator).
• GroupChatBuilder: создает рабочие процессы с настраиваемым выбором говорящего
• GroupChatState: предоставляет состояние беседы для принятия решений о выборе
• Итеративная совместная работа: агенты опираются на вклад друг друга
• Выходные данные AgentResponse: данные терминала содержат AgentResponse сообщение о завершении работы оркестратора.
• Потоковая передача событий: обработка AgentResponseUpdate событий в режиме реального времени с помощью workflow.run(task, stream=True)
• Промежуточные выходные данные: передайте intermediate_output_from=[participant, ...], чтобы выводить результаты каждого указанного участника в виде событий "intermediate", помимо завершающего события "output" оркестратора

• GroupChatWorkflowBuilder: создает схему со звездообразной топологией, где в центре находится хост группового чата, а размещённые на нём агенты выступают в роли участников
• GroupChatManager: выбирает следующего участника, может обновить журнал трансляции и завершить беседу.
• NewRoundRobinGroupChatManager: встроенный менеджер, который чередует участников в циклическом порядке
• RoundRobinGroupChatOptions: настраивает максимальное количество ходов участников и необязательную функцию завершения.
• События вывода: по умолчанию результаты участников являются промежуточными событиями, а организатор группового чата возвращает итоговую стенограмму
• Настраиваемые менеджеры. Реализация SelectNextAgent и необязательный обратный вызов жизненного цикла для выбора пользовательского докладчика или контрольного состояния

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

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

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, заданные в построителе, будут игнорироваться.

Промежуточные выходные данные

По умолчанию только финальный результат оркестратора появляется как "output" (конечное) событие рабочего процесса. Передайте intermediate_output_from участникам, которые вы хотите назначить в качестве промежуточных источников, чтобы также отображать отдельные выходные данные в качестве "intermediate" событий:

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

Реализуйте настраиваемый выбор динамика, возвращая GroupChatManager из фабрики менеджера построителя:

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также поддерживает UpdateHistory, ResetOnCheckpointи OnCheckpointRestored обратные вызовы для расширенных менеджеров, которые фильтруют широковещательные сообщения или сохраняют состояние, принадлежащее менеджеру.

Промежуточные выходные данные

По умолчанию GroupChatWorkflowBuilder передаёт результаты участника как промежуточные выходные данные рабочего процесса, а накопленную стенограмму беседы — как итоговый выходной результат. Используйте OutputEvent.IsIntermediate(), чтобы отличать обновления участников от окончательной расшифровки:

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

Вызов WithOutputFrom или WithIntermediateOutputFrom у конструктора группового чата переключает его на явное указание выходных данных. Используйте эти методы, если вам нужны выходные данные выбранных участников вместо окончательной расшифровки по умолчанию и всех промежуточных выходных данных участников.

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

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

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

Tip

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

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

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

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

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

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

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

Дальнейшие действия

Magentic Orchestration