Compartir a través de

Orquestaciones de flujos de trabajo de Microsoft Agent Framework: Magentic

La Orquestación Magnética aún no se admite en C#.

La orquestación magentic está diseñada basada en el sistema Magentic-One inventado por AutoGen. Es un patrón multiagente flexible y de uso general diseñado para tareas complejas y abiertas que requieren colaboración dinámica. En este patrón, un administrador de Magentic dedicado coordina un equipo de agentes especializados, seleccionando qué agente debe actuar a continuación en función del contexto en constante evolución, el progreso de las tareas y las funcionalidades del agente.

El administrador de Magentic mantiene un contexto compartido, realiza un seguimiento del progreso y adapta el flujo de trabajo en tiempo real. Esto permite al sistema desglosar problemas complejos, delegar subtareas y refinar soluciones de forma iterativa a través de la colaboración del agente. La orquestación es especialmente adecuada para escenarios en los que la ruta de solución no se conoce de antemano y podría requerir varias rondas de razonamiento, investigación y cálculo.

Orquestación Magnética

Sugerencia

La orquestación magnética tiene la misma arquitectura que el patrón de orquestación de chat en grupo, con un administrador muy eficaz que usa la planificación para coordinar la colaboración de los agentes. Si su escenario requiere una coordinación más sencilla sin una planeación compleja, considere la posibilidad de usar el patrón chat en grupo en su lugar.

Nota:

En el documento Magentic-One , 4 agentes altamente especializados están diseñados para resolver un conjunto muy específico de tareas. En la orquestación magnética en el marco de trabajo Agent, puede definir sus propios agentes especializados para satisfacer las necesidades específicas de su aplicación. Sin embargo, aún no se ha probado cuán bien la orquestación Magentic funcionará fuera del diseño original de Magentic-One.

Temas que se abordarán

  • Cómo configurar un administrador magentic para coordinar varios agentes especializados
  • Cómo controlar eventos de streaming con WorkflowOutputEvent
  • Implementación de la revisión de planes con intervención humana
  • Cómo realizar un seguimiento de la colaboración del agente y el progreso a través de tareas complejas

Definir los agentes especializados

En la orquestación magentic, se definen agentes especializados que el administrador puede seleccionar dinámicamente en función de los requisitos de la tarea:

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

Construir el flujo de trabajo magnético

Use MagenticBuilder para configurar el flujo de trabajo con un administrador estándar(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()

Sugerencia

Un administrador estándar se implementa basado en el diseño de Magnetic-One, con indicaciones fijas tomadas del artículo original. Puede personalizar el comportamiento del administrador pasando sus propias indicaciones a través de los parámetros del MagenticBuilder constructor. Para personalizar aún más el administrador, también puede implementar su propio administrador subclasificando la clase MagenticManagerBase.

Ejecución del flujo de trabajo con streaming de eventos

Ejecute una tarea compleja y maneje eventos para la salida de streaming y actualizaciones de la orquestación.

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)

Avanzado: Revisión del plan de human-in-the-loop

Habilite human-in-the-loop (HITL) para permitir a los usuarios revisar y aprobar el plan propuesto del administrador antes de la ejecución. Esto es útil para asegurarse de que el plan se alinea con las expectativas y los requisitos del usuario.

Hay dos opciones para la revisión del plan:

  1. Revisión: el usuario puede proporcionar comentarios para revisar el plan, lo que desencadenará la administración para volver a planear en función de los comentarios.
  2. Aprobar: el usuario puede aprobar el plan as-is, lo que permite que el flujo de trabajo continúe.

Habilite la revisión del plan pasando enable_plan_review=True al compilar el flujo de trabajo de 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()

Las solicitudes de revisión del plan se emiten como WorkflowEvent con type="request_info" y MagenticPlanReviewRequest de datos. Puede controlar estas solicitudes en el flujo de eventos:

Sugerencia

Obtenga más información sobre las solicitudes y respuestas en la guía Solicitudes y respuestas .

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

Conceptos clave

  • Coordinación dinámica: el administrador magentic selecciona dinámicamente qué agente debe actuar a continuación en función del contexto en evolución.
  • Refinamiento iterativo: el sistema puede desglosar problemas complejos y refinar soluciones iterativamente a través de varias rondas
  • Seguimiento de progreso: mecanismos integrados para detectar paradas y restablecer el plan si es necesario
  • Colaboración flexible: los agentes se pueden llamar varias veces en cualquier orden determinado por el administrador.
  • Supervisión humana: mecanismos opcionales de participación humana para la revisión de planes

Flujo de ejecución del flujo de trabajo

La orquestación magnética sigue este patrón de ejecución:

  1. Fase de planeación: el administrador analiza la tarea y crea un plan inicial
  2. Revisión opcional del plan: si está habilitada, los seres humanos pueden revisar y aprobar o modificar el plan.
  3. Selección del agente: el administrador selecciona el agente más adecuado para cada subtarea.
  4. Ejecución: el agente seleccionado ejecuta su parte de la tarea.
  5. Evaluación del progreso: el administrador evalúa el progreso y actualiza el plan.
  6. Detección de estancamiento: si el progreso se detiene, reprogramación automática con un proceso de revisión humana opcional.
  7. Iteración: pasos 3-6 repetir hasta que se complete la tarea o se alcancen los límites
  8. Síntesis final: el administrador sintetiza todas las salidas del agente en un resultado final

Ejemplo completo

Consulte ejemplos completos en el repositorio Agent Framework Samples.

Pasos siguientes

Orquestación de entrega

  • Last updated on