Magentic - Orkestreringar för arbetsflöden i Microsoft Agent Framework

Magentisk orkestrering är utformad baserat på Magentic-One-systemet som uppfunnits av AutoGen. Det är ett flexibelt, generellt mönster för flera agenter som är utformat för komplexa, öppna uppgifter som kräver dynamiskt samarbete. I det här mönstret samordnar en dedikerad Magentic Manager ett team med specialiserade agenter och väljer vilken agent som ska agera härnäst baserat på den föränderliga kontexten, aktivitetsstatusen och agentfunktionerna.

Magentic Manager har en delad kontext, spårar förloppet och anpassar arbetsflödet i realtid. På så sätt kan systemet dela upp komplexa problem, delegera underaktiviteter och iterativt förfina lösningar genom agentsamarbete. Orkestreringen passar särskilt bra för scenarier där lösningsvägen inte är känd i förväg och kan kräva flera omgångar av resonemang, forskning och beräkning.

Magentisk orkestrering

Tip

Den magentiska orkestreringen har samma arkitektur som mönstret för gruppchattorkestrering , med en mycket kraftfull chef som använder planering för att samordna agentsamarbete. Om ditt scenario kräver enklare samordning utan komplex planering kan du överväga att använda mönstret Gruppchatt i stället.

Anmärkning

I dokumentet Magentic-One är 4 högspecialiserade agenter utformade för att lösa en mycket specifik uppsättning uppgifter. I den magentiska orkestreringen i Agent Framework kan du definiera dina egna specialiserade agenter så att de passar dina specifika programbehov. Det är dock oprövat hur väl Magentic-orkestreringen kommer att fungera utanför den ursprungliga Magentic-One-designen.

Vad du ska lära dig

  • Så här konfigurerar du en magentisk chef för att samordna flera specialiserade agenter
  • Hantera strömmande händelser med WorkflowEvent
  • Så här implementerar du en human-in-the-loop-översyn av planer
  • Så här spårar du agentsamarbete och framsteg genom komplexa uppgifter

Definiera dina specialiserade agenter

I Magentisk orkestrering definierar du specialiserade agenter som chefen dynamiskt kan välja baserat på uppgiftskrav:

#pragma warning disable MAAIW001  // Magentic types are experimental
#pragma warning disable OPENAI001 // HostedCodeInterpreterTool is experimental

using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Workflows;
using Microsoft.Agents.AI.Workflows.Specialized.Magentic;
using Microsoft.Extensions.AI;

string endpoint = Environment.GetEnvironmentVariable("AZURE_AI_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("AZURE_AI_PROJECT_ENDPOINT is not set.");
string deploymentName = Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME") ?? "gpt-5.4-mini";

AIProjectClient projectClient = new(new Uri(endpoint), new DefaultAzureCredential());

AIAgent researcherAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "ResearcherAgent",
    description: "Specialist in research and information gathering.",
    instructions: "You are a researcher. Find relevant information without doing additional computation or quantitative analysis.");

AIAgent coderAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "CoderAgent",
    description: "A helpful assistant that writes and executes code to analyze data.",
    instructions: "You solve quantitative questions by writing and running code. Show the analysis and the computation process clearly.",
    tools: [new HostedCodeInterpreterTool()]);

AIAgent managerAgent = projectClient.AsAIAgent(
    deploymentName,
    name: "MagenticManager",
    description: "Orchestrator that coordinates the research and coding workflow.",
    instructions: "You coordinate the team to complete complex tasks efficiently.");

import os

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

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

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

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

Skapa det magentiska arbetsflödet

Använd Magentic-arbetsflödesbyggaren för att konfigurera arbetsflödet med en chef och ett antal deltagare. Byggverktyget exponerar också gränser för den inre loopen (maximalt antal samordningsrundor, maximalt antal på varandra följande stopp innan omplanering, maximalt antal återställningar av planen) samt en flagga för mänsklig granskning av planen.

Workflow workflow = new MagenticWorkflowBuilder(managerAgent)
    .AddParticipants([researcherAgent, coderAgent])
    .WithName("Magentic Orchestration Workflow")
    .WithDescription("Coordinates a researcher and coder to solve a complex analytical task.")
    .RequirePlanSignoff(false)
    .WithMaxRounds(10)
    .WithMaxStalls(3)
    .WithMaxResets(2)
    .Build();

from agent_framework.orchestrations import MagenticBuilder

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

Tip

En standardhanterare implementeras baserat på Magentic-One design, med fasta uppmaningar från det ursprungliga papperet. Du kan anpassa chefens beteende genom att skicka in dina egna frågor via MagenticBuilder konstruktorparametrarna. Om du vill anpassa chefen ytterligare kan du även implementera din egen chef genom att underklassa MagenticManagerBase klassen.

