Microsoft Agent Framework 工作流程編排 - Magentic

C# 尚未支援 Magentic Orchestration。

Magentic 編排是基於 AutoGen 發明的 Magentic-One 系統設計的。它是一種靈活、通用的多代理模式，專為需要動態協作的複雜、開放式任務而設計。在這種模式中，專門的 Magentic 經理協調一個專業代理團隊，根據不斷變化的上下文、任務進度和代理能力選擇下一步應該採取行動的代理。

Magentic 管理器維護共享上下文、跟踪進度並實時調整工作流程。這使得系統能夠分解複雜的問題、委派子任務，並透過代理協作迭代完善解決方案。這種協調特別適合事先未知解路徑，且可能需要多輪推理、研究與計算的情況。

磁性編排

小提示

Magentic 編排架構與群組聊天編排模式相同，擁有一個非常強大的管理者，利用規劃來協調代理間的協作。如果你的情境需要更簡單的協調，且不需要複雜的規劃，可以考慮改用群組聊天模式。

備註

在 Magentic-One 論文中，有 4 種高度專業化的代理被設計用來解決一組非常特定的任務。在 Agent Framework 的 Magentic orchestration 中，你可以定義自己的專用代理，以符合你特定的應用需求。然而，Magentic 編曲在原始 Magentic-One 設計之外的表現尚未經過測試。

您將學到的內容

如何設定 Magentic 管理程式以協調多個專業代理程式
如何處理串流事件 WorkflowOutputEvent
如何實施人機迴路計劃審查
如何追蹤客服專員協作和複雜任務的進度

定義您的專業代理

在 Magentic 協調流程中，您可以定義經理可以根據任務需求動態選取的專用代理程式：

from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient

responses_client = OpenAIResponsesClient()

researcher_agent = Agent(
    name="ResearcherAgent",
    description="Specialist in research and information gathering",
    instructions=(
        "You are a Researcher. You find information without additional computation or quantitative analysis."
    ),
    # This agent requires the gpt-4o-search-preview model to perform web searches
    chat_client=OpenAIChatClient(model_id="gpt-4o-search-preview"),
)

coder_agent = Agent(
    name="CoderAgent",
    description="A helpful assistant that writes and executes code to process and analyze data.",
    instructions="You solve questions using code. Please provide detailed analysis and computation process.",
    chat_client=responses_client,
    tools=responses_client.get_code_interpreter_tool(),
)

# Create a manager agent for orchestration
manager_agent = Agent(
    name="MagenticManager",
    description="Orchestrator that coordinates the research and coding workflow",
    instructions="You coordinate a team to complete complex tasks efficiently.",
    chat_client=OpenAIChatClient(),
)

建立 Magentic 工作流程

用MagenticBuilder標準管理器配置工作流程（）：StandardMagenticManager

from agent_framework.orchestrations import MagenticBuilder

workflow = MagenticBuilder(
    participants=[researcher_agent, coder_agent],
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=3,
    max_reset_count=2,
).build()

小提示

基於 Magnetic-One 設計實作一個標準管理器，並採用原始論文中的固定提示。你可以通過在建構函數參數中傳入自己的提示 MagenticBuilder 來自訂管理器的行為。為了進一步自訂管理器，你也可以透過子 MagenticManagerBase 類別來實作自己的管理器。

用事件串流執行工作流程

執行複雜任務並處理串流輸出與編排更新的事件：

import json
import asyncio
from typing import cast

from agent_framework import (
    AgentResponseUpdate,
    Message,
    WorkflowEvent,
)
from agent_framework.orchestrations import MagenticProgressLedger

task = (
    "I am preparing a report on the energy efficiency of different machine learning model architectures. "
    "Compare the estimated training and inference energy consumption of ResNet-50, BERT-base, and GPT-2 "
    "on standard datasets (for example, ImageNet for ResNet, GLUE for BERT, WebText for GPT-2). "
    "Then, estimate the CO2 emissions associated with each, assuming training on an Azure Standard_NC6s_v3 "
    "VM for 24 hours. Provide tables for clarity, and recommend the most energy-efficient model "
    "per task type (image classification, text classification, and text generation)."
)

