Бөлісу құралы:

Оркестрации рабочих процессов Microsoft Agent Framework — magentic

Magentic Orchestration пока не поддерживается в C#.

Магнитная оркестрация разработана на основе системы Magentic-One, созданной AutoGen. Это гибкий универсальный многоагентный шаблон, предназначенный для сложных открытых задач, требующих динамической совместной работы. В этом шаблоне выделенный диспетчер Магнетик координирует команду специализированных агентов, выбирая, какому агенту следует действовать следующим, основываясь на изменяющемся контексте, прогрессе задач и возможностях агентов.

Диспетчер Magentic поддерживает общий контекст, отслеживает ход выполнения и адаптирует рабочий процесс в режиме реального времени. Это позволяет системе разбить сложные проблемы, делегировать подзадачи и итеративно уточнять решения с помощью совместной работы агента. Оркестрация особенно хорошо подходит для сценариев, когда путь решения не известен заранее и может потребовать нескольких раундов рассуждений, исследований и вычислений.

Magentic Orchestration

Подсказка

Magentic orchestration имеет ту же архитектуру, что и шаблон соркестрации группового общения, с очень мощным менеджером, использующим планирование для координации совместной работы агентов. Если в вашем сценарии требуется более простая координация без сложного планирования, рекомендуется использовать шаблон группового чата.

Замечание

В документе Magentic-One 4 высоко специализированных агентов предназначены для решения очень конкретного набора задач. В Magentic orchestration в Agent Framework можно определить собственные специализированные агенты под нужды вашего конкретного приложения. Тем не менее, не проверено, насколько хорошо Magentic оркестрация будет работать за пределами исходного проекта Magentic-One.

Цели обучения

  • Как настроить менеджер Magentic для координации нескольких специализированных агентов
  • Обработка событий потоковой передачи с помощью WorkflowEvent
  • Как реализовать обзор плана с участием человека
  • Отслеживание совместной работы агента и ход выполнения сложных задач

Определите своих специализированных агентов

В magentic orchestration вы определяете специализированные агенты, которые диспетчер может динамически выбирать на основе требований к задачам:

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ["FOUNDRY_MODEL"],
    credential=AzureCliCredential(),
)

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."
    ),
    client=client,
)

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.",
    client=client,
    tools=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.",
    client=client,
)

Создание магентического рабочего процесса

Используется MagenticBuilder для настройки рабочего процесса с помощью стандартного диспетчера(StandardMagenticManager):

from agent_framework.orchestrations import MagenticBuilder

workflow = MagenticBuilder(
    participants=[researcher_agent, coder_agent],
    intermediate_outputs=True,
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=3,
    max_reset_count=2,
).build()

Подсказка

Стандартный менеджер внедрен на основе дизайна Magnetic-One с фиксированными подсказками, взятыми из исходной статьи. Вы можете настроить поведение руководителя, передав собственные запросы с помощью MagenticBuilder параметров конструктора. Чтобы дополнительно настроить диспетчер, вы также можете реализовать собственный менеджер MagenticManagerBase посредством создания подкласса этого класса.

Запуск рабочего процесса с потоковой передачей событий

Выполните сложную задачу и обработайте события для потоковой передачи выходных данных и обновлений оркестрации:

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(task, stream=True):
    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, Message):
            print(f"Please review the plan:\n{event.data.content.text}")
        elif 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)

Продвинутый: обзор плана "Человек в контуре"

Включите механизм "человек в цикле" (HITL), чтобы пользователи могли просматривать и утверждать предлагаемый план руководителя перед выполнением. Это полезно для обеспечения соответствия плана ожиданиям и требованиям пользователей.

Существует два варианта проверки плана:

  1. Изменение. Пользователь может предоставить отзыв о пересмотре плана, который активирует руководителя для перепланирования на основе обратной связи.
  2. Утвердить: пользователь может утвердить план as-is, что позволит рабочему процессу продолжить работу.

Включите проверку плана, передавая enable_plan_review=True при создании рабочего процесса Magentic:

import json
import asyncio
from typing import cast

from agent_framework import (
    AgentResponseUpdate,
    Agent,
    Message,
    WorkflowEvent,
)
from agent_framework.orchestrations import (
    MagenticBuilder,
    MagenticPlanReviewRequest,
    MagenticPlanReviewResponse,
)

workflow = MagenticBuilder(
    participants=[researcher_agent, analyst_agent],
    intermediate_outputs=True,
    enable_plan_review=True,
    manager_agent=manager_agent,
    max_round_count=10,
    max_stall_count=1,
    max_reset_count=2,
).build()

Запросы на проверку плана создаются как WorkflowEvent с данными type="request_info" и MagenticPlanReviewRequest. Эти запросы можно обрабатывать в потоке событий:

Подсказка

Дополнительные сведения о запросах и ответах см. в руководстве по запросам и ответам .

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(stream=True, responses=pending_responses)
    else:
        stream = workflow.run(task, stream=True)

    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

Основные понятия

  • Динамическая координация: диспетчер Magentic динамически выбирает, какой агент должен действовать следующим на основе изменяющегося контекста.
  • Итеративное уточнение: система может разбить сложные проблемы и итеративно уточнить решения с помощью нескольких раундов
  • Отслеживание хода выполнения: Встроенные механизмы для обнаружения застоев и сброса плана при необходимости
  • Гибкая совместная работа: агенты могут вызываться несколько раз в любом порядке по решению менеджера
  • Человеческий надзор: необязательные механизмы с участием человека для проверки плана

Поток выполнения рабочего процесса

Оркестрация Magnetic следует этому образцу выполнения.

  1. Этап планирования: менеджер анализирует задачу и создает начальный план
  2. Необязательный обзор плана: если включен, люди могут просматривать и утверждать или изменять план.
  3. Выбор агента: менеджер выбирает наиболее подходящий агент для каждой подзадаки.
  4. Выполнение: выбранный агент выполняет свою часть задачи.
  5. Оценка хода выполнения: менеджер оценивает ход выполнения и обновляет план
  6. Обнаружение остановки: если прогресс застопорился, автоматически перепланировать с возможностью проверки человеком
  7. Итерация. Шаги 3-6 повторяются до завершения задачи или достижения ограничений
  8. Окончательный синтез: менеджер синтезирует все выходные данные агента в окончательный результат

Полный пример

Полные примеры см. в репозитории примеров Agent Framework.

Дальнейшие шаги

Оркестрация переключения

  • Last updated on