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 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：提供選擇決策的對話狀態
• 迭代協作：代理建立在彼此的貢獻之上
• 事件串流：即時處理WorkflowOutputEventAgentResponseUpdate資料
• list[訊息] 輸出：所有編排都會回傳聊天訊息清單

進階：自訂揚聲器選擇

您可以透過建立自訂群組聊天管理員來實作自訂管理員邏輯：

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 。 共用同一個會話實例可能導致各代理處理與維護上下文的方式不一致。

廣播回應後，編排器決定下一位講者，並將請求發送給選定的代理，該代理此時擁有完整的對話歷史以產生回應。

何時使用群聊

群聊編排非常適合：

• 反覆精煉：多輪審查和改進
• 協作解決問題：具有互補專業知識的代理商協同工作
• 內容創建： 用於文檔創建的作家-審閱者工作流程
• 多視角分析：對同一輸入獲得不同的觀點
• 質量保證：自動化審查和批准流程

在以下情況下考慮替代方案：

• 您需要嚴格的循序處理 （使用循序協調流程）
• 代理程式應該完全獨立運作 （使用並行協調流程）
• 需要直接代理程式對代理程式的交接 （使用交接協調流程）
• 需要複雜的動態規劃（使用 Magentic 編排）

後續步驟

Magnetic 協調