Sdílet prostřednictvím

Orchestrace pracovních postupů Microsoft Agent Framework – Magentic

Magentic Orchestration se v jazyce C# zatím nepodporuje.

Magnetická orchestrace je navržena na základě systému Magentic-One vynalezeného AutoGen. Jedná se o flexibilní vzor s více agenty pro obecné účely navržený pro složité otevřené úlohy, které vyžadují dynamickou spolupráci. V tomto modelu vyhrazený magentický manažer koordinuje tým specializovaných agentů a vybere, který agent by měl pokračovat na základě vyvíjejícího se kontextu, průběhu úkolů a možností agenta.

Magentický manažer udržuje sdílený kontext, sleduje průběh a přizpůsobuje pracovní postup v reálném čase. Díky tomu může systém rozdělit složité problémy, delegovat dílčí úkoly a iterativně zpřesňovat řešení prostřednictvím spolupráce agentů. Orchestrace je obzvláště vhodná pro scénáře, ve kterých není cesta řešení předem známá a může vyžadovat několik cyklů odůvodnění, výzkumu a výpočtů.

Magentická orchestrace

Návod

Magentická orchestrace má stejnou architekturu jako model orchestrace skupinového chatu s velmi výkonným manažerem, který používá plánování ke koordinaci spolupráce agentů. Pokud váš scénář vyžaduje jednodušší koordinaci bez složitého plánování, zvažte místo toho použití vzoru Skupinový chat.

Poznámka:

V dokumentu Magentic-One jsou 4 vysoce specializovaní agenti navrženi tak, aby vyřešili velmi specifickou sadu úloh. V magentické orchestraci v rozhraní Agent Framework můžete definovat vlastní specializované agenty tak, aby vyhovovaly vašim konkrétním potřebám aplikace. Není však otestováno, jak dobře magentická orchestrace bude fungovat mimo původní Magentic-One návrh.

Co se naučíte

  • Nastavení magentického manažera pro koordinaci více specializovaných agentů
  • Zpracování streamovaných událostí pomocí WorkflowOutputEvent
  • Postup implementace kontroly plánu s lidským prvkem v rozhodování
  • Jak sledovat spolupráci agentů a jejich pokrok při řešení složitých úkolů

Definování specializovaných agentů

V magentické orchestraci definujete specializované agenty, které může správce dynamicky vybírat na základě požadavků na úlohy:

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

Sestavení magentického pracovního postupu

Slouží MagenticBuilder ke konfiguraci pracovního postupu pomocí standardního manažera(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()

Návod

Standardní manažer je implementován na základě návrhu MagneticOne s předem danými pokyny převzatými z původní studie. Chování manažera můžete přizpůsobit předáním vlastních výzev prostřednictvím parametrů konstruktoru MagenticBuilder . Pokud chcete manažera dále přizpůsobit, můžete také implementovat vlastního manažera tím, že subclassujete MagenticManagerBase třídu.

Spuštění pracovního postupu se streamováním událostí

Spusťte komplexní úlohu a zpracujte události pro aktualizace výstupu a orchestrace streamování:

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)

Pokročilé: Přezkum plánu s člověkem v procesu

Povolte funkci HITL (human-in-the-loop), aby uživatelé mohli před provedením zkontrolovat a schválit navrhovaný plán vedoucího. To je užitečné pro zajištění toho, aby plán odpovídal očekáváním a požadavkům uživatelů.

Existují dvě možnosti kontroly plánu:

  1. Revidovat: Uživatel může poskytnout zpětnou vazbu k revizi plánu, což aktivuje správu, která bude na základě zpětné vazby znovu plánovat.
  2. Schválit: Uživatel může schválit plán as-is, což pracovnímu postupu umožní pokračovat.

Povolte kontrolu plánu pomocí enable_plan_review=True při vytváření pracovního postupu Magentic:

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

Žádosti o kontrolu plánu jsou vydány jako WorkflowEvent s type="request_info" a MagenticPlanReviewRequest daty. Tyto požadavky můžete zpracovat ve streamu událostí:

Návod

Další informace o požadavcích a odpovědích najdete v průvodci Požadavky a odpovědi .

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

Klíčové koncepty

  • Dynamická koordinace: Magentický manažer dynamicky vybírá, který agent by měl fungovat dále na základě vyvíjejícího se kontextu.
  • Iterativní zpřesnění: Systém může rozdělit složité problémy a iterativní upřesňovat řešení prostřednictvím několika kruhů.
  • Sledování průběhu: Integrované mechanismy pro detekci zastavení a resetování plánu v případě potřeby
  • Flexibilní spolupráce: Agenti se dají volat vícekrát v libovolném pořadí podle toho, co určuje manažer.
  • Lidský dohled: Volitelné mechanismy human-in-the-loop pro kontrolu plánu

Tok provádění pracovního postupu

Magentická orchestrace se řídí tímto vzorem provádění:

  1. Fáze plánování: Manažer analyzuje úkol a vytvoří počáteční plán.
  2. Volitelná kontrola plánu: Pokud je tato možnost povolená, můžou lidé plán zkontrolovat a schválit nebo upravit.
  3. Výběr agenta: Správce vybere nejvhodnějšího agenta pro každý dílčí úkol.
  4. Spuštění: Vybraný agent spustí svou část úlohy.
  5. Hodnocení průběhu: Vedoucí vyhodnocuje průběh a aktualizuje plán.
  6. Detekce zastavení: Pokud dojde k zastavení průběhu, automaticky přeplánovat s volitelným procesem kontrolovaným člověkem
  7. Iterace: Kroky 3 až 6 se opakují, dokud není úkol dokončen nebo nejsou dosaženy limity.
  8. Konečná syntéza: Manažer syntetizuje všechny výstupy agenta do konečného výsledku.

Kompletní příklad

Prohlédnou si kompletní ukázky v úložišti ukázek architektury agentů.

Další kroky

Orchestrace předání

  • Last updated on