Mellanliggande utdata

Anmärkning

Det här avsnittet gäller för närvarande endast för Python pivot.

Att skicka intermediate_output_from=[...] till MagenticBuilder innebär att specifika deltagare anges som källor för mellanliggande utdata. Deras yield_output anrop genererar "intermediate" händelser, medan chefens slutliga syntetiserade svar förblir en "output" (terminal) händelse. Utan den här parametern (standardinställningen) visas endast hanterarens terminalfönster AgentResponse.

Detta är särskilt användbart för magentiska arbetsflöden eftersom:

  • Uppgifter är ofta tidskrävande med många omgångar av agentsamarbete.
  • Du kan visa varje agent bidrag i realtid när arbetsflödet fortskrider i strömningsläge
  • Det ger insyn i de mellanliggande resonemangsstegen i arbetsflödet

Kör arbetsflödet med händelseströmning

Kör en komplex uppgift och hantera händelser för strömmande utdata och orkestreringsuppdateringar. Terminalarbetsflödets utdata innehåller chefens syntetiserade slutliga svar.

const string TaskPrompt =
    "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).";

await using StreamingRun run = await InProcessExecution.RunStreamingAsync(
    workflow,
    new List<ChatMessage> { new(ChatRole.User, TaskPrompt) });

await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

string? lastResponseId = null;
WorkflowOutputEvent? finalOutput = null;

await foreach (WorkflowEvent workflowEvent in run.WatchStreamAsync())
{
    switch (workflowEvent)
    {
        case AgentResponseUpdateEvent updateEvent:
            // Stream per-participant deltas. Group by ResponseId / MessageId / ExecutorId so
            // each new contiguous response prints its executor header once.
            string responseId = updateEvent.Update.ResponseId
                ?? updateEvent.Update.MessageId
                ?? updateEvent.ExecutorId;
            if (!string.Equals(responseId, lastResponseId, StringComparison.Ordinal))
            {
                if (lastResponseId is not null)
                {
                    Console.WriteLine();
                }
                Console.Write($"- {updateEvent.ExecutorId}: ");
                lastResponseId = responseId;
            }
            Console.Write(updateEvent.Update.Text);
            break;

        case MagenticPlanCreatedEvent planCreated:
            Console.WriteLine($"\n[Magentic Initial Plan]\n{planCreated.FullTaskLedger.Text}");
            break;

        case MagenticReplannedEvent replanned:
            Console.WriteLine($"\n[Magentic Replanned]\n{replanned.FullTaskLedger.Text}");
            break;

        case MagenticProgressLedgerUpdatedEvent progressUpdated:
            MagenticProgressLedger ledger = progressUpdated.ProgressLedger;
            Console.WriteLine(
                $"\n[Magentic Progress Ledger] satisfied={ledger.IsRequestSatisfied}, " +
                $"inLoop={ledger.IsInLoop}, progressing={ledger.IsProgressBeingMade}, " +
                $"nextSpeaker={ledger.NextSpeaker}, instruction={ledger.InstructionOrQuestion}");
            break;

        case WorkflowOutputEvent outputEvent when outputEvent.Is<List<ChatMessage>>():
            finalOutput = outputEvent;
            break;

        case WorkflowErrorEvent workflowError:
            Console.Error.WriteLine(workflowError.Exception?.ToString() ?? "Unknown workflow error.");
            break;

        case ExecutorFailedEvent executorFailed:
            Console.Error.WriteLine(
                $"Executor '{executorFailed.ExecutorId}' failed: " +
                (executorFailed.Data?.ToString() ?? "unknown error"));
            break;
    }
}