# Keep track of the last executor to format output nicely in streaming mode
last_message_id: str | None = None
output_event: WorkflowEvent | None = None
async for event in workflow.run_stream(task):
    if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
        message_id = event.data.message_id
        if message_id != last_message_id:
            if last_message_id is not None:
                print("\n")
            print(f"- {event.executor_id}:", end=" ", flush=True)
            last_message_id = message_id
        print(event.data, end="", flush=True)

    elif event.type == "magentic_orchestrator":
        print(f"\n[Magentic Orchestrator Event] Type: {event.data.event_type.name}")
        if isinstance(event.data.content, MagenticProgressLedger):
            print(f"Please review progress ledger:\n{json.dumps(event.data.content.to_dict(), indent=2)}")
        else:
            print(f"Unknown data type in MagenticOrchestratorEvent: {type(event.data.content)}")

        # Block to allow user to read the plan/progress before continuing
        # Note: this is for demonstration only and is not the recommended way to handle human interaction.
        # Please refer to `with_plan_review` for proper human interaction during planning phases.
        await asyncio.get_event_loop().run_in_executor(None, input, "Press Enter to continue...")

    elif event.type == "output":
        output_event = event

# The output of the Magentic workflow is a list of ChatMessages with only one final message
# generated by the orchestrator.
output_messages = cast(list[Message], output_event.data)
output = output_messages[-1].text
print(output)

進階：人為介入規劃審查

啟用人員介入（HITL）功能，讓使用者能在計畫執行前審查並核准經理提出的計畫。這有助於確保計畫符合使用者的期望與需求。

圖則審查有兩種選項：

修訂：使用者可以提供回饋以修訂計畫，這會觸發管理者根據回饋重新規劃。
批准：使用者可保持原樣批准計畫，讓工作流程繼續進行。

在建置 Magentic 工作流程時，傳遞 enable_plan_review=True 以啟用計畫審查：

from agent_framework import (
    AgentResponseUpdate,
    Agent,
    Message,
    MagenticPlanReviewRequest,
    WorkflowEvent,
)
from agent_framework.orchestrations import MagenticBuilder

workflow = MagenticBuilder(
    participants=[researcher_agent, analyst_agent],
    enable_plan_review=True,
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=1,
    max_reset_count=2,
).build()

計畫審查請求會以WorkflowEvent搭配type="request_info"和MagenticPlanReviewRequest資料的形式發出。你可以在事件串流中處理這些請求：

小提示

想了解更多關於請求與回應的資訊，請參閱「請求與回應」指南。

pending_request: WorkflowEvent | None = None
pending_responses: dict[str, MagenticPlanReviewResponse] | None = None
output_event: WorkflowEvent | None = None

while not output_event:
    if pending_responses is not None:
        stream = workflow.run(responses=pending_responses)
    else:
        stream = workflow.run_stream(task)

    last_message_id: str | None = None
    async for event in stream:
        if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
            message_id = event.data.message_id
            if message_id != last_message_id:
                if last_message_id is not None:
                    print("\n")
                print(f"- {event.executor_id}:", end=" ", flush=True)
                last_message_id = message_id
            print(event.data, end="", flush=True)

        elif event.type == "request_info" and event.request_type is MagenticPlanReviewRequest:
            pending_request = event

        elif event.type == "output":
            output_event = event

    pending_responses = None

    # Handle plan review request if any
    if pending_request is not None:
        event_data = cast(MagenticPlanReviewRequest, pending_request.data)

        print("\n\n[Magentic Plan Review Request]")
        if event_data.current_progress is not None:
            print("Current Progress Ledger:")
            print(json.dumps(event_data.current_progress.to_dict(), indent=2))
            print()
        print(f"Proposed Plan:\n{event_data.plan.text}\n")
        print("Please provide your feedback (press Enter to approve):")

        reply = await asyncio.get_event_loop().run_in_executor(None, input, "> ")
        if reply.strip() == "":
            print("Plan approved.\n")
            pending_responses = {pending_request.request_id: event_data.approve()}
        else:
            print("Plan revised by human.\n")
            pending_responses = {pending_request.request_id: event_data.revise(reply)}
        pending_request = None

關鍵概念

動態協調：Magentic 經理根據不斷變化的上下文動態選擇下一步應該行動的代理
迭代細化：系統可以分解複雜問題，並通過多輪迭代細化解決方案
進度追蹤：內建機制來偵測失速並在需要時重置計劃
靈活協作：可以按照經理確定的任何順序多次呼叫代理
人類監督：可選的人工參與計畫審查機制

工作流程執行流程

Magentic 協調流程遵循下列執行模式：

規劃階段：經理分析任務並建立初始計劃
可選計畫審查：若啟用，人類可審查並批准/修改計畫
代理選擇：經理為每個子任務選擇最合適的代理
執行：所選代理執行其任務部分
進度評估：經理評估進度並更新計劃
停滯偵測：若進度停滯，請自動重新規劃，並可選擇加入人工審查流程。
迭代：步驟3至6重複，直到任務完成或達到極限
最終綜合：經理將所有代理輸出綜合成最終結果

完整範例

完整範例請參閱代理框架範例庫。

後續步驟

交接協調流程

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-02-13