diff --git a/python/.cspell.json b/python/.cspell.json index f8ea3b8ea374..592c977609a1 100644 --- a/python/.cspell.json +++ b/python/.cspell.json @@ -47,6 +47,7 @@ "logit", "logprobs", "lowlevel", + "Magentic", "mistralai", "mongocluster", "nd", @@ -77,4 +78,4 @@ "vertexai", "Weaviate" ] -} \ No newline at end of file +} diff --git a/python/samples/getting_started_with_agents/README.md b/python/samples/getting_started_with_agents/README.md index fd24e8aac009..fdc7e8aa5996 100644 --- a/python/samples/getting_started_with_agents/README.md +++ b/python/samples/getting_started_with_agents/README.md @@ -49,20 +49,6 @@ Example|Description _Note: For details on configuring an Azure AI Agent, please see [here](../getting_started_with_agents/azure_ai_agent/README.md)._ -## Multi Agent Orchestration - -Example|Description ----|--- -[step1_concurrent](../getting_started_with_agents/multi_agent_orchestration/step1_concurrent.py)|How to run multiple agents concurrently and manage their output. -[step1a_concurrent_structure_output](../getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structure_output.py)|How to run concurrent agents that return structured outputs. -[step2_sequential](../getting_started_with_agents/multi_agent_orchestration/step2_sequential.py)|How to run agents sequentially where each one depends on the previous. -[step2a_sequential_cancellation_token](../getting_started_with_agents/multi_agent_orchestration/step2a_sequential_cancellation_token.py)|How to use cancellation tokens in a sequential agent flow. -[step3_group_chat](../getting_started_with_agents/multi_agent_orchestration/step3_group_chat.py)|How to create a group chat with multiple agents interacting together. -[step3a_group_chat_human_in_the_loop](../getting_started_with_agents/multi_agent_orchestration/step3a_group_chat_human_in_the_loop.py)|How to include a human participant in a group chat with agents. -[step3b_group_chat_with_chat_completion_manager](../getting_started_with_agents/multi_agent_orchestration/step3b_group_chat_with_chat_completion_manager.py)|How to manage a group chat with agents using a chat completion manager. -[step4_handoff](../getting_started_with_agents/multi_agent_orchestration/step4_handoff.py)|How to hand off conversation or tasks from one agent to another. -[step4a_handoff_structured_inputs](../getting_started_with_agents/multi_agent_orchestration/step4a_handoff_structured_inputs.py)|How to perform structured inputs handoffs between agents. - ## OpenAI Assistant Agent @@ -86,6 +72,21 @@ Example|Description [step6_responses_agent_vision](../getting_started_with_agents/openai_responses/step6_responses_agent_vision.py)|How to provide an image as input to an OpenAI Responses agent. [step7_responses_agent_structured_outputs](../getting_started_with_agents/openai_responses/step7_responses_agent_structured_outputs.py)|How to use have an OpenAI Responses agent use structured outputs. +## Multi-Agent Orchestration + +Example|Description +---|--- +[step1_concurrent](../getting_started_with_agents/multi_agent_orchestration/step1_concurrent.py)|How to run agents in parallel on the same task. +[step1a_concurrent_structure_output](../getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structure_output.py)|How to run agents in parallel on the same task and return structured output. +[step2_sequential](../getting_started_with_agents/multi_agent_orchestration/step2_sequential.py)|How to run agents in sequence to complete a task. +[step2a_sequential_cancellation_token](../getting_started_with_agents/multi_agent_orchestration/step2a_sequential_cancellation_token.py)|How to cancel an invocation while it is in progress. +[step3_group_chat](../getting_started_with_agents/multi_agent_orchestration/step3_group_chat.py)|How to run agents in a group chat to complete a task. +[step3a_group_chat_human_in_the_loop](../getting_started_with_agents/multi_agent_orchestration/step3a_group_chat_human_in_the_loop.py)|How to run agents in a group chat with human in the loop. +[step3b_group_chat_with_chat_completion_manager](../getting_started_with_agents/multi_agent_orchestration/step3b_group_chat_with_chat_completion_manager.py)|How to run agents in a group chat with a more dynamic manager. +[step4_handoff](../getting_started_with_agents/multi_agent_orchestration/step4_handoff.py)|How to run agents in a handoff orchestration to complete a task. +[step4a_handoff_structure_input](../getting_started_with_agents/multi_agent_orchestration/step4a_handoff_structure_input.py)|How to run agents in a handoff orchestration to complete a task with structured input. +[step5_magentic](../getting_started_with_agents/multi_agent_orchestration/step5_magentic.py)|How to run agents in a Magentic orchestration to complete a task. + ## Configuring the Kernel Similar to the Semantic Kernel Python concept samples, it is necessary to configure the secrets diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/README.md b/python/samples/getting_started_with_agents/multi_agent_orchestration/README.md new file mode 100644 index 000000000000..2ce05547b4b0 --- /dev/null +++ b/python/samples/getting_started_with_agents/multi_agent_orchestration/README.md @@ -0,0 +1,33 @@ +# Multi-agent orchestration + +The Semantic Kernel Agent Framework now supports orchestrating multiple agents to work together to complete a task. + +## Background + +The following samples are beneficial if you are just getting started with Semantic Kernel. + +- [Chat Completion](../../concepts/chat_completion/) +- [Auto Function Calling](../../concepts/auto_function_calling/) +- [Structured Output](../../concepts/structured_output/) +- [Getting Started with Agents](../../getting_started_with_agents/) +- [More advanced agent samples](../../concepts/agents/) + +## Prerequisites + +The following environment variables are required to run the samples: + +- OPENAI_API_KEY +- OPENAI_CHAT_MODEL_ID + +However, if you are using other model services, feel free to switch to those in the samples. +Refer to [here](../../concepts/setup/README.md) on how to set up the environment variables for your model service. + +## Orchestrations + +| **Orchestrations** | **Description** | +| ------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **Concurrent** | Useful for tasks that will benefit from independent analysis from multiple agents. | +| **Sequential** | Useful for tasks that require a well-defined step-by-step approach. | +| **Handoff** | Useful for tasks that are dynamic in nature and don't have a well-defined step-by-step approach. | +| **GroupChat** | Useful for tasks that will benefit from inputs from multiple agents and a highly configurable conversation flow. | +| **Magentic** | GroupChat like with a planner based manager. Inspired by [Magentic One](https://www.microsoft.com/en-us/research/articles/magentic-one-a-generalist-multi-agent-system-for-solving-complex-tasks/). | diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structure_output.py b/python/samples/getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structured_outputs.py similarity index 100% rename from python/samples/getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structure_output.py rename to python/samples/getting_started_with_agents/multi_agent_orchestration/step1a_concurrent_structured_outputs.py diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/step5_magentic.py b/python/samples/getting_started_with_agents/multi_agent_orchestration/step5_magentic.py new file mode 100644 index 000000000000..cfe6bbcaeb21 --- /dev/null +++ b/python/samples/getting_started_with_agents/multi_agent_orchestration/step5_magentic.py @@ -0,0 +1,199 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio + +from semantic_kernel.agents import ( + Agent, + ChatCompletionAgent, + MagenticOrchestration, + OpenAIAssistantAgent, + StandardMagenticManager, +) +from semantic_kernel.agents.runtime import InProcessRuntime +from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion +from semantic_kernel.contents import ChatMessageContent + +""" +The following sample demonstrates how to create a Magentic orchestration with two agents: +- A Research agent that can perform web searches +- A Coder agent that can run code using the code interpreter + +Read more about Magentic here: +https://www.microsoft.com/en-us/research/articles/magentic-one-a-generalist-multi-agent-system-for-solving-complex-tasks/ + +This sample demonstrates the basic steps of creating and starting a runtime, creating +a Magentic orchestration with two agents and a Magentic manager, invoking the +orchestration, and finally waiting for the results. + +The Magentic manager requires a chat completion model that supports structured output. +""" + + +async def agents() -> list[Agent]: + """Return a list of agents that will participate in the Magentic orchestration. + + Feel free to add or remove agents. + """ + research_agent = ChatCompletionAgent( + name="ResearchAgent", + description="A helpful assistant with access to web search. Ask it to perform web searches.", + 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. + # Feel free to explore with other agents that support web search, for example, + # the `OpenAIResponseAgent` or `AzureAIAgent` with bing grounding. + service=OpenAIChatCompletion(ai_model_id="gpt-4o-search-preview"), + ) + + # Create an OpenAI Assistant agent with code interpreter capability + client, model = OpenAIAssistantAgent.setup_resources() + code_interpreter_tool, code_interpreter_tool_resources = OpenAIAssistantAgent.configure_code_interpreter_tool() + definition = await client.beta.assistants.create( + model=model, + 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.", + tools=code_interpreter_tool, + tool_resources=code_interpreter_tool_resources, + ) + coder_agent = OpenAIAssistantAgent( + client=client, + definition=definition, + ) + + return [research_agent, coder_agent] + + +def agent_response_callback(message: ChatMessageContent) -> None: + """Observer function to print the messages from the agents.""" + print(f"**{message.name}**\n{message.content}") + + +async def main(): + """Main function to run the agents.""" + # 1. Create a Magentic orchestration with two agents and a Magentic manager + # Note, the Standard Magentic manager uses prompts that have been tuned very + # carefully but it accepts custom prompts for advanced users and scenarios. + # For even more advanced scenarios, you can subclass the MagenticManagerBase + # and implement your own manager logic. + # The standard manager also requires a chat completion model that supports + # structured output. + magentic_orchestration = MagenticOrchestration( + members=await agents(), + manager=StandardMagenticManager(chat_completion_service=OpenAIChatCompletion()), + agent_response_callback=agent_response_callback, + ) + + # 2. Create a runtime and start it + runtime = InProcessRuntime() + runtime.start() + + # 3. Invoke the orchestration with a task and the runtime + orchestration_result = await magentic_orchestration.invoke( + 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 (e.g., 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)." + ), + runtime=runtime, + ) + + # 4. Wait for the results + value = await orchestration_result.get() + + print(f"\nFinal result:\n{value}") + + # 5. Stop the runtime when idle + await runtime.stop_when_idle() + + """ + Sample output: + **ResearchAgent** + Estimating the energy consumption and associated CO₂ emissions for training and inference of ResNet-50, BERT-base... + + **CoderAgent** + Here is the comparison of energy consumption and CO₂ emissions for each model (ResNet-50, BERT-base, and GPT-2) + over a 24-hour period: + + | Model | Training Energy (kWh) | Inference Energy (kWh) | Total Energy (kWh) | CO₂ Emissions (kg) | + |-----------|------------------------|------------------------|---------------------|---------------------| + | ResNet-50 | 21.11 | 0.08232 | 21.19232 | 19.50 | + | BERT-base | 0.048 | 0.23736 | 0.28536 | 0.26 | + | GPT-2 | 42.22 | 0.35604 | 42.57604 | 39.17 | + + ### Recommendations: + ... + + **CoderAgent** + Here are the recalibrated results for energy consumption and CO₂ emissions, assuming a more conservative approach + for models like GPT-2: + + | Model | Training Energy (kWh) | Inference Energy (kWh) | Total Energy (kWh) | CO₂ Emissions (kg) | + |------------------|------------------------|------------------------|---------------------|---------------------| + | ResNet-50 | 21.11 | 0.08232 | 21.19232 | 19.50 | + | BERT-base | 0.048 | 0.23736 | 0.28536 | 0.26 | + | GPT-2 (Adjusted) | 42.22 | 0.35604 | 42.57604 | 39.17 | + + ... + + **ResearchAgent** + Estimating the energy consumption and associated CO₂ emissions for training and inference of machine learning ... + + **ResearchAgent** + Estimating the energy consumption and CO₂ emissions of training and inference for ResNet-50, BERT-base, and ... + + **CoderAgent** + Here is the estimated energy use and CO₂ emissions for a full day of operation for each model on an Azure ... + + **ResearchAgent** + Recent analyses have highlighted the substantial energy consumption and carbon emissions associated with ... + + **CoderAgent** + Here's the refined estimation for the energy use and CO₂ emissions for optimized models on an Azure ... + + **CoderAgent** + To provide precise estimates for CO₂ emissions based on Azure's regional data centers' carbon intensity, we need ... + + **ResearchAgent** + To refine the CO₂ emission estimates for training and inference of ResNet-50, BERT-base, and GPT-2 on an Azure ... + + **CoderAgent** + Here's the refined comparative table for energy consumption and CO₂ emissions for ResNet-50, BERT-base, and GPT-2, + taking into account carbon intensity data for Azure's West Europe and Sweden Central regions: + + | Model | Energy (kWh) | CO₂ Emissions West Europe (kg) | CO₂ Emissions Sweden Central (kg) | + |------------|--------------|--------------------------------|-----------------------------------| + | ResNet-50 | 5.76 | 0.639 | 0.086 | + | BERT-base | 9.18 | 1.019 | 0.138 | + | GPT-2 | 12.96 | 1.439 | 0.194 | + + **Refined Recommendations:** + + ... + + Final result: + Here is the comprehensive report on energy efficiency and CO₂ emissions for ResNet-50, BERT-base, and GPT-2 models + when trained and inferred on an Azure Standard_NC6s_v3 VM for 24 hours. + + ### Energy Consumption and CO₂ Emissions: + + Based on refined analyses, here are the estimated energy consumption and CO₂ emissions for each model: + + | Model | Energy (kWh) | CO₂ Emissions West Europe (kg) | CO₂ Emissions Sweden Central (kg) | + |------------|--------------|--------------------------------|-----------------------------------| + | ResNet-50 | 5.76 | 0.639 | 0.086 | + | BERT-base | 9.18 | 1.019 | 0.138 | + | GPT-2 | 12.96 | 1.439 | 0.194 | + + ### Recommendations for Energy Efficiency: + + ... + """ + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/semantic_kernel/agents/__init__.py b/python/semantic_kernel/agents/__init__.py index b89af9f0106a..b00316dec715 100644 --- a/python/semantic_kernel/agents/__init__.py +++ b/python/semantic_kernel/agents/__init__.py @@ -40,6 +40,15 @@ "HandoffOrchestration": ".orchestration.handoffs", "OrchestrationHandoffs": ".orchestration.handoffs", "GroupChatOrchestration": ".orchestration.group_chat", + "RoundRobinGroupChatManager": ".orchestration.group_chat", + "BooleanResult": ".orchestration.group_chat", + "StringResult": ".orchestration.group_chat", + "MessageResult": ".orchestration.group_chat", + "GroupChatManager": ".orchestration.group_chat", + "MagenticOrchestration": ".orchestration.magentic", + "ProgressLedger": ".orchestration.magentic", + "MagenticManagerBase": ".orchestration.magentic", + "StandardMagenticManager": ".orchestration.magentic", } diff --git a/python/semantic_kernel/agents/__init__.pyi b/python/semantic_kernel/agents/__init__.pyi index 9b6a795aabb1..60ed72badb98 100644 --- a/python/semantic_kernel/agents/__init__.pyi +++ b/python/semantic_kernel/agents/__init__.pyi @@ -27,8 +27,16 @@ from .open_ai.open_ai_assistant_agent import AssistantAgentThread, OpenAIAssista from .open_ai.openai_responses_agent import OpenAIResponsesAgent, ResponsesAgentThread from .open_ai.run_polling_options import RunPollingOptions from .orchestration.concurrent import ConcurrentOrchestration -from .orchestration.group_chat import GroupChatManager, GroupChatOrchestration, RoundRobinGroupChatManager +from .orchestration.group_chat import ( + BooleanResult, + GroupChatManager, + GroupChatOrchestration, + MessageResult, + RoundRobinGroupChatManager, + StringResult, +) from .orchestration.handoffs import HandoffOrchestration, OrchestrationHandoffs +from .orchestration.magentic import MagenticManagerBase, MagenticOrchestration, ProgressLedger, StandardMagenticManager from .orchestration.sequential import SequentialOrchestration __all__ = [ @@ -49,6 +57,7 @@ __all__ = [ "AzureResponsesAgent", "BedrockAgent", "BedrockAgentThread", + "BooleanResult", "ChatCompletionAgent", "ChatHistoryAgentThread", "ConcurrentOrchestration", @@ -60,15 +69,21 @@ __all__ = [ "GroupChatManager", "GroupChatOrchestration", "HandoffOrchestration", + "MagenticManagerBase", + "MagenticOrchestration", + "MessageResult", "ModelConnection", "ModelSpec", "OpenAIAssistantAgent", "OpenAIResponsesAgent", "OrchestrationHandoffs", + "ProgressLedger", "ResponsesAgentThread", "RoundRobinGroupChatManager", "RunPollingOptions", "SequentialOrchestration", + "StandardMagenticManager", + "StringResult", "ToolSpec", "register_agent_type", ] diff --git a/python/semantic_kernel/agents/orchestration/magentic.py b/python/semantic_kernel/agents/orchestration/magentic.py new file mode 100644 index 000000000000..9970ae3506d7 --- /dev/null +++ b/python/semantic_kernel/agents/orchestration/magentic.py @@ -0,0 +1,879 @@ +# Copyright (c) Microsoft. All rights reserved. + +import asyncio +import logging +import sys +from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable +from typing import Annotated + +from pydantic import Field + +from semantic_kernel.agents.agent import Agent +from semantic_kernel.agents.orchestration.agent_actor_base import ActorBase, AgentActorBase +from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationBase, TIn, TOut +from semantic_kernel.agents.orchestration.prompts._magentic_prompts import ( + ORCHESTRATOR_FINAL_ANSWER_PROMPT, + ORCHESTRATOR_PROGRESS_LEDGER_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT, + ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT, + ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT, +) +from semantic_kernel.agents.runtime.core.cancellation_token import CancellationToken +from semantic_kernel.agents.runtime.core.core_runtime import CoreRuntime +from semantic_kernel.agents.runtime.core.message_context import MessageContext +from semantic_kernel.agents.runtime.core.routed_agent import message_handler +from semantic_kernel.agents.runtime.core.topic import TopicId +from semantic_kernel.agents.runtime.in_process.type_subscription import TypeSubscription +from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase +from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings +from semantic_kernel.contents.chat_history import ChatHistory +from semantic_kernel.contents.chat_message_content import ChatMessageContent +from semantic_kernel.contents.utils.author_role import AuthorRole +from semantic_kernel.functions.kernel_arguments import KernelArguments +from semantic_kernel.kernel import Kernel +from semantic_kernel.kernel_pydantic import KernelBaseModel +from semantic_kernel.prompt_template.kernel_prompt_template import KernelPromptTemplate +from semantic_kernel.prompt_template.prompt_template_config import PromptTemplateConfig +from semantic_kernel.utils.feature_stage_decorator import experimental + +if sys.version_info >= (3, 12): + from typing import override # pragma: no cover +else: + from typing_extensions import override # pragma: no cover + +logger: logging.Logger = logging.getLogger(__name__) + + +# region Messages and Types + + +@experimental +class MagenticStartMessage(KernelBaseModel): + """A message to start a magentic group chat.""" + + body: ChatMessageContent + + +@experimental +class MagenticRequestMessage(KernelBaseModel): + """A request message type for agents in a magentic group chat.""" + + agent_name: str + + +@experimental +class MagenticResponseMessage(KernelBaseModel): + """A response message type from agents in a magentic group chat.""" + + body: ChatMessageContent + + +@experimental +class MagenticResetMessage(KernelBaseModel): + """A message to reset a participant's chat history in a magentic group chat.""" + + pass + + +@experimental +class ProgressLedgerItem(KernelBaseModel): + """A progress ledger item.""" + + reason: str + answer: str | bool + + +@experimental +class ProgressLedger(KernelBaseModel): + """A progress ledger.""" + + is_request_satisfied: ProgressLedgerItem + is_in_loop: ProgressLedgerItem + is_progress_being_made: ProgressLedgerItem + next_speaker: ProgressLedgerItem + instruction_or_question: ProgressLedgerItem + + +@experimental +class MagenticContext(KernelBaseModel): + """Context for the Magentic manager.""" + + task: Annotated[ChatMessageContent, Field(description="The task to be completed.")] + chat_history: Annotated[ + ChatHistory, Field(description="The chat history to be used to generate the facts and plan.") + ] = ChatHistory() + participant_descriptions: Annotated[ + dict[str, str], Field(description="The descriptions of the participants in the group.") + ] + round_count: Annotated[int, Field(description="The number of rounds completed.")] = 0 + stall_count: Annotated[int, Field(description="The number of stalls detected.")] = 0 + reset_count: Annotated[int, Field(description="The number of resets detected.")] = 0 + + def reset(self) -> None: + """Reset the context. + + This will clear the chat history and reset the stall count. + This won't reset the task, round count, or participant descriptions. + """ + self.chat_history.clear() + self.stall_count = 0 + self.reset_count += 1 + + +# endregion Messages and Types + +# region MagenticManager + + +@experimental +class MagenticManagerBase(KernelBaseModel, ABC): + """Base class for the Magentic One manager.""" + + max_stall_count: Annotated[int, Field(description="The maximum number of stalls allowed before a reset.", ge=0)] = 3 + max_reset_count: Annotated[int | None, Field(description="The maximum number of resets allowed.", ge=0)] = None + max_round_count: Annotated[ + int | None, Field(description="The maximum number of rounds (agent responses) allowed.", gt=0) + ] = None + + @abstractmethod + async def plan(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Create a plan for the task. + + This is called when the task is first started. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The task ledger. + """ + ... + + @abstractmethod + async def replan(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Replan for the task. + + This is called when the task is stalled or looping. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The updated task ledger. + """ + ... + + @abstractmethod + async def create_progress_ledger(self, magentic_context: MagenticContext) -> ProgressLedger: + """Create a progress ledger. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ProgressLedger: The progress ledger. + """ + ... + + @abstractmethod + async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Prepare the final answer. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The final answer. + """ + ... + + +@experimental +class _TaskLedger(KernelBaseModel): + """Task ledger for the Standard Magentic manager.""" + + facts: Annotated[ChatMessageContent, Field(description="The facts about the task.")] + plan: Annotated[ChatMessageContent, Field(description="The plan for the task.")] + + +@experimental +class StandardMagenticManager(MagenticManagerBase): + """Standard Magentic manager implementation. + + This is the default implementation of the Magentic manager. + It uses the task ledger to keep track of the facts and plan for the task. + + This implementation requires a chat completion model with structured outputs. + """ + + chat_completion_service: ChatCompletionClientBase + prompt_execution_settings: PromptExecutionSettings + + task_ledger_facts_prompt: str = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT + task_ledger_plan_prompt: str = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT + task_ledger_full_prompt: str = ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT + task_ledger_facts_update_prompt: str = ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT + task_ledger_plan_update_prompt: str = ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT + progress_ledger_prompt: str = ORCHESTRATOR_PROGRESS_LEDGER_PROMPT + final_answer_prompt: str = ORCHESTRATOR_FINAL_ANSWER_PROMPT + + task_ledger: _TaskLedger | None = None + + def __init__( + self, + chat_completion_service: ChatCompletionClientBase, + prompt_execution_settings: PromptExecutionSettings | None = None, + **kwargs, + ) -> None: + """Initialize the Standard Magentic manager. + + Args: + chat_completion_service (ChatCompletionClientBase): The chat completion service to use. + prompt_execution_settings (PromptExecutionSettings | None): The prompt execution settings to use. + **kwargs: Additional keyword arguments for prompts: + - task_ledger_facts_prompt: The prompt to use for the task ledger facts. + - task_ledger_plan_prompt: The prompt to use for the task ledger plan. + - task_ledger_full_prompt: The prompt to use for the full task ledger. + - task_ledger_facts_update_prompt: The prompt to use for the task ledger facts update. + - task_ledger_plan_update_prompt: The prompt to use for the task ledger plan update. + - progress_ledger_prompt: The prompt to use for the progress ledger. + - final_answer_prompt: The prompt to use for the final answer. + """ + # Bast effort to make sure the service supports structured output. Even if the service supports + # structured output, the model may not support it, in which case there is no good way to check. + if prompt_execution_settings is None: + prompt_execution_settings = chat_completion_service.instantiate_prompt_execution_settings() + if not hasattr(prompt_execution_settings, "response_format"): + raise ValueError("The service must support structured output.") + else: + if not hasattr(prompt_execution_settings, "response_format"): + raise ValueError("The service must support structured output.") + if getattr(prompt_execution_settings, "response_format", None) is not None: + raise ValueError("The prompt execution settings must not have a response format set.") + + super().__init__( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + **kwargs, + ) + + @override + async def plan(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Plan the task. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The task ledger. + """ + # 1. Gather the facts + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.task_ledger_facts_prompt) + ) + magentic_context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=await prompt_template.render(Kernel(), KernelArguments(task=magentic_context.task.content)), + ) + ) + facts = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + self.prompt_execution_settings, + ) + assert facts is not None # nosec B101 + magentic_context.chat_history.add_message(facts) + + # 2. Create the plan + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.task_ledger_plan_prompt) + ) + magentic_context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=await prompt_template.render( + Kernel(), + KernelArguments(team=magentic_context.participant_descriptions), + ), + ) + ) + plan = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + self.prompt_execution_settings, + ) + assert plan is not None # nosec B101 + + self.task_ledger = _TaskLedger(facts=facts, plan=plan) + return await self._render_task_ledger(magentic_context) + + @override + async def replan(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Replan the task. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The updated task ledger. + """ + if self.task_ledger is None: + raise RuntimeError("The task ledger is not initialized. Planning needs to happen first.") + + # 1. Update the facts + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.task_ledger_facts_update_prompt) + ) + magentic_context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=await prompt_template.render( + Kernel(), + KernelArguments(task=magentic_context.task.content, old_facts=self.task_ledger.facts.content), + ), + ) + ) + facts = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + self.prompt_execution_settings, + ) + assert facts is not None # nosec B101 + magentic_context.chat_history.add_message(facts) + + # 2. Update the plan + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.task_ledger_plan_update_prompt) + ) + magentic_context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=await prompt_template.render( + Kernel(), + KernelArguments(team=magentic_context.participant_descriptions), + ), + ) + ) + plan = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + self.prompt_execution_settings, + ) + assert plan is not None # nosec B101 + + self.task_ledger.facts = facts + self.task_ledger.plan = plan + return await self._render_task_ledger(magentic_context) + + async def _render_task_ledger(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Render the task ledger to a string. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The rendered task ledger. + """ + if self.task_ledger is None: + raise RuntimeError("The task ledger is not initialized. Planning needs to happen first.") + + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.task_ledger_full_prompt) + ) + + rendered_task_ledger = await prompt_template.render( + Kernel(), + KernelArguments( + task=magentic_context.task.content, + team=magentic_context.participant_descriptions, + facts=self.task_ledger.facts.content, + plan=self.task_ledger.plan.content, + ), + ) + + return ChatMessageContent(role=AuthorRole.ASSISTANT, content=rendered_task_ledger) + + @override + async def create_progress_ledger(self, magentic_context: MagenticContext) -> ProgressLedger: + """Create a progress ledger. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ProgressLedger: The progress ledger. + """ + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.progress_ledger_prompt) + ) + progress_ledger_prompt = await prompt_template.render( + Kernel(), + KernelArguments( + task=magentic_context.task.content, + team=magentic_context.participant_descriptions, + names=", ".join(magentic_context.participant_descriptions.keys()), + ), + ) + magentic_context.chat_history.add_message( + ChatMessageContent(role=AuthorRole.USER, content=progress_ledger_prompt) + ) + + prompt_execution_settings_clone = PromptExecutionSettings.from_prompt_execution_settings( + self.prompt_execution_settings + ) + prompt_execution_settings_clone.update_from_prompt_execution_settings( + PromptExecutionSettings(extension_data={"response_format": ProgressLedger}) + ) + + response = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + prompt_execution_settings_clone, + ) + assert response is not None # nosec B101 + + return ProgressLedger.model_validate_json(response.content) + + @override + async def prepare_final_answer(self, magentic_context: MagenticContext) -> ChatMessageContent: + """Prepare the final answer. + + Args: + magentic_context (MagenticContext): The context for the Magentic manager. + + Returns: + ChatMessageContent: The final answer. + """ + prompt_template = KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(template=self.final_answer_prompt) + ) + magentic_context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=await prompt_template.render(Kernel(), KernelArguments(task=magentic_context.task)), + ) + ) + + response = await self.chat_completion_service.get_chat_message_content( + magentic_context.chat_history, + self.prompt_execution_settings, + ) + assert response is not None # nosec B101 + + return response + + +# endregion MagenticManager + +# region MagenticManagerActor + + +@experimental +class MagenticManagerActor(ActorBase): + """Actor for the Magentic One manager.""" + + def __init__( + self, + manager: MagenticManagerBase, + internal_topic_type: str, + participant_descriptions: dict[str, str], + result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None, + ) -> None: + """Initialize the Magentic One manager actor. + + Args: + manager (MagenticManagerBase): The Magentic One manager. + internal_topic_type (str): The internal topic type. + participant_descriptions (dict[str, str]): The participant descriptions. + result_callback (Callable | None): A callback function to handle the final answer. + """ + self._manager = manager + self._internal_topic_type = internal_topic_type + self._result_callback = result_callback + self._participant_descriptions = participant_descriptions + self._context: MagenticContext | None = None + self._task_ledger: ChatMessageContent | None = None + + super().__init__(description="Magentic One Manager") + + @message_handler + async def _handle_start_message(self, message: MagenticStartMessage, ctx: MessageContext) -> None: + """Handle the start message for the Magentic One manager.""" + logger.debug(f"{self.id}: Received Magentic One start message.") + + self._context = MagenticContext( + task=message.body, + participant_descriptions=self._participant_descriptions, + ) + + # Initial planning + self._task_ledger = await self._manager.plan(self._context.model_copy(deep=True)) + + await self._run_outer_loop(ctx.cancellation_token) + + @message_handler + async def _handle_response_message(self, message: MagenticResponseMessage, ctx: MessageContext) -> None: + """Handle the response message for the Magentic One manager.""" + if self._context is None or self._task_ledger is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + if message.body.role != AuthorRole.USER: + self._context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=f"Transferred to {message.body.name}", + ) + ) + self._context.chat_history.add_message(message.body) + + logger.debug(f"{self.id}: Running inner loop.") + await self._run_inner_loop(ctx.cancellation_token) + + async def _run_outer_loop(self, cancellation_token: CancellationToken) -> None: + if self._context is None or self._task_ledger is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + # 1. Publish the rendered task ledger to the group chat. + # Need to add the task ledger to the orchestrator's chat history + # since the publisher won't receive the message it sends even though + # the publisher also subscribes to the topic. + self._context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.ASSISTANT, + content=self._task_ledger.content, + name=self.__class__.__name__, + ) + ) + + logger.debug(f"Initial task ledger:\n{self._task_ledger.content}") + await self.publish_message( + MagenticResponseMessage( + body=self._context.chat_history.messages[-1], + ), + TopicId(self._internal_topic_type, self.id.key), + cancellation_token=cancellation_token, + ) + + # 2. Start the inner loop. + await self._run_inner_loop(cancellation_token) + + async def _run_inner_loop(self, cancellation_token: CancellationToken) -> None: + if self._context is None or self._task_ledger is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + within_limits = await self._check_within_limits() + if not within_limits: + return + self._context.round_count += 1 + + # 1. Create a progress ledger + current_progress_ledger = await self._manager.create_progress_ledger(self._context.model_copy(deep=True)) + logger.debug(f"Current progress ledger:\n{current_progress_ledger.model_dump_json(indent=2)}") + + # 2. Process the progress ledger + # 2.1 Check for task completion + if current_progress_ledger.is_request_satisfied.answer: + logger.debug("Task completed.") + await self._prepare_final_answer() + return + # 2.2 Check for stalling or looping + if not current_progress_ledger.is_progress_being_made.answer or current_progress_ledger.is_in_loop.answer: + self._context.stall_count += 1 + else: + self._context.stall_count = max(0, self._context.stall_count - 1) + + if self._context.stall_count > self._manager.max_stall_count: + logger.debug("Stalling detected. Resetting the task.") + self._task_ledger = await self._manager.replan(self._context.model_copy(deep=True)) + await self._reset_for_outer_loop(cancellation_token) + logger.debug("Restarting outer loop.") + await self._run_outer_loop(cancellation_token) + return + + # 2.3 Publish for next step + next_step = current_progress_ledger.instruction_or_question.answer + self._context.chat_history.add_message( + ChatMessageContent( + role=AuthorRole.ASSISTANT, + content=next_step if isinstance(next_step, str) else str(next_step), + name=self.__class__.__name__, + ) + ) + await self.publish_message( + MagenticResponseMessage( + body=self._context.chat_history.messages[-1], + ), + TopicId(self._internal_topic_type, self.id.key), + cancellation_token=cancellation_token, + ) + + # 2.4 Request the next speaker to speak + next_speaker = current_progress_ledger.next_speaker.answer + if next_speaker not in self._participant_descriptions: + raise ValueError(f"Unknown speaker: {next_speaker}") + + logger.debug(f"Magentic One manager selected agent: {next_speaker}") + + await self.publish_message( + MagenticRequestMessage(agent_name=next_speaker), + TopicId(self._internal_topic_type, self.id.key), + cancellation_token=cancellation_token, + ) + + async def _reset_for_outer_loop(self, cancellation_token: CancellationToken) -> None: + """Reset the context for the outer loop.""" + if self._context is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + await self.publish_message( + MagenticResetMessage(), + TopicId(self._internal_topic_type, self.id.key), + cancellation_token=cancellation_token, + ) + self._context.reset() + + async def _prepare_final_answer(self) -> None: + """Prepare the final answer and send it to the result callback.""" + if self._context is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + final_answer = await self._manager.prepare_final_answer(self._context.model_copy(deep=True)) + + if self._result_callback: + await self._result_callback(final_answer) + + async def _check_within_limits(self) -> bool: + """Check if the manager is within the limits.""" + if self._context is None: + raise RuntimeError("The Magentic manager is not started yet. Make sure to send a start message first.") + + if ( + self._manager.max_round_count is not None and self._context.round_count >= self._manager.max_round_count + ) or (self._manager.max_reset_count is not None and self._context.reset_count > self._manager.max_reset_count): + message = ( + "Max round count reached." + if self._manager.max_round_count and self._context.round_count >= self._manager.max_round_count + else "Max reset count reached." + ) + logger.debug(message) + if self._result_callback: + await self._result_callback( + ChatMessageContent(role=AuthorRole.ASSISTANT, content=message, name=self.__class__.__name__) + ) + return False + + return True + + +# endregion MagenticManagerActor + +# region MagenticAgentActor + + +@experimental +class MagenticAgentActor(AgentActorBase): + """An agent actor that process messages in a Magentic One group chat.""" + + @message_handler + async def _handle_response_message(self, message: MagenticResponseMessage, ctx: MessageContext) -> None: + logger.debug(f"{self.id}: Received response message.") + if self._agent_thread is not None: + if message.body.role != AuthorRole.USER: + await self._agent_thread.on_new_message( + ChatMessageContent( + role=AuthorRole.USER, + content=f"Transferred to {message.body.name}", + ) + ) + await self._agent_thread.on_new_message(message.body) + else: + if message.body.role != AuthorRole.USER: + self._chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=f"Transferred to {message.body.name}", + ) + ) + self._chat_history.add_message(message.body) + + @message_handler + async def _handle_request_message(self, message: MagenticRequestMessage, ctx: MessageContext) -> None: + if message.agent_name != self._agent.name: + return + + logger.debug(f"{self.id}: Received request message.") + if self._agent_thread is None: + # Add a user message to steer the agent to respond more closely to the instructions. + self._chat_history.add_message( + ChatMessageContent( + role=AuthorRole.USER, + content=f"Transferred to {self._agent.name}, adopt the persona immediately.", + ) + ) + response_item = await self._agent.get_response(messages=self._chat_history.messages) # type: ignore[arg-type] + self._agent_thread = response_item.thread + else: + # Add a user message to steer the agent to respond more closely to the instructions. + new_message = ChatMessageContent( + role=AuthorRole.USER, + content=f"Transferred to {self._agent.name}, adopt the persona immediately.", + ) + response_item = await self._agent.get_response(messages=new_message, thread=self._agent_thread) + + logger.debug(f"{self.id} responded with {response_item.message.content}.") + await self._call_agent_response_callback(response_item.message) + + await self.publish_message( + MagenticResponseMessage(body=response_item.message), + TopicId(self._internal_topic_type, self.id.key), + cancellation_token=ctx.cancellation_token, + ) + + @message_handler + async def _handle_reset_message(self, message: MagenticResetMessage, ctx: MessageContext) -> None: + """Handle the reset message for the Magentic One group chat.""" + logger.debug(f"{self.id}: Received reset message.") + self._chat_history.clear() + if self._agent_thread: + await self._agent_thread.delete() + self._agent_thread = None + + +# endregion MagenticAgentActor + +# region MagenticOrchestration + + +@experimental +class MagenticOrchestration(OrchestrationBase[TIn, TOut]): + """The Magentic One pattern orchestration.""" + + def __init__( + self, + members: list[Agent], + manager: MagenticManagerBase, + name: str | None = None, + description: str | None = None, + input_transform: Callable[[TIn], Awaitable[DefaultTypeAlias] | DefaultTypeAlias] | None = None, + output_transform: Callable[[DefaultTypeAlias], Awaitable[TOut] | TOut] | None = None, + agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None, + ) -> None: + """Initialize the Magentic One orchestration. + + Args: + members (list[Agent]): A list of agents. + manager (MagenticManagerBase): The manager for the Magentic One pattern. + name (str | None): The name of the orchestration. + description (str | None): The description of the orchestration. + input_transform (Callable | None): A function that transforms the external input message. + output_transform (Callable | None): A function that transforms the internal output message. + agent_response_callback (Callable | None): A function that is called when a response is produced + by the agents. + """ + self._manager = manager + + for member in members: + if member.description is None: + raise ValueError("All members must have a description.") + + super().__init__( + members=members, + name=name, + description=description, + input_transform=input_transform, + output_transform=output_transform, + agent_response_callback=agent_response_callback, + ) + + @override + async def _start( + self, + task: DefaultTypeAlias, + runtime: CoreRuntime, + internal_topic_type: str, + cancellation_token: CancellationToken, + ) -> None: + """Start the Magentic pattern.""" + if not isinstance(task, ChatMessageContent): + # Magentic One only supports ChatMessageContent as input. + raise ValueError("The task must be a ChatMessageContent object.") + + target_actor_id = await runtime.get(self._get_manager_actor_type(internal_topic_type)) + await runtime.send_message( + MagenticStartMessage(body=task), + target_actor_id, + cancellation_token=cancellation_token, + ) + + @override + async def _prepare( + self, + runtime: CoreRuntime, + internal_topic_type: str, + result_callback: Callable[[DefaultTypeAlias], Awaitable[None]], + ) -> None: + """Register the actors and orchestrations with the runtime and add the required subscriptions.""" + await self._register_members(runtime, internal_topic_type) + await self._register_manager(runtime, internal_topic_type, result_callback=result_callback) + await self._add_subscriptions(runtime, internal_topic_type) + + async def _register_members(self, runtime: CoreRuntime, internal_topic_type: str) -> None: + """Register the agents.""" + await asyncio.gather(*[ + MagenticAgentActor.register( + runtime, + self._get_agent_actor_type(agent, internal_topic_type), + lambda agent=agent: MagenticAgentActor( # type: ignore[misc] + agent, + internal_topic_type, + self._agent_response_callback, + ), + ) + for agent in self._members + ]) + + async def _register_manager( + self, + runtime: CoreRuntime, + internal_topic_type: str, + result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None, + ) -> None: + """Register the group chat manager.""" + await MagenticManagerActor.register( + runtime, + self._get_manager_actor_type(internal_topic_type), + lambda: MagenticManagerActor( + self._manager, + internal_topic_type=internal_topic_type, + participant_descriptions={agent.name: agent.description for agent in self._members}, # type: ignore[misc] + result_callback=result_callback, + ), + ) + + async def _add_subscriptions(self, runtime: CoreRuntime, internal_topic_type: str) -> None: + subscriptions: list[TypeSubscription] = [] + for agent in self._members: + subscriptions.append( + TypeSubscription(internal_topic_type, self._get_agent_actor_type(agent, internal_topic_type)) + ) + subscriptions.append(TypeSubscription(internal_topic_type, self._get_manager_actor_type(internal_topic_type))) + + await asyncio.gather(*[runtime.add_subscription(sub) for sub in subscriptions]) + + def _get_agent_actor_type(self, agent: Agent, internal_topic_type: str) -> str: + """Get the actor type for an agent. + + The type is appended with the internal topic type to ensure uniqueness in the runtime + that may be shared by multiple orchestrations. + """ + return f"{agent.name}_{internal_topic_type}" + + def _get_manager_actor_type(self, internal_topic_type: str) -> str: + """Get the actor type for the group chat manager. + + The type is appended with the internal topic type to ensure uniqueness in the runtime + that may be shared by multiple orchestrations. + """ + return f"{MagenticManagerActor.__name__}_{internal_topic_type}" + + +# endregion MagenticOrchestration diff --git a/python/samples/getting_started_with_agents/multi_agent_orchestration/__init__.py b/python/semantic_kernel/agents/orchestration/prompts/__init__.py similarity index 100% rename from python/samples/getting_started_with_agents/multi_agent_orchestration/__init__.py rename to python/semantic_kernel/agents/orchestration/prompts/__init__.py diff --git a/python/semantic_kernel/agents/orchestration/prompts/_magentic_prompts.py b/python/semantic_kernel/agents/orchestration/prompts/_magentic_prompts.py new file mode 100644 index 000000000000..4cd79f80c1fe --- /dev/null +++ b/python/semantic_kernel/agents/orchestration/prompts/_magentic_prompts.py @@ -0,0 +1,150 @@ +# Copyright (c) Microsoft. All rights reserved. + +ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT = """Below I will present you a request. + +Before we begin addressing the request, please answer the following pre-survey to the best of your ability. +Keep in mind that you are Ken Jennings-level with trivia, and Mensa-level with puzzles, so there should be +a deep well to draw from. + +Here is the request: + +{{$task}} + +Here is the pre-survey: + + 1. Please list any specific facts or figures that are GIVEN in the request itself. It is possible that + there are none. + 2. Please list any facts that may need to be looked up, and WHERE SPECIFICALLY they might be found. + In some cases, authoritative sources are mentioned in the request itself. + 3. Please list any facts that may need to be derived (e.g., via logical deduction, simulation, or computation) + 4. Please list any facts that are recalled from memory, hunches, well-reasoned guesses, etc. + +When answering this survey, keep in mind that "facts" will typically be specific names, dates, statistics, etc. +Your answer should use headings: + + 1. GIVEN OR VERIFIED FACTS + 2. FACTS TO LOOK UP + 3. FACTS TO DERIVE + 4. EDUCATED GUESSES + +DO NOT include any other headings or sections in your response. DO NOT list next steps or plans until asked to do so. +""" + +ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT = """Fantastic. To address this request we have assembled the following team: + +{{$team}} + +Based on the team composition, and known and unknown facts, please devise a short bullet-point plan for addressing the +original request. Remember, there is no requirement to involve all team members -- a team member's particular expertise +may not be needed for this task. +""" + +ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT = """ +We are working to address the following user request: + +{{$task}} + + +To answer this request we have assembled the following team: + +{{$team}} + + +Here is an initial fact sheet to consider: + +{{$facts}} + + +Here is the plan to follow as best as possible: + +{{$plan}} +""" + +ORCHESTRATOR_PROGRESS_LEDGER_PROMPT = """ +Recall we are working on the following request: + +{{$task}} + +And we have assembled the following team: + +{{$team}} + +To make progress on the request, please answer the following questions, including necessary reasoning: + + - Is the request fully satisfied? (True if complete, or False if the original request has yet to be + SUCCESSFULLY and FULLY addressed) + - Are we in a loop where we are repeating the same requests and / or getting the same responses as before? + Loops can span multiple turns, and can include repeated actions like scrolling up or down more than a + handful of times. + - Are we making forward progress? (True if just starting, or recent messages are adding value. False if recent + messages show evidence of being stuck in a loop or if there is evidence of significant barriers to success + such as the inability to read from a required file) + - Who should speak next? (select from: {{$names}}) + - What instruction or question would you give this team member? (Phrase as if speaking directly to them, and + include any specific information they may need) + +Please output an answer in pure JSON format according to the following schema. The JSON object must be parsable as-is. +DO NOT OUTPUT ANYTHING OTHER THAN JSON, AND DO NOT DEVIATE FROM THIS SCHEMA: + +{ + "is_request_satisfied": { + "reason": string, + "answer": boolean + }, + "is_in_loop": { + "reason": string, + "answer": boolean + }, + "is_progress_being_made": { + "reason": string, + "answer": boolean + }, + "next_speaker": { + "reason": string, + "answer": string (select from: {{$names}}) + }, + "instruction_or_question": { + "reason": string, + "answer": string + } +} +""" + +ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT = """As a reminder, we are working to solve the following task: + +{{$task}} + +It's clear we aren't making as much progress as we would like, but we may have learned something new. +Please rewrite the following fact sheet, updating it to include anything new we have learned that may be helpful. + +Example edits can include (but are not limited to) adding new guesses, moving educated guesses to verified facts +if appropriate, etc. Updates may be made to any section of the fact sheet, and more than one section of the fact +sheet can be edited. This is an especially good time to update educated guesses, so please at least add or update +one educated guess or hunch, and explain your reasoning. + +Here is the old fact sheet: + +{{$old_facts}} +""" + + +ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT = """Please briefly explain what went wrong on this last run (the root +cause of the failure), and then come up with a new plan that takes steps and/or includes hints to overcome prior +challenges and especially avoids repeating the same mistakes. As before, the new plan should be concise, be expressed +in bullet-point form, and consider the following team composition (do not involve any other outside people since we +cannot contact anyone else): + +{{$team}} +""" + +ORCHESTRATOR_FINAL_ANSWER_PROMPT = """ +We are working on the following task: +{{$task}} + +We have completed the task. + +The above messages contain the conversation that took place to complete the task. + +Based on the information gathered, provide the final answer to the original request. +The answer should be phrased as if you were speaking to the user. +""" diff --git a/python/tests/unit/agents/orchestration/test_magentic.py b/python/tests/unit/agents/orchestration/test_magentic.py new file mode 100644 index 000000000000..bb416b327622 --- /dev/null +++ b/python/tests/unit/agents/orchestration/test_magentic.py @@ -0,0 +1,718 @@ +# Copyright (c) Microsoft. All rights reserved. + +import sys +from typing import Any, Literal +from unittest.mock import AsyncMock, patch + +import pytest +from pydantic import BaseModel + +from semantic_kernel.agents.orchestration.magentic import ( + MagenticContext, + MagenticOrchestration, + ProgressLedger, + ProgressLedgerItem, + StandardMagenticManager, +) +from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationResult +from semantic_kernel.agents.orchestration.prompts._magentic_prompts import ( + ORCHESTRATOR_FINAL_ANSWER_PROMPT, + ORCHESTRATOR_PROGRESS_LEDGER_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT, + ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT, + ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT, + ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT, +) +from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime +from semantic_kernel.connectors.ai.chat_completion_client_base import ChatCompletionClientBase +from semantic_kernel.connectors.ai.prompt_execution_settings import PromptExecutionSettings +from semantic_kernel.contents.chat_history import ChatHistory +from semantic_kernel.contents.chat_message_content import ChatMessageContent +from semantic_kernel.contents.utils.author_role import AuthorRole +from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime + + +class MockChatCompletionService(ChatCompletionClientBase): + """A mock chat completion service for testing purposes.""" + + pass + + +class MockPromptExecutionSettings(PromptExecutionSettings): + """A mock prompt execution settings class for testing purposes.""" + + response_format: ( + dict[Literal["type"], Literal["text", "json_object"]] | dict[str, Any] | type[BaseModel] | type | None + ) = None + + +# region MagenticOrchestration + + +async def test_init_member_without_description_throws(): + """Test the prepare method of the MagenticOrchestration with a member without description.""" + agent_a = MockAgent() + agent_b = MockAgent() + + with pytest.raises(ValueError): + MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + ), + ) + + +async def test_prepare(): + """Test the prepare method of the MagenticOrchestration.""" + agent_a = MockAgent(description="test agent") + agent_b = MockAgent(description="test agent") + + runtime = MockRuntime() + + package_path = "semantic_kernel.agents.orchestration.magentic" + with ( + patch(f"{package_path}.MagenticOrchestration._start"), + patch(f"{package_path}.MagenticAgentActor.register") as mock_agent_actor_register, + patch(f"{package_path}.MagenticManagerActor.register") as mock_manager_actor_register, + patch.object(runtime, "add_subscription") as mock_add_subscription, + ): + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + ), + ) + await orchestration.invoke(task="test_message", runtime=runtime) + + assert mock_agent_actor_register.call_count == 2 + assert mock_manager_actor_register.call_count == 1 + assert mock_add_subscription.call_count == 3 + + +ManagerProgressList = [ + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_a", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_b", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="N/A", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), +] + +ManagerProgressListStalling = [ + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_a", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=True, reason="mock_reasoning"), # is_in_loop=True + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_a", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=True, reason="mock_reasoning"), # is_in_loop=True + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="N/A", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_b", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="N/A", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), +] + +ManagerProgressListUnknownSpeaker = [ + ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=True, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="unknown", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ), +] + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke(): + """Test the invoke method of the MagenticOrchestration.""" + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response, + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, "create_progress_ledger", new_callable=AsyncMock, side_effect=ManagerProgressList + ), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + runtime = InProcessRuntime() + runtime.start() + + try: + orchestration = MagenticOrchestration(members=[agent_a, agent_b], manager=manager) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + result = await orchestration_result.get() + finally: + await runtime.stop_when_idle() + + assert isinstance(orchestration_result, OrchestrationResult) + assert isinstance(result, ChatMessageContent) + assert result.role == AuthorRole.ASSISTANT + assert result.content == "mock_response" + + assert mock_get_response.call_count == 2 + assert mock_get_chat_message_content.call_count == 3 + + +async def test_invoke_with_list_error(): + """Test the invoke method of the MagenticOrchestration with a list of messages which raises an error.""" + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + messages = [ + ChatMessageContent(role=AuthorRole.USER, content="test_message_1"), + ChatMessageContent(role=AuthorRole.USER, content="test_message_2"), + ] + + runtime = MockRuntime() + + package_path = "semantic_kernel.agents.orchestration.magentic" + with ( + patch(f"{package_path}.MagenticAgentActor.register"), + patch(f"{package_path}.MagenticManagerActor.register"), + patch.object(runtime, "add_subscription"), + pytest.raises(ValueError), + ): + orchestration = MagenticOrchestration(members=[agent_a, agent_b], manager=manager) + orchestration_result = await orchestration.invoke(task=messages, runtime=runtime) + await orchestration_result.get() + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke_with_response_callback(): + """Test the invoke method of the MagenticOrchestration with a response callback.""" + + runtime = InProcessRuntime() + runtime.start() + + responses: list[DefaultTypeAlias] = [] + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True), + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, "create_progress_ledger", new_callable=AsyncMock, side_effect=ManagerProgressList + ), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + try: + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + ), + agent_response_callback=lambda x: responses.append(x), + ) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + await orchestration_result.get(1.0) + finally: + await runtime.stop_when_idle() + + assert len(responses) == 2 + assert all(isinstance(item, ChatMessageContent) for item in responses) + assert all(item.content == "mock_response" for item in responses) + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke_with_max_stall_count_exceeded(): + """ "Test the invoke method of the MagenticOrchestration with max stall count exceeded.""" + runtime = InProcessRuntime() + runtime.start() + + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response, + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, + "create_progress_ledger", + new_callable=AsyncMock, + side_effect=ManagerProgressListStalling, + ), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + try: + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + max_stall_count=1, + ), + ) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + await orchestration_result.get(1.0) + finally: + await runtime.stop_when_idle() + + assert mock_get_response.call_count == 3 + # Exceeding max stall count will trigger replanning, which will recreate the facts and plan, + # resulting in two additional calls to get_chat_message_content compared to the `test_invoke` test. + assert mock_get_chat_message_content.call_count == 5 + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke_with_max_round_count_exceeded(): + """ "Test the invoke method of the MagenticOrchestration with max round count exceeded.""" + runtime = InProcessRuntime() + runtime.start() + + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response, + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, + "create_progress_ledger", + new_callable=AsyncMock, + side_effect=ManagerProgressListStalling, + ), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + try: + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + max_round_count=1, + ), + ) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + result = await orchestration_result.get(1.0) + finally: + await runtime.stop_when_idle() + + assert result.content == "Max round count reached." + assert mock_get_response.call_count == 1 + # Planning will be called once, so the facts and plan will be created once. + assert mock_get_chat_message_content.call_count == 2 + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke_with_max_reset_count_exceeded(): + """ "Test the invoke method of the MagenticOrchestration with max reset count exceeded.""" + runtime = InProcessRuntime() + runtime.start() + + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response, + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, + "create_progress_ledger", + new_callable=AsyncMock, + side_effect=ManagerProgressListStalling, + ), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + try: + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + max_stall_count=0, # No stall allowed + max_reset_count=0, # No reset allowed + ), + ) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + result = await orchestration_result.get(1.0) + finally: + await runtime.stop_when_idle() + + assert result.content == "Max reset count reached." + assert mock_get_response.call_count == 1 + # Planning and replanning will be each called once, so the facts and plan will be created twice. + assert mock_get_chat_message_content.call_count == 4 + + +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Python 3.10 doesn't bound the original function provided to the wraps argument of the patch object.", +) +async def test_invoke_with_unknown_speaker(): + """Test the invoke method of the MagenticOrchestration with an unknown speaker.""" + runtime = InProcessRuntime() + runtime.start() + + with ( + patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True), + patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content, + patch.object( + StandardMagenticManager, + "create_progress_ledger", + new_callable=AsyncMock, + side_effect=ManagerProgressListUnknownSpeaker, + ), + pytest.raises(ValueError), + ): + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + agent_a = MockAgent(name="agent_a", description="test agent") + agent_b = MockAgent(name="agent_b", description="test agent") + + try: + orchestration = MagenticOrchestration( + members=[agent_a, agent_b], + manager=StandardMagenticManager( + chat_completion_service=MockChatCompletionService(ai_model_id="test"), + prompt_execution_settings=MockPromptExecutionSettings(), + ), + ) + orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime) + await orchestration_result.get() + finally: + await runtime.stop_when_idle() + + +# endregion MagenticOrchestration + +# region StandardMagenticManager + + +def test_standard_magentic_manager_init(): + """Test the initialization of the StandardMagenticManager.""" + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + assert manager.max_stall_count > 0 + assert manager.max_reset_count is None + assert manager.max_round_count is None + assert ( + manager.task_ledger_facts_prompt is not None + and manager.task_ledger_facts_prompt == ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT + ) + assert ( + manager.task_ledger_plan_prompt is not None + and manager.task_ledger_plan_prompt == ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT + ) + assert ( + manager.task_ledger_full_prompt is not None + and manager.task_ledger_full_prompt == ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT + ) + assert ( + manager.task_ledger_facts_update_prompt is not None + and manager.task_ledger_facts_update_prompt == ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT + ) + assert ( + manager.task_ledger_plan_update_prompt is not None + and manager.task_ledger_plan_update_prompt == ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT + ) + assert ( + manager.progress_ledger_prompt is not None + and manager.progress_ledger_prompt == ORCHESTRATOR_PROGRESS_LEDGER_PROMPT + ) + assert manager.final_answer_prompt is not None and manager.final_answer_prompt == ORCHESTRATOR_FINAL_ANSWER_PROMPT + + +def test_standard_magentic_manager_init_with_custom_prompts(): + """Test the initialization of the StandardMagenticManager with custom prompts.""" + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + task_ledger_facts_prompt="custom_task_ledger_facts_prompt", + task_ledger_plan_prompt="custom_task_ledger_plan_prompt", + task_ledger_full_prompt="custom_task_ledger_full_prompt", + task_ledger_facts_update_prompt="custom_task_ledger_facts_update_prompt", + task_ledger_plan_update_prompt="custom_task_ledger_plan_update_prompt", + progress_ledger_prompt="custom_progress_ledger_prompt", + final_answer_prompt="custom_final_answer_prompt", + ) + + assert manager.task_ledger_facts_prompt == "custom_task_ledger_facts_prompt" + assert manager.task_ledger_plan_prompt == "custom_task_ledger_plan_prompt" + assert manager.task_ledger_full_prompt == "custom_task_ledger_full_prompt" + assert manager.task_ledger_facts_update_prompt == "custom_task_ledger_facts_update_prompt" + assert manager.task_ledger_plan_update_prompt == "custom_task_ledger_plan_update_prompt" + assert manager.progress_ledger_prompt == "custom_progress_ledger_prompt" + assert manager.final_answer_prompt == "custom_final_answer_prompt" + + +def test_standard_magentic_manager_init_with_invalid_prompt_execution_settings(): + """Test the initialization of the StandardMagenticManager with invalid prompt execution settings.""" + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = PromptExecutionSettings() + + with pytest.raises(ValueError): + StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + +def test_standard_magentic_manager_init_without_prompt_execution_settings(): + """Test the initialization of the StandardMagenticManager without prompt execution settings.""" + # The default prompt execution settings of the mock chat completion service + # does not support structured output. + chat_completion_service = MockChatCompletionService(ai_model_id="test") + + with pytest.raises(ValueError): + StandardMagenticManager(chat_completion_service=chat_completion_service) + + +async def test_standard_magentic_manager_plan(): + """Test the plan method of the StandardMagenticManager.""" + + with patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content: + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + task_ledger_facts_prompt="custom_task_ledger_facts_prompt", + task_ledger_plan_prompt="custom_task_ledger_plan_prompt {{$team}}", + ) + + magentic_context = MagenticContext( + chat_history=ChatHistory(), + task=ChatMessageContent(role="user", content="test_message"), + participant_descriptions={"agent_a": "test_agent_a", "agent_b": "test_agent_b"}, + ) + + task_ledger = await manager.plan(magentic_context.model_copy(deep=True)) + + assert isinstance(task_ledger, ChatMessageContent) + assert task_ledger.content.count("mock_response") == 2 + assert "test_message" in task_ledger.content + assert "{'agent_a': 'test_agent_a', 'agent_b': 'test_agent_b'}" in task_ledger.content + + assert mock_get_chat_message_content.call_count == 2 + assert ( + mock_get_chat_message_content.call_args_list[0][0][0].messages[0].content + == "custom_task_ledger_facts_prompt" + ) + assert ( + mock_get_chat_message_content.call_args_list[1][0][0].messages[2].content + == "custom_task_ledger_plan_prompt {'agent_a': 'test_agent_a', 'agent_b': 'test_agent_b'}" + ) + + +async def test_standard_magentic_manager_replan(): + """Test the replan method of the StandardMagenticManager.""" + + with patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content: + mock_get_chat_message_content.return_value = ChatMessageContent(role="assistant", content="mock_response") + + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + task_ledger_facts_update_prompt="custom_task_ledger_facts_prompt {{$old_facts}}", + task_ledger_plan_update_prompt="custom_task_ledger_plan_prompt {{$team}}", + ) + + magentic_context = MagenticContext( + chat_history=ChatHistory(), + task=ChatMessageContent(role="user", content="test_message"), + participant_descriptions={"agent_a": "test_agent_a", "agent_b": "test_agent_b"}, + ) + + # Need to plan before replanning + _ = await manager.plan(magentic_context.model_copy(deep=True)) + task_ledger = await manager.replan(magentic_context.model_copy(deep=True)) + + assert isinstance(task_ledger, ChatMessageContent) + assert task_ledger.content.count("mock_response") == 2 + assert "test_message" in task_ledger.content + assert "{'agent_a': 'test_agent_a', 'agent_b': 'test_agent_b'}" in task_ledger.content + + assert mock_get_chat_message_content.call_count == 4 + assert ( + mock_get_chat_message_content.call_args_list[2][0][0].messages[0].content + == "custom_task_ledger_facts_prompt mock_response" + ) + assert ( + mock_get_chat_message_content.call_args_list[3][0][0].messages[2].content + == "custom_task_ledger_plan_prompt {'agent_a': 'test_agent_a', 'agent_b': 'test_agent_b'}" + ) + + +async def test_standard_magentic_manager_replan_without_plan(): + """Test the replan method of the StandardMagenticManager.""" + + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + magentic_context = MagenticContext( + chat_history=ChatHistory(), + task=ChatMessageContent(role="user", content="test_message"), + participant_descriptions={"agent_a": "test_agent_a", "agent_b": "test_agent_b"}, + ) + + with pytest.raises(RuntimeError): + _ = await manager.replan(magentic_context.model_copy(deep=True)) + + +async def test_standard_magentic_manager_create_progress_ledger(): + """Test the create_progress_ledger method of the StandardMagenticManager.""" + + mock_progress_ledger = ProgressLedger( + is_request_satisfied=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_in_loop=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + is_progress_being_made=ProgressLedgerItem(answer=False, reason="mock_reasoning"), + next_speaker=ProgressLedgerItem(answer="agent_a", reason="mock_reasoning"), + instruction_or_question=ProgressLedgerItem(answer="mock_instruction", reason="mock_reasoning"), + ) + + with patch.object( + MockChatCompletionService, "get_chat_message_content", new_callable=AsyncMock + ) as mock_get_chat_message_content: + mock_get_chat_message_content.return_value = ChatMessageContent( + role="assistant", content=mock_progress_ledger.model_dump_json() + ) + + chat_completion_service = MockChatCompletionService(ai_model_id="test") + prompt_execution_settings = MockPromptExecutionSettings() + + manager = StandardMagenticManager( + chat_completion_service=chat_completion_service, + prompt_execution_settings=prompt_execution_settings, + ) + + magentic_context = MagenticContext( + chat_history=ChatHistory(), + task=ChatMessageContent(role="user", content="test_message"), + participant_descriptions={"agent_a": "test_agent_a", "agent_b": "test_agent_b"}, + ) + + progress_ledger = await manager.create_progress_ledger(magentic_context.model_copy(deep=True)) + + assert isinstance(progress_ledger, ProgressLedger) + assert progress_ledger == mock_progress_ledger + + assert ( + "{'agent_a': 'test_agent_a', 'agent_b': 'test_agent_b'}" + in mock_get_chat_message_content.call_args_list[0][0][0].messages[0].content + ) + assert "agent_a, agent_b" in mock_get_chat_message_content.call_args_list[0][0][0].messages[0].content + assert ( + magentic_context.task.content in mock_get_chat_message_content.call_args_list[0][0][0].messages[0].content + ) + assert mock_get_chat_message_content.call_args_list[0][0][1].extension_data["response_format"] == ProgressLedger + + +# endregion MagenticManager