if (finalOutput?.As<List<ChatMessage>>() is { } transcript)
{
    Console.WriteLine("\n\n=== Final Conversation Transcript ===\n");
    foreach (ChatMessage message in transcript)
    {
        Console.WriteLine($"{message.AuthorName ?? message.Role.ToString()}: {message.Text}");
    }
}

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
stream = workflow.run(task, stream=True)
async for event in stream:
    if event.type in ("intermediate", "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, Message):
            print(f"Please review the plan:\n{event.data.content.text}")
        elif 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...")

result = await stream.get_final_response()
if outputs := result.get_outputs():
    print(outputs[-1])

Magentic lyfter fram tre orkestreringshändelser som markerar milstolpar för planering och framsteg:

  • Den första planen skapades – chefen har tagit fram den första uppgiftsplanen.
  • Omplanerad – en ny plan skapades, antingen på grund av detektering av stopp eller för att en person reviderade planen genom plangranskning.
  • Framstegslogg uppdaterad – genereras en gång per samordningsrunda; innehåller den aktuella framstegsloggen (om förfrågan har uppfyllts, om teamet har fastnat i en loop, om framsteg görs, vem som ska tala härnäst och instruktionen som ska skickas till den personen).

I Python ryms dessa i en enda MagenticOrchestratorEvent vars event_type enum särskiljer PLAN_CREATED, REPLANNED och PROGRESS_LEDGER_UPDATED. I .NET genereras de som tre olika typer – MagenticPlanCreatedEvent, MagenticReplannedEvent och MagenticProgressLedgerUpdatedEvent – som alla härleds från MagenticOrchestratorEvent.

Avancerat: Granskning av planering för Människan i loopen

Aktivera HITL (Human-in-the-loop) så att användare kan granska och godkänna den föreslagna planen av chefen före genomförande. Detta är användbart för att säkerställa att planen överensstämmer med användarens förväntningar och krav.

Det finns två alternativ för plangranskning:

  1. Revidera: Användaren ger feedback för att revidera planen, vilket utlöser att chefen planerar om baserat på feedbacken.
  2. Godkänn: Användaren godkänner planen i sin nuvarande form, så att arbetsflödet kan fortsätta.

Aktivera plangranskning när du skapar det magentiska arbetsflödet. Standardinställningarna skiljer sig mellan språk: i Python är plangranskningen off som standard (enable_plan_review=False) och du väljer uttryckligen. I .NET är plangranskningen on som standard (RequirePlanSignoff standardvärdet true), och det grundläggande exemplet tidigare på den här sidan valde bort så att det kunde köras från slutpunkt till slutpunkt utan interaktion. Koden nedan visar hur du anmäler dig och hanterar de resulterande granskningsbegäranden.

Pauser i plangranskningen görs synliga via arbetsflödets mekanism för begäran och svar med data i MagenticPlanReviewRequest. Du hanterar dessa i händelseströmmen och återupptar arbetsflödet med en MagenticPlanReviewResponse när människan har godkänt eller reviderat planen.

Tip

Läs mer om begäranden och svar i guiden Begäranden och svar .

Workflow workflow = new MagenticWorkflowBuilder(managerAgent)
    .AddParticipants([researcherAgent, coderAgent])
    .RequirePlanSignoff(true)
    .WithMaxRounds(10)
    .WithMaxStalls(1)
    .WithMaxResets(2)
    .Build();

CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
InProcessExecutionEnvironment environment = ExecutionEnvironment.InProcess_Lockstep
    .ToWorkflowExecutionEnvironment()
    .WithCheckpointing(checkpointManager);

await using StreamingRun run = await environment.OpenStreamingAsync(workflow);
await run.TrySendMessageAsync(new List<ChatMessage> { new(ChatRole.User, TaskPrompt) });
await run.TrySendMessageAsync(new TurnToken(emitEvents: true));

ExternalRequest? pendingRequest = null;
CheckpointInfo? lastCheckpoint = null;
WorkflowOutputEvent? finalOutput = null;

async Task<WorkflowOutputEvent?> DrainAsync(StreamingRun activeRun)
{
    WorkflowOutputEvent? output = null;
    await foreach (WorkflowEvent evt in activeRun.WatchStreamAsync(blockOnPendingRequest: false))
    {
        switch (evt)
        {
            case AgentResponseUpdateEvent updateEvent:
                Console.Write(updateEvent.Update.Text);
                break;
            case RequestInfoEvent requestInfo
                when requestInfo.Request.Data.As<MagenticPlanReviewRequest>() is not null:
                pendingRequest = requestInfo.Request;
                break;
            case SuperStepCompletedEvent stepCompleted:
                lastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint ?? lastCheckpoint;
                break;
            case WorkflowOutputEvent outputEvent when outputEvent.Is<List<ChatMessage>>():
                output = outputEvent;
                break;
        }
    }
    return output;
}

finalOutput = await DrainAsync(run);

// Loop until the workflow finishes or the user accepts a plan that runs to completion.
while (finalOutput is null && pendingRequest is not null)
{
    MagenticPlanReviewRequest reviewRequest = pendingRequest.Data.As<MagenticPlanReviewRequest>()!;

    Console.WriteLine("\n\n[Magentic Plan Review Request]");
    if (reviewRequest.CurrentProgress is { } progress)
    {
        Console.WriteLine(
            $"Current progress: satisfied={progress.IsRequestSatisfied}, " +
            $"inLoop={progress.IsInLoop}, progressing={progress.IsProgressBeingMade}");
    }
    if (reviewRequest.IsStalled)
    {
        Console.WriteLine("(Replan triggered by stall detection.)");
    }
    Console.WriteLine($"Proposed plan:\n{reviewRequest.Plan.Text}\n");
    Console.Write("Press Enter to approve, or type feedback to request a revision: ");

    string reply = Console.ReadLine() ?? string.Empty;
    MagenticPlanReviewResponse reviewResponse = string.IsNullOrWhiteSpace(reply)
        ? reviewRequest.Approve()
        : reviewRequest.Revise(reply);

    ExternalResponse response = pendingRequest.CreateResponse(reviewResponse);
    pendingRequest = null;

    await using StreamingRun resumed = await environment.ResumeStreamingAsync(workflow, lastCheckpoint!);
    await resumed.SendResponseAsync(response);
    finalOutput = await DrainAsync(resumed);
}

if (finalOutput?.As<List<ChatMessage>>() is { } transcript)
{
    Console.WriteLine("\n\n=== Final Conversation Transcript ===\n");
    foreach (ChatMessage message in transcript)
    {
        Console.WriteLine($"{message.AuthorName ?? message.Role.ToString()}: {message.Text}");
    }
}

import json
import asyncio
from typing import cast

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

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

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

while not final_response:
    if pending_responses is not None:
        stream = workflow.run(stream=True, responses=pending_responses)
    else:
        stream = workflow.run(task, stream=True)

    last_message_id: str | None = None
    async for event in stream:
        if event.type in ("intermediate", "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

    result = await stream.get_final_response()
    if outputs := result.get_outputs():
        final_response = outputs[-1]

    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

En MagenticPlanReviewRequest innehåller den föreslagna planen, den aktuella förloppsloggen (null / None tom vid den första granskningen och ifylld vid omplaneringar som utlöses av stallidentifiering) samt en flagga som anger om omplaneringen utlöstes av stallidentifiering. Bygg svaret genom att anropa antingen approve() för att acceptera planen utan ändringar eller revise(...) med återkoppling för att be chefen att omplanera.

Viktiga begrepp

  • Dynamisk samordning: Den magentiska chefen väljer dynamiskt vilken agent som ska agera härnäst baserat på den föränderliga kontexten.
  • Terminal Output: Terminalarbetsflödets utdata bär chefens syntetiserade slutliga svar (en AgentResponse i Python; en WorkflowOutputEvent med en List<ChatMessage> nyttolast i .NET).
  • Orchestrator-händelser: planskapade, omplanerade och i förloppsloggen uppdaterade milstolpar görs tillgängliga via MagenticOrchestratorEvent (en händelse med en event_type enum i Python; tre härledda typer i .NET). Streamingdelta per deltagare levereras via ramverkets standardhändelser för uppdateringar av agentsvar.
  • Iterativ förfining: Systemet kan bryta ned komplexa problem och iterativt förfina lösningar genom flera rundor.
  • Förloppsuppföljning och upptäckt av avstannande: Förloppsloggen spårar om begäran har uppfyllts, om teamet har fastnat i en loop och om framsteg görs. På varandra följande rundor utan framsteg ökar en stallräknare, och när det konfigurerade maxvärdet överskrids utlöses en automatisk återställning och omplanering.
  • Flexibelt samarbete: Agenter kan anropas flera gånger i valfri ordning enligt chefens beslut.
  • Mänsklig övervakning: Valfri granskning av planer med en människa i loopen via MagenticPlanReviewRequest / MagenticPlanReviewResponse.
  • Mellanliggande utdata (endast för Python för tillfället): Ange deltagare vars yield_output-anrop ska visas som "intermediate"-händelser tillsammans med hanterarens terminalutdata.

Arbetsflödeskörning

Den magnetiska orkestreringen följer detta utförandemönster:

  1. Planeringsfas: Chefen analyserar uppgiften och skapar en första plan
  2. Valfri plangranskning: Om det är aktiverat kan människor granska och godkänna/ändra planen
  3. Val av agent: Chefen väljer den lämpligaste agenten för varje underavdelning
  4. Körning: Den valda agenten kör sin del av uppgiften
  5. Förloppsbedömning: Chefen utvärderar förloppet och uppdaterar planen
  6. Stallidentifiering: Om förloppet stannar ska du automatiskt planera om med en valfri mänsklig granskningsprocess
  7. Iteration: Steg 3–6 upprepas tills aktiviteten har slutförts eller gränserna har nåtts
  8. Slutlig syntes: Chefen syntetiserar alla agentutdata till ett slutligt resultat

Fullständigt exempel

Se fullständiga exempel i arkivet för Agent Framework-exempel.

Se fullständiga exempel i arkivet för Agent Framework-exempel.

Anmärkning

Go-stöd för den här funktionen kommer snart. Se Agent Framework Go-lagringsplatsen för den senaste statusen.

Nästa steg

Överlämningsorkestrering

  • Last updated on