Оркестрации рабочих процессов Microsoft Agent Framework — параллельные

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

Параллельная оркестрация

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

Определение нескольких агентов с различным опытом
Оркестрация этих агентов для параллельной работы с одной задачей
Как собирать и обрабатывать результаты

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

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

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Extensions.AI;
using Microsoft.Agents.AI;

// 1) 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), чтобы избежать проблем с задержкой, непреднамеренной проверки данных аутентификации и потенциальных рисков безопасности из-за резервных механизмов.

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

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

// 2) Helper method to create translation agents
static ChatClientAgent GetTranslationAgent(string targetLanguage, IChatClient chatClient) =>
    new(chatClient,
        $"You are a translation assistant who only responds in {targetLanguage}. Respond to any " +
        $"input by outputting the name of the input language and then translating the input to {targetLanguage}.");

// Create translation agents for concurrent processing
var translationAgents = (from lang in (string[])["French", "Spanish", "English"]
                         select GetTranslationAgent(lang, client));

Настройка параллельной оркестрации

Создайте рабочий процесс, используя AgentWorkflowBuilder для параллельного запуска агентов:

// 3) Build concurrent workflow
var workflow = AgentWorkflowBuilder.BuildConcurrent(translationAgents);

Запуск параллельного рабочего процесса и сбор результатов

Запустите рабочий процесс и обработайте события от всех агентов, работающих одновременно.

// 4) Run the workflow
var messages = new List<ChatMessage> { new(ChatRole.User, "Hello, world!") };

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, messages);
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

List<ChatMessage> result = new();
await foreach (WorkflowEvent evt in run.WatchStreamAsync())
{
    if (evt is AgentResponseUpdateEvent e)
    {
        Console.WriteLine($"{e.ExecutorId}: {e.Update.Text}");
    }
    else if (evt is WorkflowOutputEvent outputEvt)
    {
        result = outputEvt.As<List<ChatMessage>>()!;
        break;
    }
}

// Display aggregated results from all agents
Console.WriteLine("===== Final Aggregated Results =====");
foreach (var message in result)
{
    Console.WriteLine($"{message.Role}: {message.Text}");
}

Пример выходных данных

French_Agent: English detected. Bonjour, le monde !
Spanish_Agent: English detected. ¡Hola, mundo!
English_Agent: English detected. Hello, world!

===== Final Aggregated Results =====
User: Hello, world!
Assistant: English detected. Bonjour, le monde !
Assistant: English detected. ¡Hola, mundo!
Assistant: English detected. Hello, world!

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

Параллельное выполнение: все агенты обрабатывают входные данные одновременно и независимо
AgentWorkflowBuilder.BuildConcurrent(): создает параллельный рабочий процесс из коллекции агентов
Автоматическая агрегирование: Результаты всех агентов автоматически собираются в окончательный результат.
Потоковая передача событий: мониторинг в реальном времени прогресса агента AgentResponseUpdateEvent
Разнообразные перспективы: каждый агент приносит свой уникальный опыт в одну и ту же проблему

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

import os

from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

# 1) Create three domain agents using FoundryChatClient
chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher = chat_client.as_agent(
    instructions=(
        "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
        " opportunities, and risks."
    ),
    name="researcher",
)

marketer = chat_client.as_agent(
    instructions=(
        "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
        " aligned to the prompt."
    ),
    name="marketer",
)

legal = chat_client.as_agent(
    instructions=(
        "You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns"
        " based on the prompt."
    ),
    name="legal",
)

Настройка параллельной оркестрации

Класс ConcurrentBuilder позволяет создать рабочий процесс для параллельного запуска нескольких агентов. Вы передаете список агентов в качестве участников.

from agent_framework.orchestrations import ConcurrentBuilder

# 2) Build a concurrent workflow
# Participants are either Agents (type of SupportsAgentRun) or Executors
workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

Запуск параллельного рабочего процесса и сбор результатов

