Indelingen van Microsoft Agent Framework-werkstromen - Magentic

Binnenkort beschikbaar...

In Magentic Orchestration definieert u gespecialiseerde agents die de manager dynamisch kan selecteren op basis van taakvereisten:

from agent_framework import ChatAgent, HostedCodeInterpreterTool
from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient

researcher_agent = ChatAgent(
    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 = ChatAgent(
    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=OpenAIResponsesClient(),
    tools=HostedCodeInterpreterTool(),
)

# Create a manager agent for orchestration
manager_agent = ChatAgent(
    name="MagenticManager",
    description="Orchestrator that coordinates the research and coding workflow",
    instructions="You coordinate a team to complete complex tasks efficiently.",
    chat_client=OpenAIChatClient(),
)

De magentische werkstroom bouwen

Gebruik MagenticBuilder dit om de werkstroom te configureren met een standaardmanager:

from agent_framework import MagenticBuilder

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

De werkstroom uitvoeren met gebeurtenisstreaming

Voer een complexe taak uit en verwerkt gebeurtenissen voor streaming-uitvoer- en indelingsupdates:

import json
import asyncio
from typing import cast

from agent_framework import (
    AgentRunUpdateEvent,
    ChatMessage,
    MagenticOrchestratorEvent,
    MagenticProgressLedger,
    WorkflowOutputEvent,
)

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: WorkflowOutputEvent | None = None
async for event in workflow.run_stream(task):
    if isinstance(event, AgentRunUpdateEvent):
        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 isinstance(event, MagenticOrchestratorEvent):
        print(f"\n[Magentic Orchestrator Event] Type: {event.event_type.name}")
        if isinstance(event.data, MagenticProgressLedger):
            print(f"Please review progress ledger:\n{json.dumps(event.data.to_dict(), indent=2)}")
        else:
            print(f"Unknown data type in MagenticOrchestratorEvent: {type(event.data)}")

        # 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 isinstance(event, WorkflowOutputEvent):
        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[ChatMessage], output_event.data)
output = output_messages[-1].text
print(output)

Geavanceerd: Beoordeling van het human-in-the-loop-plan

Schakel HITL (Human-In-The-Loop) in zodat gebruikers het voorgestelde plan van de manager kunnen beoordelen en goedkeuren voordat ze worden uitgevoerd. Dit is handig om ervoor te zorgen dat het plan overeenkomt met de verwachtingen en vereisten van gebruikers.

Er zijn twee opties voor planbeoordeling:

1. Herzien: De gebruiker kan feedback geven om het plan te herzien, waardoor het beheer opnieuw kan worden gepland op basis van de feedback.
2. Goedkeuren: De gebruiker kan het plan goedkeuren as-is, zodat de werkstroom kan worden voortgezet.

Planbeoordeling kan eenvoudig worden ingeschakeld door .with_plan_review() toe te voegen aan het bouwen van de Magentic-werkstroom.

from agent_framework import (
    AgentRunUpdateEvent,
    ChatAgent,
    ChatMessage,
    MagenticBuilder,
    MagenticPlanReviewRequest,
    RequestInfoEvent,
    WorkflowOutputEvent,
)

workflow = (
    MagenticBuilder()
    .participants([researcher_agent, analyst_agent])
    .with_standard_manager(
        agent=manager_agent,
        max_round_count=10,
        max_stall_count=1,
        max_reset_count=2,
    )
    .with_plan_review()  # Request human input for plan review
    .build()
)

Planbeoordelingsaanvragen worden verzonden als RequestInfoEvent met MagenticPlanReviewRequest gegevens. U kunt deze aanvragen afhandelen in de gebeurtenisstroom:

pending_request: RequestInfoEvent | None = None
pending_responses: dict[str, MagenticPlanReviewResponse] | None = None
output_event: WorkflowOutputEvent | None = None

while not output_event:
    if pending_responses is not None:
        stream = workflow.send_responses_streaming(pending_responses)
    else:
        stream = workflow.run_stream(task)

    last_message_id: str | None = None
    async for event in stream:
        if isinstance(event, AgentRunUpdateEvent):
            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 isinstance(event, RequestInfoEvent) and event.request_type is MagenticPlanReviewRequest:
            pending_request = event

        elif isinstance(event, WorkflowOutputEvent):
            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

Sleutelbegrippen

• Dynamische coördinatie: De Magentic Manager selecteert dynamisch welke agent moet handelen op basis van de veranderende context
• Iteratieve verfijning: Het systeem kan complexe problemen opsplitsen en iteratief verfijn oplossingen via meerdere rondes
• Voortgangstracering: ingebouwde mechanismen om vertragingen te detecteren en het plan indien nodig opnieuw in te stellen
• Flexibele samenwerking: Agents kunnen meerdere keren worden aangeroepen in elke volgorde zoals bepaald door de manager
• Menselijk toezicht: optionele mechanismen voor menselijke tussenkomst bij het beoordelen van plannen

Stroom voor werkstroomuitvoering

De Magentic orchestration volgt dit uitvoeringspatroon:

1. Planningsfase: De manager analyseert de taak en maakt een eerste plan
2. Optionele planbeoordeling: indien ingeschakeld, kunnen mensen het plan beoordelen en goedkeuren/wijzigen
3. Agentselectie: De manager selecteert de meest geschikte agent voor elke subtaak
4. Uitvoering: De geselecteerde agent voert het gedeelte van de taak uit
5. Voortgangsevaluatie: De manager evalueert de voortgang en werkt het plan bij
6. Stagnatiedetectie: als de voortgang stagneert, automatisch opnieuw plannen met een optionele menselijke controle
7. Iteratie: stap 3-6 herhalen totdat de taak is voltooid of de limieten zijn bereikt
8. Uiteindelijke synthese: De managersynthetiseert alle agentuitvoer in een eindresultaat

Volledig voorbeeld

Bekijk volledige voorbeelden in de opslagplaats met voorbeelden van agentframeworks.