Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Microsoft Autogen MagenticOne Prompts

From Leeroopedia
Key Value
id Microsoft_Autogen_MagenticOne_Prompts
source Microsoft_Autogen
category Prompts

Overview

Description

The MagenticOne Prompts module defines the comprehensive prompt templates and data models used by the MagenticOneOrchestrator for intelligent multi-agent coordination. These prompts guide the LLM in performing sophisticated task analysis, planning, progress monitoring, and adaptive replanning.

The module provides:

  • Task Ledger Prompts: Prompts for initial task analysis (facts, plan)
  • Progress Ledger Prompts: Prompts for monitoring progress and selecting next speaker
  • Update Prompts: Prompts for replanning when progress stalls
  • Final Answer Prompt: Prompt for synthesizing final response
  • Structured Output Models: Pydantic models for parsing LLM JSON responses

The prompts implement a sophisticated orchestration strategy:

  1. Fact Gathering: Identify given facts, facts to look up, facts to derive, and educated guesses
  2. Planning: Create initial plan considering team composition
  3. Progress Monitoring: Assess task completion, loop detection, and progress
  4. Dynamic Routing: Select next speaker and provide specific instructions
  5. Adaptive Replanning: Update facts and plans when progress stalls

All prompts use template variables (task, team, facts, plan, names) that are populated at runtime with conversation context.

Usage

These prompts are used internally by MagenticOneOrchestrator to:

  • Analyze tasks before starting execution
  • Make intelligent agent selection decisions at each turn
  • Detect when the team is stuck or in loops
  • Trigger replanning when progress stalls
  • Generate final answers from conversation transcripts

The structured output models (LedgerEntry, LedgerEntryBooleanAnswer, LedgerEntryStringAnswer) ensure the LLM provides parseable JSON responses that the orchestrator can use for decision-making.

Code Reference

Source Location

  • Repository: https://github.com/microsoft/autogen
  • File Path: /tmp/kapso_repo_2mr4n2g4/python/packages/autogen-agentchat/src/autogen_agentchat/teams/_group_chat/_magentic_one/_prompts.py
  • Lines: 1-150

Signature

# Prompt constants
ORCHESTRATOR_SYSTEM_MESSAGE = ""

ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT = """..."""

ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT = """..."""

ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT = """..."""

ORCHESTRATOR_PROGRESS_LEDGER_PROMPT = """..."""

ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT = """..."""

ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT = """..."""

ORCHESTRATOR_FINAL_ANSWER_PROMPT = """..."""


# Structured output models
class LedgerEntryBooleanAnswer(BaseModel):
    reason: str
    answer: bool


class LedgerEntryStringAnswer(BaseModel):
    reason: str
    answer: str


class LedgerEntry(BaseModel):
    is_request_satisfied: LedgerEntryBooleanAnswer
    is_in_loop: LedgerEntryBooleanAnswer
    is_progress_being_made: LedgerEntryBooleanAnswer
    next_speaker: LedgerEntryStringAnswer
    instruction_or_question: LedgerEntryStringAnswer

Import

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_SYSTEM_MESSAGE,
    ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT,
    ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT,
    ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT,
    ORCHESTRATOR_PROGRESS_LEDGER_PROMPT,
    ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT,
    ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT,
    ORCHESTRATOR_FINAL_ANSWER_PROMPT,
    LedgerEntry,
    LedgerEntryBooleanAnswer,
    LedgerEntryStringAnswer
)

I/O Contract

Prompt Templates

Prompt Constant Template Variables Purpose
ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT {task} Initial fact gathering: given facts, facts to look up, facts to derive, educated guesses
ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT {team} Plan creation based on team composition
ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT {task}, {team}, {facts}, {plan} Combined task context for agents
ORCHESTRATOR_PROGRESS_LEDGER_PROMPT {task}, {team}, {names} Progress assessment and next speaker selection (returns JSON)
ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT {task}, {facts} Update facts when progress stalls
ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT {team} Update plan after failure/stall
ORCHESTRATOR_FINAL_ANSWER_PROMPT {task} Generate final answer from conversation

Structured Output Models

Model Fields Description
LedgerEntryBooleanAnswer reason: str, answer: bool Reasoned boolean response
LedgerEntryStringAnswer reason: str, answer: str Reasoned string response
LedgerEntry is_request_satisfied, is_in_loop, is_progress_being_made, next_speaker, instruction_or_question Complete progress assessment

LedgerEntry JSON Schema

{
    "is_request_satisfied": {
        "reason": "string explaining assessment",
        "answer": true/false
    },
    "is_in_loop": {
        "reason": "string explaining loop detection",
        "answer": true/false
    },
    "is_progress_being_made": {
        "reason": "string explaining progress",
        "answer": true/false
    },
    "next_speaker": {
        "reason": "string explaining selection",
        "answer": "agent_name"
    },
    "instruction_or_question": {
        "reason": "string explaining instruction",
        "answer": "instruction text"
    }
}

Usage Examples

Initial Fact Gathering

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT
)


async def gather_facts(task: str, model_client):
    # Format prompt with task
    prompt = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT.format(task=task)

    # Send to LLM
    response = await model_client.create([{"role": "user", "content": prompt}])

    # Response contains:
    # 1. GIVEN OR VERIFIED FACTS
    # 2. FACTS TO LOOK UP
    # 3. FACTS TO DERIVE
    # 4. EDUCATED GUESSES

    facts = response.content
    return facts

Creating Initial Plan

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT
)