Агрегатор по умолчанию создает одно AgentResponse сообщение помощника для каждого участника:

from agent_framework import AgentResponse

# 3) Run with a single prompt and print the aggregated agent responses
events = await workflow.run("We are launching a new budget-friendly electric bike for urban commuters.")
outputs = events.get_outputs()

if outputs:
    print("===== Final Aggregated Results =====")
    final: AgentResponse = outputs[0]
    for msg in final.messages:
        name = msg.author_name or "assistant"
        print(f"{'-' * 60}\n\n[{name}]:\n{msg.text}")

Пример выходных данных

===== Final Aggregated Results =====
------------------------------------------------------------

[researcher]:
**Insights:**

- **Target Demographic:** Urban commuters seeking affordable, eco-friendly transport;
    likely to include students, young professionals, and price-sensitive urban residents.
- **Market Trends:** E-bike sales are growing globally, with increasing urbanization,
    higher fuel costs, and sustainability concerns driving adoption.
...
------------------------------------------------------------

[marketer]:
**Value Proposition:**
"Empowering your city commute: Our new electric bike combines affordability, reliability, and
    sustainable design—helping you conquer urban journeys without breaking the bank."
...
------------------------------------------------------------

[legal]:
**Constraints, Disclaimers, & Policy Concerns for Launching a Budget-Friendly Electric Bike for Urban Commuters:**

**1. Regulatory Compliance**
- Verify that the electric bike meets all applicable federal, state, and local regulations
    regarding e-bike classification, speed limits, power output, and safety features.

Расширенные возможности: пользовательские выполнители агента

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

Определение исполнителей пользовательских агентов

from agent_framework import (
    AgentExecutorRequest,
    AgentExecutorResponse,
    Agent,
    Executor,
    WorkflowContext,
    handler,
)

class ResearcherExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "researcher"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
                " opportunities, and risks."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

class MarketerExec(Executor):
    def __init__(self, chat_client: FoundryChatClient, id: str = "marketer"):
        self.agent = chat_client.as_agent(
            instructions=(
                "You're a creative marketing strategist. Craft compelling value propositions and target messaging"
                " aligned to the prompt."
            ),
            name=id,
        )
        super().__init__(id=id)

    @handler
    async def run(self, request: AgentExecutorRequest, ctx: WorkflowContext[AgentExecutorResponse]) -> None:
        response = await self.agent.run(request.messages)
        full_conversation = list(request.messages) + list(response.messages)
        await ctx.send_message(AgentExecutorResponse(self.id, response, full_conversation=full_conversation))

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

chat_client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

researcher = ResearcherExec(chat_client)
marketer = MarketerExec(chat_client)
legal = LegalExec(chat_client)

workflow = ConcurrentBuilder(participants=[researcher, marketer, legal]).build()

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

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

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

from agent_framework import AgentExecutorResponse

# Create a summarizer agent for the aggregator
summarizer_agent = chat_client.as_agent(
    instructions=(
        "You are a helpful assistant that consolidates multiple domain expert outputs "
        "into one cohesive, concise summary with clear takeaways. Keep it under 200 words."
    ),
    name="summarizer",
)

# Define a custom aggregator callback
async def summarize_results(results: list[AgentExecutorResponse]) -> str:
    # Extract one final assistant message per agent
    expert_sections: list[str] = []
    for r in results:
        try:
            messages = getattr(r.agent_response, "messages", [])
            final_text = messages[-1].text if messages and hasattr(messages[-1], "text") else "(no content)"
            expert_sections.append(f"{r.executor_id}:\n{final_text}")
        except Exception as e:
            expert_sections.append(f"{r.executor_id}: (error: {type(e).__name__}: {e})")

    # Ask the model to synthesize a concise summary of the experts' outputs
    prompt = "\n\n".join(expert_sections)
    response = await summarizer_agent.run(prompt)
    # Return the model's final assistant text as the completion result
    return response.messages[-1].text if response.messages else ""

