Aracılığıyla paylaş

Microsoft Agent Framework İş Akışları Düzenlemeleri - Magentic

Magentic Orchestration henüz C# dilinde desteklenmiyor.

Magentic orchestration, AutoGen tarafından icat edilen Magentic-One sistemine göre tasarlanmıştır. Dinamik işbirliği gerektiren karmaşık, açık uçlu görevler için tasarlanmış esnek, genel amaçlı çok aracılı bir desendir. Bu düzende, ayrılmış bir Magentic yöneticisi özelleştirilmiş aracılardan oluşan bir ekibi koordine eder ve gelişen bağlam, görev ilerleme durumu ve aracı özelliklerine göre hangi aracının bir sonraki adıma geçileceğini seçer.

Magentic yöneticisi paylaşılan bir bağlam tutar, ilerleme durumunu izler ve iş akışını gerçek zamanlı olarak uyarlar. Bu, sistemin karmaşık sorunları ayırmasını, alt görevleri devretmesini ve aracı işbirliği aracılığıyla çözümleri yinelemeli olarak geliştirmesini sağlar. Düzenleme özellikle çözüm yolunun önceden bilinmediği ve birden çok mantık, araştırma ve hesaplama turu gerektirebileceği senaryolar için uygundur.

Manyetik Orkestrasyon

Tip

Magentic orchestration, aracı işbirliğini koordine etmek için planlamayı kullanan çok güçlü bir yönetici ile Grup Sohbeti düzenleme düzeniyle aynı mimariye sahiptir. Senaryonuz karmaşık planlama olmadan daha basit bir koordinasyon gerektiriyorsa bunun yerine Grup Sohbeti düzenini kullanmayı göz önünde bulundurun.

Uyarı

Magentic-One makalesinde, çok özel bir görev kümesini çözmek için 4 yüksek oranda özelleştirilmiş aracı tasarlanmıştır. Agent Framework'teki Magentic düzenlemesinde, kendi özel aracılarınızı kendi uygulama gereksinimlerinize uyacak şekilde tanımlayabilirsiniz. Ancak, Magentic düzenlemesinin özgün Magentic-One tasarımı dışında ne kadar iyi performans göstereceği test edilmemiştir.

Öğrenecekler

  • Birden çok özel aracıyı koordine etmek için bir Magentic yöneticisini ayarlamak
  • Akış olayları WorkflowOutputEvent ile nasıl işlenir?
  • İnsan etkileşimli plan incelemesini uygulama
  • Karmaşık görevlerde ajanların işbirliği ve gelişimini izleme

Özel Aracılarınızı Tanımlayın

Magentic orchestration'da, yöneticinin görev gereksinimlerine göre dinamik olarak seçebileceği özelleştirilmiş aracılar tanımlarsınız:

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

Manyetik İş Akışı Oluşturun

İş akışını standart bir yöneticiyle yapılandırmak için kullanınMagenticBuilder: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

Standart bir yönetici, orijinal makaleden alınan belirli istemlerle Magnetic-One tasarımına göre uygulanır. Oluşturucu parametreleri aracılığıyla kendi istemlerinizi geçirerek yöneticinin MagenticBuilder davranışını özelleştirebilirsiniz. Yöneticiyi daha fazla özelleştirmek için, MagenticManagerBase sınıfını alt sınıflayarak kendi yöneticinizi uygulayabilirsiniz.

Etkinlik Aktarımı ile İş Akışını Çalıştırma

Karmaşık bir görevi gerçekleştirir ve yayın çıktısı ve orkestrasyon güncelleştirmeleri için olayları işler.

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)

Gelişmiş: Döngüdeki İnsan Planı İncelemesi

Kullanıcıların yürütmeden önce yöneticinin önerdiği planı gözden geçirmesine ve onaylamasına izin vermek için döngüdeki insan (HITL) özelliğini etkinleştirin. Bu, planın kullanıcı beklentileri ve gereksinimleriyle uyumlu olmasını sağlamak için kullanışlıdır.

Plan incelemesi için iki seçenek vardır:

  1. Düzelt: Kullanıcı planı gözden geçirmesi için geri bildirim sağlayabilir ve bu da geri bildirime göre yeniden planlamayı tetikler.
  2. Onayla: Kullanıcı, planı olduğu gibi onaylayarak iş akışının devam etmesine izin verebilir.

Magentic iş akışını oluştururken enable_plan_review=True ile plan gözden geçirmesini etkinleştirin.

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

Plan gözden geçirme istekleri, WorkflowEvent ve type="request_info" verileriyle birlikte MagenticPlanReviewRequest olarak gönderilir. Bu istekleri olay akışında işleyebilirsiniz:

Tip

İstekler ve Yanıtlar kılavuzunda istekler ve yanıtlar hakkında daha fazla bilgi edinin.

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

Önemli Kavramlar

  • Dinamik Koordinasyon: Magentic yöneticisi, gelişen bağlama göre bir sonraki eyleme geçilmesi gereken aracıyı dinamik olarak seçer
  • Yinelemeli İyileştirme: Sistem karmaşık sorunları ayırabilir ve çözümleri birden çok tur aracılığıyla yinelemeli olarak geliştirebilir
  • İlerleme İzleme: Duraklamaları algılamak ve gerekirse planı sıfırlamak için yerleşik mekanizmalar.
  • Esnek İşbirliği: Aracılar, yönetici tarafından belirlenen herhangi bir sırayla birden çok kez çağrılabilir
  • İnsan Gözetimi: Plan incelemesi için isteğe bağlı insan müdahalesi ile gerçekleştirilen mekanizmalar

İş Akışı Yürütme Süreci

Magentic düzenlemesi şu yürütme desenini izler:

  1. Planlama Aşaması: Yönetici görevi analiz eder ve bir ilk plan oluşturur
  2. İsteğe Bağlı Plan İncelemesi: Etkinleştirilirse, insanlar planı gözden geçirebilir ve onaylayabilir/değiştirebilir
  3. Aracı Seçimi: Yönetici her alt görev için en uygun aracıyı seçer
  4. Yürütme: Seçili aracı görevin kendi bölümünü yürütür
  5. İlerleme Değerlendirmesi: Yönetici ilerleme durumunu değerlendirir ve planı güncelleştirir
  6. Durak Algılama: İlerleme durursa, isteğe bağlı bir insan inceleme işlemiyle otomatik olarak yeniden planlayın
  7. Yineleme: 3-6 arası adımlar, görev tamamlanana veya sınırlara ulaşılana kadar yinelenir
  8. Son Sentez: Yönetici tüm aracı çıkışlarını nihai bir sonucta sentezler

Tam Örnek

Bkz. Agent Framework Örnekleri deposundaki tüm örnekler.

Sonraki Adımlar

İletim Düzenleme

  • Last updated on