Orkiestracje przepływów pracy programu Microsoft Agent Framework — Magentic

Orkiestracja magentyczna nie jest jeszcze obsługiwana w języku C#.

Orkiestracja magentyczna została zaprojektowana na podstawie systemu Magentic-One opracowanego przez AutoGen. Jest to elastyczny wzorzec wieloaplikowy ogólnego przeznaczenia przeznaczony dla złożonych, otwartych zadań wymagających dynamicznej współpracy. W tym wzorcu dedykowany menedżer magentyczny koordynuje zespół wyspecjalizowanych agentów, wybierając agenta, który powinien działać dalej na podstawie zmieniającego się kontekstu, postępu zadań i możliwości agenta.

Menedżer Magentic utrzymuje wspólny kontekst, śledzi postęp i dostosowuje przepływ pracy w czasie rzeczywistym. Dzięki temu system może rozdzielić złożone problemy, delegować podzadania i iteracyjnie uściślić rozwiązania za pośrednictwem współpracy agentów. Aranżacja jest szczególnie odpowiednia dla scenariuszy, w których ścieżka rozwiązania nie jest znana z wyprzedzeniem i może wymagać wielu rund rozumowania, badań i obliczeń.

Tip

Aranżacja Magentic ma taką samą architekturę jak wzorzec aranżacji czatu grupowego, z wysoce zaawansowanym menedżerem, który używa planowania do koordynowania współpracy agentów. Jeśli scenariusz wymaga prostszej koordynacji bez złożonego planowania, rozważ użycie wzorca czatu grupowego.

Uwaga / Notatka

W dokumencie Magentic-One 4 wysoce wyspecjalizowanych agentów zaprojektowano w celu rozwiązania bardzo konkretnego zestawu zadań. W orkiestracji magentycznej w programie Agent Framework możesz zdefiniować własnych wyspecjalizowanych agentów, aby dopasować je do konkretnych potrzeb aplikacji. Jednakże nie zostało przetestowane, jak dobrze orkiestracja Magentic będzie działać poza oryginalnym projektem Magentic-One.

Czego nauczysz się

• Jak skonfigurować menedżera magentycznego w celu koordynowania wielu wyspecjalizowanych agentów
• Jak obsługiwać zdarzenia przesyłania strumieniowego za pomocą polecenia WorkflowOutputEvent
• Jak wdrożyć przegląd planu przy udziale człowieka
• Jak śledzić współpracę i postęp agenta za pośrednictwem złożonych zadań

Definiowanie wyspecjalizowanych agentów

W orkiestracji magentycznej definiujesz wyspecjalizowanych agentów, których menedżer może dynamicznie wybierać na podstawie wymagań dotyczących zadań:

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

Tworzenie przepływu pracy magentycznego

Służy MagenticBuilder do konfigurowania przepływu pracy za pomocą menedżera standardowego(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()

Tip

Menedżer standardowy jest implementowany na podstawie projektu Magnetic-One, ze stałymi monitami pobranymi z oryginalnego dokumentu. Zachowanie menedżera można dostosować, przekazując własne instrukcje za pomocą parametrów konstruktora MagenticBuilder. Aby jeszcze bardziej dostosować menedżera, możesz również zaimplementować własnego menedżera, klasy podrzędnej MagenticManagerBase .

Uruchamianie przepływu pracy za pomocą przesyłania strumieniowego zdarzeń

Wykonaj złożone zadanie i obsłuż zdarzenia dla strumieniowego przesyłania danych wyjściowych i aktualizacji orkiestracji.

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)

Zaawansowane: Przegląd planu z udziałem człowieka

Aby umożliwić użytkownikom przeglądanie i zatwierdzanie proponowanego planu menedżera przed wykonaniem, włącz tryb z udziałem człowieka (HITL). Jest to przydatne w celu zapewnienia, że plan jest zgodny z oczekiwaniami i wymaganiami użytkownika.

Istnieją dwie opcje przeglądu planu:

1. Korekta: użytkownik może przekazać opinię do skorygowania planu, co zainicjuje ponowne planowanie na podstawie tej opinii.
2. Zatwierdź: użytkownik może zatwierdzić plan as-is, umożliwiając kontynuowanie przepływu pracy.

Włącz przegląd planu, przekazując enable_plan_review=True podczas kompilowania przepływu pracy 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()

Żądania przeglądu planu są przesyłane jako WorkflowEvent z danymi type="request_info" i MagenticPlanReviewRequest. Te żądania można obsłużyć w strumieniu zdarzeń:

Tip

Dowiedz się więcej o żądaniach i odpowiedziach w przewodniku Żądania i odpowiedzi .

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

Kluczowe pojęcia

• Dynamiczna koordynacja: Menedżer magentyczny dynamicznie wybiera agenta, który powinien działać dalej w oparciu o zmieniający się kontekst
• Uściślenie iteracyjne: system może podzielić złożone problemy i iteracyjnie uściślić rozwiązania za pomocą wielu rund
• Śledzenie postępu: wbudowane mechanizmy wykrywania zatrzymań i resetowania planu w razie potrzeby
• Elastyczna współpraca: agenci mogą być wywoływani wiele razy w dowolnej kolejności określonej przez menedżera
• Nadzór ludzki: opcjonalne mechanizmy z udziałem człowieka na potrzeby przeglądu planu

Przepływ wykonywania zadań w przepływie pracy

Orkiestracja Magentic jest zgodna z tym wzorcem wykonania:

1. Faza planowania: Menedżer analizuje zadanie i tworzy plan początkowy
2. Przegląd planu opcjonalnego: w przypadku włączenia użytkownicy mogą przeglądać i zatwierdzać/modyfikować plan
3. Wybór agenta: Menedżer wybiera najbardziej odpowiedniego agenta dla każdej podtaki
4. Wykonanie: wybrany agent wykonuje część zadania
5. Ocena postępu: menedżer ocenia postęp i aktualizuje plan
6. Wykrywanie blokady: jeśli postęp zostanie zatrzymany, przeplanuj automatycznie z opcjonalnym procesem przeglądu, który może być przeprowadzony przez człowieka
7. Iteracja: kroki 3–6 powtarzają się do momentu zakończenia zadania lub osiągnięcia limitów
8. Końcowa synteza: Menedżer syntetyzuje wszystkie dane wyjściowe agenta do końcowego wyniku

Kompletny przykład

Zobacz pełne przykłady w repozytorium Agent Framework Samples.

Dalsze kroki

Orkiestracja przekazywania