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

Warnung

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 Orchestrator を使用してグループ チャットを構成する

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-Based Orchestrator を使用してグループ チャットを構成する

または、エージェントベースのオーケストレーターを使用してインテリジェントなスピーカーを選択することもできます。 オーケストレーターは、ツール、コンテキスト、および可観測性にアクセスできる完全な 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] Output: すべてのオーケストレーションがチャット メッセージの一覧を返す

詳細: カスタム スピーカーの選択

カスタム グループ チャット マネージャーを作成することで、カスタム マネージャー ロジックを実装できます。

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

Important

高度なシナリオで BaseGroupChatOrchestrator のカスタム実装を使用する場合は、 participant_registry、 max_rounds、 termination_conditionなど、すべてのプロパティを設定する必要があります。 max_rounds およびビルダーで設定 termination_condition は無視されます。

コンテキスト同期

このガイドの冒頭で説明したように、グループ チャット内のすべてのエージェントは完全な会話履歴を表示します。

Agent Framework のエージェントは、エージェント セッション (AgentSession) に依存してコンテキストを管理します。 グループ チャット オーケストレーションでは、エージェントは同じセッション インスタンスを共有 しません が、オーケストレーターは各エージェントのセッションが各ターンの前に完全な会話履歴と同期されるようにします。 これを実現するために、各エージェントのターンの後、オーケストレーターは他のすべてのエージェントに応答をブロードキャストし、すべての参加者が次のターンの最新のコンテキストを持っていることを確認します。

ヒント

エージェントの種類によって抽象化の実装が異なる場合があるため、AgentSessionは同じセッション インスタンスを共有しません。 同じセッション インスタンスを共有すると、各エージェントがコンテキストを処理および維持する方法に不整合が生じる可能性があります。

応答をブロードキャストした後、オーケストレーターは次の話者を決定し、選択したエージェントに要求を送信します。これで、応答を生成するための完全な会話履歴が表示されます。

グループ チャットを使用するタイミング

グループ チャット オーケストレーションは、次の場合に最適です。

• 反復的な精錬: 複数回のレビューと改善
• 共同作業による問題解決: 補完的な専門知識を持つエージェントが連携
• コンテンツの作成: ドキュメント作成のためのライターレビュー担当者ワークフロー
• マルチパースペクティブ分析: 同じ入力に対する多様な視点の取得
• 品質保証: 自動レビューと承認プロセス

次のような場合は、別の方法を検討してください。

• 厳密な順次処理が必要です (シーケンシャル オーケストレーションを使用する)
• エージェントは完全に独立して動作する必要があります (同時実行オーケストレーションを使用する)
• エージェント間での直接引き継ぎが必要です (ハンドオフ オーケストレーションを使用)
• 複雑な動的計画が必要です (マゼンティック オーケストレーションを使用する)

次のステップ

マゼンティック オーケストレーション