Implementation:Microsoft Autogen MagenticOne Prompts
| 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:
- Fact Gathering: Identify given facts, facts to look up, facts to derive, and educated guesses
- Planning: Create initial plan considering team composition
- Progress Monitoring: Assess task completion, loop detection, and progress
- Dynamic Routing: Select next speaker and provide specific instructions
- 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
- Microsoft_Autogen_MagenticOneGroupChat - Team that uses these prompts
- Microsoft_Autogen_MagenticOne_Orchestrator - Orchestrator implementation
- Microsoft_Autogen_ChatCompletionClient - LLM client interface
- Microsoft_Autogen_MagenticOneOrchestratorState - State model including task, facts, plan
- Microsoft_Autogen_BaseGroupChat - Base class for group chat teams
- Microsoft_Autogen_ChatAgent_Protocol - Protocol for participant agents
- Microsoft_Autogen_PromptTemplate - General prompt template utilities
- Microsoft_Autogen_AssistantAgent - Agent that receives instructions from orchestrator