async def create_plan(team_description: str, model_client):
    # Format team member descriptions
    team = "\n".join([
        f"- {agent.name}: {agent.description}"
        for agent in participants
    ])

    # Format prompt
    prompt = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT.format(team=team)

    # Send to LLM
    response = await model_client.create([{"role": "user", "content": prompt}])

    # Response contains bullet-point plan
    plan = response.content
    return plan

Progress Assessment

import json
from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_PROGRESS_LEDGER_PROMPT,
    LedgerEntry
)


async def assess_progress(task: str, team: str, names: list, model_client):
    # Format prompt
    names_str = ", ".join(names)
    prompt = ORCHESTRATOR_PROGRESS_LEDGER_PROMPT.format(
        task=task,
        team=team,
        names=names_str
    )

    # Request JSON response
    response = await model_client.create([
        {"role": "user", "content": prompt}
    ])

    # Parse JSON response
    ledger_dict = json.loads(response.content)
    ledger = LedgerEntry.model_validate(ledger_dict)

    # Use ledger for decision making
    if ledger.is_request_satisfied.answer:
        print("Task is complete!")
    elif ledger.is_in_loop.answer:
        print("Detected loop, need to replan")
    elif not ledger.is_progress_being_made.answer:
        print("Progress stalled, need to replan")
    else:
        next_agent = ledger.next_speaker.answer
        instruction = ledger.instruction_or_question.answer
        print(f"Next: {next_agent} - {instruction}")

    return ledger

Updating Facts When Stalled

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT
)


async def update_facts_on_stall(task: str, old_facts: str, model_client):
    # Format prompt
    prompt = ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT.format(
        task=task,
        facts=old_facts
    )

    # Get updated facts
    response = await model_client.create([{"role": "user", "content": prompt}])

    # Response contains updated fact sheet with:
    # - New educated guesses
    # - Moved items between sections
    # - Updated information based on conversation

    updated_facts = response.content
    return updated_facts

Updating Plan After Failure

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT
)


async def update_plan_on_failure(team: str, model_client):
    # Format prompt
    prompt = ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT.format(team=team)

    # Get updated plan
    response = await model_client.create([{"role": "user", "content": prompt}])

    # Response contains:
    # 1. Explanation of what went wrong
    # 2. New plan that avoids previous mistakes

    updated_plan = response.content
    return updated_plan

Generating Final Answer

from autogen_agentchat.teams._group_chat._magentic_one._prompts import (
    ORCHESTRATOR_FINAL_ANSWER_PROMPT
)


async def generate_final_answer(task: str, conversation_history: list, model_client):
    # Format prompt
    prompt = ORCHESTRATOR_FINAL_ANSWER_PROMPT.format(task=task)

    # Include conversation history as context
    messages = conversation_history + [{"role": "user", "content": prompt}]

    # Generate final answer
    response = await model_client.create(messages)

    # Response is phrased as if speaking to user
    final_answer = response.content
    return final_answer

Complete Orchestration Flow

async def orchestration_example(task: str, participants: list, model_client):
    # Step 1: Gather facts
    facts_prompt = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT.format(task=task)
    facts_response = await model_client.create([{"role": "user", "content": facts_prompt}])
    facts = facts_response.content

    # Step 2: Create plan
    team = "\n".join([f"- {a.name}: {a.description}" for a in participants])
    plan_prompt = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT.format(team=team)
    plan_response = await model_client.create([{"role": "user", "content": plan_prompt}])
    plan = plan_response.content

    # Step 3: Execute with progress monitoring
    n_stalls = 0
    max_stalls = 3

    while True:
        # Assess progress
        names = [a.name for a in participants]
        progress_prompt = ORCHESTRATOR_PROGRESS_LEDGER_PROMPT.format(
            task=task, team=team, names=", ".join(names)
        )
        progress_response = await model_client.create([{"role": "user", "content": progress_prompt}])
        ledger = LedgerEntry.model_validate(json.loads(progress_response.content))

        # Check completion
        if ledger.is_request_satisfied.answer:
            break

        # Check for stalls
        if ledger.is_in_loop.answer or not ledger.is_progress_being_made.answer:
            n_stalls += 1
            if n_stalls >= max_stalls:
                break

            # Replan
            facts = await update_facts_on_stall(task, facts, model_client)
            plan = await update_plan_on_failure(team, model_client)

        # Execute next step
        next_agent = ledger.next_speaker.answer
        instruction = ledger.instruction_or_question.answer
        # ... invoke agent ...

    # Step 4: Generate final answer
    final_prompt = ORCHESTRATOR_FINAL_ANSWER_PROMPT.format(task=task)
    final_response = await model_client.create([{"role": "user", "content": final_prompt}])
    return final_response.content

Custom Prompt Template

# Create custom final answer prompt
CUSTOM_FINAL_ANSWER = """
Task: {task}

The conversation above shows how the team solved this task.

Please provide:
1. Executive summary (2-3 sentences)
2. Key findings (bullet points)
3. Methodology used
4. Recommendations (if applicable)

Format in markdown.
"""


async def use_custom_prompt(task: str):
    from autogen_agentchat.teams import MagenticOneGroupChat

    team = MagenticOneGroupChat(
        participants=[agent1, agent2],
        model_client=model_client,
        final_answer_prompt=CUSTOM_FINAL_ANSWER
    )

    result = await team.run(task=task)
    # Final answer uses custom template

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment