Bagikan melalui

Orkestrasi Alur Kerja Microsoft Agent Framework - Magentic

Magentic Orchestration belum didukung di C#.

Orkestrasi magentik dirancang berdasarkan sistem Magentic-One yang ditemukan oleh AutoGen. Ini adalah pola multi-agen tujuan umum yang fleksibel yang dirancang untuk tugas kompleks dan terbuka yang memerlukan kolaborasi dinamis. Dalam pola ini, manajer Magentic khusus mengoordinasikan tim agen khusus, memilih agen mana yang harus bertindak selanjutnya berdasarkan konteks yang berkembang, kemajuan tugas, dan kemampuan agen.

Manajer Magentic mempertahankan konteks bersama, melacak kemajuan, dan mengadaptasi alur kerja secara real time. Ini memungkinkan sistem untuk memecah masalah kompleks, mendelegasikan subtugas, dan memperbaiki solusi secara berulang melalui kolaborasi agen. Orkestrasi sangat cocok untuk skenario di mana jalur solusi tidak diketahui sebelumnya dan mungkin memerlukan beberapa putaran penalaran, penelitian, dan komputasi.

Orkestrasi Magnetik

Petunjuk / Saran

Orkestrasi Magentic memiliki arsitektur yang sama dengan pola orkestrasi Obrolan Grup , dengan manajer yang sangat kuat yang menggunakan perencanaan untuk mengoordinasikan kolaborasi agen. Jika skenario Anda memerlukan koordinasi yang lebih sederhana tanpa perencanaan yang kompleks, pertimbangkan untuk menggunakan pola Obrolan Grup sebagai gantinya.

Nota

Dalam makalah Magentic-One , 4 agen yang sangat khusus dirancang untuk menyelesaikan serangkaian tugas yang sangat spesifik. Dalam orkestrasi Magentic dalam Agent Framework, Anda dapat menentukan agen khusus Anda sendiri agar sesuai dengan kebutuhan aplikasi spesifik Anda. Namun, belum diuji seberapa baik orkestrasi Magentic akan berkinerja dengan baik di luar desain asli Magentic-One.

Apa yang akan Anda Pelajari

  • Cara menyiapkan manajer Magentic untuk mengoordinasikan beberapa agen khusus
  • Cara menangani peristiwa streaming dengan WorkflowOutputEvent
  • Cara menerapkan tinjauan rencana human-in-the-loop
  • Cara melacak kolaborasi agen dan kemajuan melalui tugas yang kompleks

Tentukan Agen Khusus Anda

Dalam orkestrasi Magentic, Anda menentukan agen khusus yang dapat dipilih manajer secara dinamis berdasarkan persyaratan tugas:

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

Membangun Alur Kerja Magentic

Gunakan MagenticBuilder untuk mengonfigurasi alur kerja dengan manajer standar(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()

Petunjuk / Saran

Manajer standar diimplementasikan berdasarkan desain Magentic-One, dengan perintah tetap yang diambil dari kertas asli. Anda dapat menyesuaikan perilaku manajer dengan meneruskan perintah Anda sendiri melalui MagenticBuilder parameter konstruktor. Untuk menyesuaikan manajer lebih lanjut, Anda juga dapat menerapkan manajer Anda sendiri dengan sub kelas MagenticManagerBase .

Jalankan Alur Kerja dengan Streaming Peristiwa

Jalankan tugas yang kompleks dan tangani peristiwa untuk output streaming dan pembaruan orkestrasi:

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)

Tingkat Lanjut: Tinjauan Rencana Human-in-the-Loop

Aktifkan human-in-the-loop (HITL) untuk memungkinkan pengguna meninjau dan menyetujui rencana yang diusulkan manajer sebelum eksekusi. Ini berguna untuk memastikan bahwa paket selaras dengan harapan dan persyaratan pengguna.

Ada dua opsi untuk peninjauan rencana.

  1. Revisi: Pengguna dapat memberikan umpan balik untuk merevisi rencana, yang akan memicu tim manajemen untuk merencanakan ulang berdasarkan umpan balik.
  2. Setujui: Pengguna dapat menyetujui paket as-is, memungkinkan alur kerja untuk melanjutkan.

Aktifkan tinjauan rencana dengan meneruskan enable_plan_review=True ketika membangun ulang alur kerja Magnetic:

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

Permintaan ulasan rencana dikirimkan sebagai WorkflowEvent bersama type="request_info" dan data MagenticPlanReviewRequest. Anda dapat menangani permintaan ini di aliran peristiwa:

Petunjuk / Saran

Pelajari selengkapnya tentang permintaan dan respons di panduan Permintaan dan Respons .

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

Konsep utama

  • Koordinasi Dinamis: Manajer Magentic secara dinamis memilih agen mana yang harus bertindak berikutnya berdasarkan konteks yang berkembang
  • Penyempurnaan Berulang: Sistem dapat memecah masalah kompleks dan secara berulang menyempurnakan solusi melalui beberapa putaran
  • Pelacakan Kemajuan: Mekanisme bawaan untuk mendeteksi macet dan mereset rencana jika diperlukan
  • Kolaborasi Fleksibel: Agen dapat dipanggil beberapa kali dalam urutan apa pun seperti yang ditentukan oleh manajer
  • Pengawasan Manusia: Mekanisme mengiringi manusia secara opsional untuk peninjauan rencana

Proses Eksekusi Alur Kerja

Orkestrasi Magentik mengikuti pola eksekusi ini:

  1. Fase Perencanaan: Manajer menganalisis tugas dan membuat rencana awal
  2. Tinjauan Paket Opsional: Jika diaktifkan, manusia dapat meninjau dan menyetujui/memodifikasi paket
  3. Pemilihan Agen: Manajer memilih agen yang paling tepat untuk setiap subtugas
  4. Eksekusi: Agen yang dipilih menjalankan bagian tugas mereka
  5. Penilaian Kemajuan: Manajer mengevaluasi kemajuan dan memperbarui rencana
  6. Deteksi Macet: Jika kemajuan macet, rencana ulang otomatis dengan opsi peninjauan manusia.
  7. Iterasi: Langkah 3-6 berulang hingga tugas selesai atau batas tercapai
  8. Sintesis Akhir: Manajer mensintesis semua output agen menjadi hasil akhir

Contoh Lengkap

Lihat sampel lengkap di repositori Sampel Kerangka Kerja Agen.

Langkah selanjutnya

Orkestrasi Handoff

  • Last updated on