Megosztás:

Microsoft Agent Framework munkafolyamat-orkesztációk – Magentic

A Magentic Orchestration még nem támogatott a C#-ban.

A Magentic vezénylés az AutoGen által feltalált Magentic-One rendszer alapján készült. Ez egy rugalmas, általános célú többügynök-minta, amelyet összetett, nyitott feladatokhoz terveztek, amelyek dinamikus együttműködést igényelnek. Ebben a mintában egy dedikált Magentic-vezető koordinálja a specializált ügynökök csapatát, és kiválasztja, hogy melyik ügynök legyen a következő lépés a változó környezet, a tevékenység előrehaladása és az ügynök képességei alapján.

A Magentic manager megosztott környezetet tart fenn, nyomon követi az előrehaladást, és valós időben igazítja a munkafolyamatot. Ez lehetővé teszi a rendszer számára az összetett problémák lebontására, az altevékenységek delegálására és a megoldások iteratív finomítására az ügynökök együttműködésével. A vezénylés különösen alkalmas olyan helyzetekre, amikor a megoldás elérési útja nem ismert előre, és több érvelési, kutatási és számítási fordulót igényelhet.

Magentic Orchestration

Jótanács

A Magentic vezénylés architektúrája megegyezik a csoportos csevegés vezénylési mintájával, egy nagyon hatékony vezetővel, aki tervezéssel koordinálja az ügynökök együttműködését. Ha a forgatókönyv összetett tervezés nélkül egyszerűbb koordinációt igényel, fontolja meg inkább a csoportos csevegési mintát.

Megjegyzés:

A Magentic-One tanulmányban 4 nagyon specializált ügynököt úgy terveztek, hogy egy nagyon specifikus feladathalmazt oldjanak meg. Az Agent Framework Magentic vezénylésében saját speciális ügynököket határozhat meg az adott alkalmazás igényeinek megfelelően. Azt azonban nem tesztelték, hogy a Magentic orchestráció milyen jól fog teljesíteni az eredeti Magentic-One tervezeten kívül.

Tudnivalók

  • Hogyan állítsunk be egy Magentic managert több specializált ügynök koordinálásához
  • A(z) streamelési események kezelése a WorkflowOutputEvent
  • Az emberi beavatkozású tervfelülvizsgálat végrehajtása
  • Az ügynökök együttműködésének és előrehaladásának nyomon követése összetett feladatokon keresztül

A specializált ügynökök definiálása

A Magentic orchestrationben speciális ügynököket határoz meg, amelyeket a menedzser dinamikusan választhat a feladatkövetelmények alapján:

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

A Magentic Munkafolyamat létrehozása

A MagenticBuilder munkafolyamat konfigurálása standard kezelővel(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()

Jótanács

A standard menedzsert a Magnetic-One tervezés alapján implementáljuk, a rögzített kéréseket az eredeti dokumentumból véve. A vezető viselkedését úgy szabhatja testre, hogy a konstruktorparamétereken keresztül adja meg a MagenticBuilder saját kéréseit. A kezelő további testreszabásához megvalósíthatja saját kezelőjét az MagenticManagerBase osztály alosztályozásával.

A munkafolyamat futtatása eseménystreameléssel

Összetett feladat végrehajtása és események kezelése streamelési kimenethez és vezénylési frissítésekhez:

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)

Speciális: Emberi közreműködéssel történő tervellenőrzés

Engedélyezze a human-in-the-loop (HITL) lehetőséget, hogy a felhasználók a végrehajtás előtt áttekinthessék és jóváhagyhassák a vezető által javasolt tervet. Ez annak biztosításához hasznos, hogy a terv megfeleljen a felhasználói elvárásoknak és követelményeknek.

A terv felülvizsgálatának két lehetősége van:

  1. Módosítás: A felhasználó visszajelzést küldhet a terv felülvizsgálatához, amely elindítja a kezelés újratervezését a visszajelzés alapján.
  2. Jóváhagyás: A felhasználó jóváhagyhatja a tervet as-is, amely lehetővé teszi a munkafolyamat folytatását.

A Magentic munkafolyamat felépítésekor engedélyezze a terv felülvizsgálatát azáltal, hogy enable_plan_review=True átadja.

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

A terv felülvizsgálati kérelmei WorkflowEvent-ként type="request_info" és MagenticPlanReviewRequest adatokkal kerülnek kibocsátásra. Ezeket a kéréseket az eseménystreamben kezelheti:

Jótanács

További információ a kérésekről és válaszokról a Kérések és válaszok útmutatóban.

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

Alapfogalmak

  • Dinamikus koordináció: A Magentic manager dinamikusan kiválasztja, hogy melyik ügynök legyen a következő lépés a változó környezet alapján
  • Iteratív finomítás: A rendszer több fordulóval képes lebontani az összetett problémákat, és iteratív módon finomítani a megoldásokat
  • Folyamatkövetés: Beépített mechanizmusok az elakadások észlelésére és szükség esetén a terv alaphelyzetbe állítására
  • Rugalmas együttműködés: Az ügynökök a felettes által meghatározott sorrendben többször is meghívhatók
  • Emberi felügyelet: A terv felülvizsgálatához választható emberi beavatkozási mechanizmusok

Munkafolyamat végrehajtási folyamata

A Magentic vezénylés a következő végrehajtási mintát követi:

  1. Tervezési fázis: A vezető elemzi a feladatot, és létrehoz egy kezdeti tervet
  2. Választható terv áttekintése: Ha engedélyezve van, az emberek áttekinthetik és jóváhagyhatják/módosíthatják a tervet
  3. Ügynök kiválasztása: A felettes kiválasztja az egyes altevékenységekhez legmegfelelőbb ügynököt
  4. Végrehajtás: A kijelölt ügynök végrehajtja a feladat egy részét
  5. Állapotfelmérés: A vezető kiértékeli az előrehaladást, és frissíti a tervet
  6. Elakadások észlelése: Ha a folyamat leáll, automatikusan újratervezhető egy opcionális emberi felülvizsgálati folyamattal
  7. Iteráció: A 3–6. lépés ismétlése a feladat befejezéséig vagy a korlátok eléréséig
  8. Végső szintézis: A kezelő az összes ügynökkimenetet végleges eredményként szintetizálja

Teljes példa

Tekintse meg a teljes mintákat az Agent Framework Samples adattárban.

Következő lépések

Handoff Orchestration

  • Last updated on