Создание рабочего процесса с помощью настраиваемого агрегатора

workflow = (
    ConcurrentBuilder(participants=[researcher, marketer, legal])
    .with_aggregator(summarize_results)
    .build()
)

output = None
async for event in workflow.run("We are launching a new budget-friendly electric bike for urban commuters.", stream=True):
    if event.type == "output":
        output = event.data

if output:
    print("===== Final Consolidated Output =====")
    print(output)

Пример выходных данных с пользовательским агрегатором

===== Final Consolidated Output =====
Urban e-bike demand is rising rapidly due to eco-awareness, urban congestion, and high fuel costs,
with market growth projected at a ~10% CAGR through 2030. Key customer concerns are affordability,
easy maintenance, convenient charging, compact design, and theft protection. Differentiation opportunities
include integrating smart features (GPS, app connectivity), offering subscription or leasing options, and
developing portable, space-saving designs. Partnering with local governments and bike shops can boost visibility.

Risks include price wars eroding margins, regulatory hurdles, battery quality concerns, and heightened expectations
for after-sales support. Accurate, substantiated product claims and transparent marketing (with range disclaimers)
are essential. All e-bikes must comply with local and federal regulations on speed, wattage, safety certification,
and labeling. Clear warranty, safety instructions (especially regarding batteries), and inclusive, accessible
marketing are required. For connected features, data privacy policies and user consents are mandatory.

Effective messaging should target young professionals, students, eco-conscious commuters, and first-time buyers,
emphasizing affordability, convenience, and sustainability. Slogan suggestion: "Charge Ahead—City Commutes Made
Affordable." Legal review in each target market, compliance vetting, and robust customer support policies are
critical before launch.

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

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

workflow = ConcurrentBuilder(
    participants=[researcher, marketer, legal],
    intermediate_output_from=[researcher, marketer, legal],
).build()

Эти события можно обрабатывать в режиме потоковой передачи в режиме реального времени:

from agent_framework import AgentResponseUpdate

# Track the last author to format streaming output.
last_author: str | None = None

async for event in workflow.run("Analyze our new product launch strategy.", stream=True):
    if event.type == "intermediate" and isinstance(event.data, AgentResponseUpdate):
        update = event.data
        author = update.author_name
        if author != last_author:
            if last_author is not None:
                print()  # Newline between different authors
            print(f"{author}: {update.text}", end="", flush=True)
            last_author = author
        else:
            print(update.text, end="", flush=True)

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

Параллельное выполнение. Все агенты работают над задачей одновременно и независимо
Выходные данные AgentResponse: агрегатор по умолчанию выдает один AgentResponse с сообщением помощника для каждого участника (без запроса пользователя)
Разнообразные перспективы: каждый агент приносит свой уникальный опыт в одну и ту же проблему
Гибкие участники: вы можете использовать агенты напрямую или упаковать их в настраиваемые исполнители
Настраиваемая обработка: переопределите агрегатор по умолчанию, чтобы синтезировать результаты для конкретного домена.
Промежуточные результаты: передайте intermediate_output_from=[participant, ...], чтобы вывести результаты для каждого указанного участника в виде событий "intermediate", в дополнение к завершающему событию "output" агрегатора

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

Последовательная оркестрация

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-05-26

Оркестрации рабочих процессов Microsoft Agent Framework — параллельные

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

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

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

Настройка параллельной оркестрации

Запуск параллельного рабочего процесса и сбор результатов

Пример выходных данных

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

Настройка параллельной оркестрации

Запуск параллельного рабочего процесса и сбор результатов

Пример выходных данных

Расширенные возможности: пользовательские выполнители агента

Определение исполнителей пользовательских агентов

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

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

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

Создание рабочего процесса с помощью настраиваемого агрегатора

Пример выходных данных с пользовательским агрегатором

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

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

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

Обратная связь

Дополнительные ресурсы