mirror of
https://github.com/microsoft/semantic-kernel.git
synced 2026-10-12 02:19:06 +00:00
Python: Multi-agent orchestration: Concurrent + Sequential (#11993)
### Motivation and Context <!-- Thank you for your contribution to the semantic-kernel repo! Please help reviewers and future users, providing the following information: 1. Why is this change required? 2. What problem does it solve? 3. What scenario does it contribute to? 4. If it fixes an open issue, please link to the issue here. --> Initial PR for multi-agent orchestration. ADR: https://github.com/microsoft/semantic-kernel/blob/main/docs/decisions/0071-multi-agent-orchestration.md ### Description This is the initial PR for multi-agent orchestration. This PR includes the following: 1. Base classes 2. Implementations for the concurrent and sequential orchestrations 3. Unit tests and samples Follow up PRs will add the following: - Group chat - Handoff - Magentic One <!-- Describe your changes, the overall approach, the underlying design. These notes will help understanding how your code works. Thanks! --> ### Contribution Checklist <!-- Before submitting this PR, please make sure: --> - [x] The code builds clean without any errors or warnings - [x] The PR follows the [SK Contribution Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md) and the [pre-submission formatting script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts) raises no violations - [x] All unit tests pass, and I have added new tests where possible - [x] I didn't break anyone 😄 --------- Co-authored-by: Evan Mattson <[email protected]>
This commit is contained in:
Tao Chen
+107
@@ -0,0 +1,107 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
|
||||
from semantic_kernel.agents import Agent, ChatCompletionAgent, ConcurrentOrchestration
|
||||
from semantic_kernel.agents.runtime import InProcessRuntime
|
||||
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
|
||||
|
||||
"""
|
||||
The following sample demonstrates how to create a concurrent orchestration for
|
||||
executing multiple agents on the same task in parallel.
|
||||
|
||||
This sample demonstrates the basic steps of creating and starting a runtime, creating
|
||||
a concurrent orchestration with multiple agents, invoking the orchestration, and finally
|
||||
waiting for the results.
|
||||
"""
|
||||
|
||||
|
||||
def get_agents() -> list[Agent]:
|
||||
"""Return a list of agents that will participate in the concurrent orchestration.
|
||||
|
||||
Feel free to add or remove agents.
|
||||
"""
|
||||
physics_agent = ChatCompletionAgent(
|
||||
name="PhysicsExpert",
|
||||
instructions="You are an expert in physics. You answer questions from a physics perspective.",
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
chemistry_agent = ChatCompletionAgent(
|
||||
name="ChemistryExpert",
|
||||
instructions="You are an expert in chemistry. You answer questions from a chemistry perspective.",
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
|
||||
return [physics_agent, chemistry_agent]
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main function to run the agents."""
|
||||
# 1. Create a concurrent orchestration with multiple agents
|
||||
agents = get_agents()
|
||||
concurrent_orchestration = ConcurrentOrchestration(members=agents)
|
||||
|
||||
# 2. Create a runtime and start it
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
# 3. Invoke the orchestration with a task and the runtime
|
||||
orchestration_result = await concurrent_orchestration.invoke(
|
||||
task="What is temperature?",
|
||||
runtime=runtime,
|
||||
)
|
||||
|
||||
# 4. Wait for the results
|
||||
# Note: the order of the results is not guaranteed to be the same
|
||||
# as the order of the agents in the orchestration.
|
||||
value = await orchestration_result.get(timeout=20)
|
||||
for item in value:
|
||||
print(f"# {item.name}: {item.content}")
|
||||
|
||||
# 5. Stop the runtime after the invocation is complete
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
"""
|
||||
Sample output:
|
||||
# PhysicsExpert: Temperature is a physical quantity that represents the average kinetic energy of the particles in
|
||||
a substance. It is an indicator of how hot or cold an object is and determines the direction of heat transfer
|
||||
between two objects. Heat flows from a region of higher temperature to a region of lower temperature until
|
||||
thermal equilibrium is reached.
|
||||
|
||||
In terms of molecular dynamics, at higher temperatures, particles move more vigorously and have higher kinetic
|
||||
energy, whereas at lower temperatures, their motion is less energetic. Temperature scales such as Celsius,
|
||||
Fahrenheit, and Kelvin are used to quantify temperature. The Kelvin scale is particularly important in
|
||||
scientific contexts because it starts at absolute zero—the theoretical point where particle motion would cease
|
||||
completely.
|
||||
|
||||
Temperature also affects various physical properties of materials, such as their state (solid, liquid, or gas),
|
||||
density, viscosity, and electrical conductivity. It is a crucial parameter in many areas of physics, from
|
||||
thermodynamics to statistical mechanics and beyond.
|
||||
# ChemistryExpert: Temperature is a fundamental concept in chemistry and physics, representing a measure of the
|
||||
average kinetic energy of the particles in a substance. It reflects how hot or cold an object is and determines
|
||||
the direction of heat transfer between substances. In more specific terms:
|
||||
|
||||
1. **Kinetic Energy Perspective:** At the molecular level, temperature is linked to the motions of the particles
|
||||
comprising a substance. The greater the motion (translational, rotational, vibrational), the higher the
|
||||
temperature. For example, in gases, temperature is directly related to the average kinetic energy of the gas
|
||||
particles.
|
||||
|
||||
2. **Thermodynamic View:** Temperature is an intensive property and a state function, meaning it doesn't depend
|
||||
on the amount of substance present. It is a critical parameter in the laws of thermodynamics, especially in
|
||||
determining the spontaneity of processes and the distribution of energy in a system.
|
||||
|
||||
3. **Scales:** Temperature is measured using various scales, including Celsius (°C), Fahrenheit (°F), and
|
||||
Kelvin (K). The Kelvin scale is the SI unit for temperature and starts at absolute zero (0 K), the theoretical
|
||||
point where all molecular motion ceases.
|
||||
|
||||
4. **Effect on Chemical Reactions:** Temperature affects reaction rates, equilibrium positions, and the
|
||||
solubility of substances. Generally, increasing temperature speeds up chemical reactions due to increased
|
||||
molecular collisions and energy overcoming activation barriers.
|
||||
|
||||
Understanding temperature is essential in predicting and explaining chemical behavior and interactions in
|
||||
reactions, phases changes, and even biological processes.
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+134
@@ -0,0 +1,134 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from semantic_kernel.agents import Agent, ChatCompletionAgent, ConcurrentOrchestration
|
||||
from semantic_kernel.agents.orchestration.tools import structured_outputs_transform
|
||||
from semantic_kernel.agents.runtime import InProcessRuntime
|
||||
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
|
||||
|
||||
"""
|
||||
The following sample demonstrates how to create a concurrent orchestration for
|
||||
executing multiple agents on the same task in parallel and returning a structured output.
|
||||
|
||||
This sample demonstrates the basic steps of creating and starting a runtime, creating
|
||||
a concurrent orchestration with multiple agents with a structure output transform,
|
||||
invoking the orchestration, and finally waiting for the results.
|
||||
"""
|
||||
|
||||
|
||||
class ArticleAnalysis(BaseModel):
|
||||
"""A model to hold the analysis of an article."""
|
||||
|
||||
themes: list[str]
|
||||
sentiments: list[str]
|
||||
entities: list[str]
|
||||
|
||||
|
||||
def get_agents() -> list[Agent]:
|
||||
"""Return a list of agents that will participate in the concurrent orchestration.
|
||||
|
||||
Feel free to add or remove agents.
|
||||
"""
|
||||
theme_agent = ChatCompletionAgent(
|
||||
name="ThemeAgent",
|
||||
instructions="You are an expert in identifying themes in articles. Given an article, identify the main themes.",
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
sentiment_agent = ChatCompletionAgent(
|
||||
name="SentimentAgent",
|
||||
instructions="You are an expert in sentiment analysis. Given an article, identify the sentiment.",
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
entity_agent = ChatCompletionAgent(
|
||||
name="EntityAgent",
|
||||
instructions="You are an expert in entity recognition. Given an article, extract the entities.",
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
|
||||
return [theme_agent, sentiment_agent, entity_agent]
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main function to run the agents."""
|
||||
# 1. Create a concurrent orchestration with multiple agents
|
||||
# and a structure output transform.
|
||||
# To enable structured output, you must specify the output transform
|
||||
# and the generic types for the orchestration.
|
||||
# Note: the chat completion service and model provided to the
|
||||
# structure output transform must support structured output.
|
||||
agents = get_agents()
|
||||
concurrent_orchestration = ConcurrentOrchestration[str, ArticleAnalysis](
|
||||
members=agents,
|
||||
output_transform=structured_outputs_transform(ArticleAnalysis, AzureChatCompletion()),
|
||||
)
|
||||
|
||||
# 2. Read the task from a file
|
||||
with open(os.path.join(os.path.dirname(__file__), "../resources", "Hamlet_full_play_summary.txt")) as file:
|
||||
task = file.read()
|
||||
|
||||
# 3. Create a runtime and start it
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
# 4. Invoke the orchestration with a task and the runtime
|
||||
orchestration_result = await concurrent_orchestration.invoke(
|
||||
task=task,
|
||||
runtime=runtime,
|
||||
)
|
||||
|
||||
# 5. Wait for the results
|
||||
value = await orchestration_result.get(timeout=20)
|
||||
if isinstance(value, ArticleAnalysis):
|
||||
print(value.model_dump_json(indent=2))
|
||||
else:
|
||||
print("Unexpected result type:", type(value))
|
||||
|
||||
# 6. Stop the runtime after the invocation is complete
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
"""
|
||||
Sample output:
|
||||
{
|
||||
"themes": [
|
||||
"Revenge and Justice",
|
||||
"Madness",
|
||||
"Corruption and Power",
|
||||
"Death and Mortality",
|
||||
"Appearance vs. Reality",
|
||||
"Family and Loyalty"
|
||||
],
|
||||
"sentiments": [
|
||||
"dark",
|
||||
"somber",
|
||||
"negative"
|
||||
],
|
||||
"entities": [
|
||||
"Elsinore Castle",
|
||||
"Denmark",
|
||||
"Horatio",
|
||||
"King Hamlet",
|
||||
"Claudius",
|
||||
"Queen Gertrude",
|
||||
"Prince Hamlet",
|
||||
"Rosencrantz",
|
||||
"Guildenstern",
|
||||
"Polonius",
|
||||
"Ophelia",
|
||||
"Laertes",
|
||||
"England",
|
||||
"King of England",
|
||||
"France",
|
||||
"Osric",
|
||||
"Fortinbras",
|
||||
"Poland"
|
||||
]
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+133
@@ -0,0 +1,133 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
|
||||
from semantic_kernel.agents import Agent, ChatCompletionAgent, SequentialOrchestration
|
||||
from semantic_kernel.agents.runtime import InProcessRuntime
|
||||
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
|
||||
from semantic_kernel.contents import ChatMessageContent
|
||||
|
||||
"""
|
||||
The following sample demonstrates how to create a sequential orchestration for
|
||||
executing multiple agents in sequence, i.e. the output of one agent is the input
|
||||
to the next agent.
|
||||
|
||||
This sample demonstrates the basic steps of creating and starting a runtime, creating
|
||||
a sequential orchestration, invoking the orchestration, and finally waiting for the
|
||||
results.
|
||||
"""
|
||||
|
||||
|
||||
def get_agents() -> list[Agent]:
|
||||
"""Return a list of agents that will participate in the sequential orchestration.
|
||||
|
||||
Feel free to add or remove agents.
|
||||
"""
|
||||
concept_extractor_agent = ChatCompletionAgent(
|
||||
name="ConceptExtractorAgent",
|
||||
instructions=(
|
||||
"You are a marketing analyst. Given a product description, identify:\n"
|
||||
"- Key features\n"
|
||||
"- Target audience\n"
|
||||
"- Unique selling points\n\n"
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
writer_agent = ChatCompletionAgent(
|
||||
name="WriterAgent",
|
||||
instructions=(
|
||||
"You are a marketing copywriter. Given a block of text describing features, audience, and USPs, "
|
||||
"compose a compelling marketing copy (like a newsletter section) that highlights these points. "
|
||||
"Output should be short (around 150 words), output just the copy as a single text block."
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
format_proof_agent = ChatCompletionAgent(
|
||||
name="FormatProofAgent",
|
||||
instructions=(
|
||||
"You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, "
|
||||
"give format and make it polished. Output the final improved copy as a single text block."
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
|
||||
# The order of the agents in the list will be the order in which they are executed
|
||||
return [concept_extractor_agent, writer_agent, format_proof_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 sequential orchestration with multiple agents and an agent
|
||||
# response callback to observe the output from each agent.
|
||||
agents = get_agents()
|
||||
sequential_orchestration = SequentialOrchestration(
|
||||
members=agents,
|
||||
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 sequential_orchestration.invoke(
|
||||
task="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours",
|
||||
runtime=runtime,
|
||||
)
|
||||
|
||||
# 4. Wait for the results
|
||||
value = await orchestration_result.get(timeout=20)
|
||||
print(f"***** Final Result *****\n{value}")
|
||||
|
||||
# 5. Stop the runtime when idle
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
"""
|
||||
Sample output:
|
||||
# ConceptExtractorAgent
|
||||
- Key Features:
|
||||
- Made of eco-friendly stainless steel
|
||||
- Keeps drinks cold for 24 hours
|
||||
|
||||
- Target Audience:
|
||||
- Environmentally conscious consumers
|
||||
- People who need a reliable way to keep their drinks cold for extended periods, such as athletes, travelers, and
|
||||
outdoor enthusiasts
|
||||
|
||||
- Unique Selling Points:
|
||||
- Environmentally sustainable material
|
||||
- Exceptionally long-lasting cold temperature retention (24 hours)
|
||||
# WriterAgent
|
||||
Keep your beverages refreshingly chilled all day long with our eco-friendly stainless steel bottles. Perfect for
|
||||
those who care about the planet, our bottles not only reduce waste but also promise to keep your drinks cold for
|
||||
an impressive 24 hours. Whether you're an athlete pushing your limits, a traveler on the go, or simply an outdoor
|
||||
enthusiast enjoying nature's beauty, this is the accessory you can't do without. Built from sustainable materials,
|
||||
our bottles ensure both environmental responsibility and remarkable performance. Stay refreshed, stay green, and
|
||||
make every sip a testament to your planet-friendly lifestyle. Join us in the journey towards a cooler, sustainable
|
||||
tomorrow.
|
||||
# FormatProofAgent
|
||||
Keep your beverages refreshingly chilled all day long with our eco-friendly stainless steel bottles. Perfect for
|
||||
those who care about the planet, our bottles not only reduce waste but also promise to keep your drinks cold for
|
||||
an impressive 24 hours. Whether you're an athlete pushing your limits, a traveler on the go, or simply an outdoor
|
||||
enthusiast enjoying nature's beauty, this is the accessory you can't do without. Built from sustainable materials,
|
||||
our bottles ensure both environmental responsibility and remarkable performance. Stay refreshed, stay green, and
|
||||
make every sip a testament to your planet-friendly lifestyle. Join us in the journey towards a cooler, sustainable
|
||||
tomorrow.
|
||||
***** Final Result *****
|
||||
Keep your beverages refreshingly chilled all day long with our eco-friendly stainless steel bottles. Perfect for
|
||||
those who care about the planet, our bottles not only reduce waste but also promise to keep your drinks cold for
|
||||
an impressive 24 hours. Whether you're an athlete pushing your limits, a traveler on the go, or simply an outdoor
|
||||
enthusiast enjoying nature's beauty, this is the accessory you can't do without. Built from sustainable materials,
|
||||
our bottles ensure both environmental responsibility and remarkable performance. Stay refreshed, stay green, and
|
||||
make every sip a testament to your planet-friendly lifestyle. Join us in the journey towards a cooler, sustainable
|
||||
tomorrow.
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
+109
@@ -0,0 +1,109 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from semantic_kernel.agents import Agent, ChatCompletionAgent, SequentialOrchestration
|
||||
from semantic_kernel.agents.runtime import InProcessRuntime
|
||||
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
|
||||
|
||||
"""
|
||||
The following sample demonstrates how to cancel an invocation of an orchestration
|
||||
that is still running.
|
||||
|
||||
This sample demonstrates the basic steps of creating and starting a runtime, creating
|
||||
a sequential orchestration, invoking the orchestration, and cancelling it before it
|
||||
finishes.
|
||||
"""
|
||||
|
||||
# Set up logging to see the invocation process
|
||||
logging.basicConfig(level=logging.WARNING) # Set default level to WARNING
|
||||
logging.getLogger("semantic_kernel.agents.orchestration.sequential").setLevel(logging.DEBUG)
|
||||
|
||||
|
||||
def get_agents() -> list[Agent]:
|
||||
"""Return a list of agents that will participate in the sequential orchestration.
|
||||
|
||||
Feel free to add or remove agents.
|
||||
"""
|
||||
concept_extractor_agent = ChatCompletionAgent(
|
||||
name="ConceptExtractorAgent",
|
||||
instructions=(
|
||||
"You are a marketing analyst. Given a product description, identify:\n"
|
||||
"- Key features\n"
|
||||
"- Target audience\n"
|
||||
"- Unique selling points\n\n"
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
writer_agent = ChatCompletionAgent(
|
||||
name="WriterAgent",
|
||||
instructions=(
|
||||
"You are a marketing copywriter. Given a block of text describing features, audience, and USPs, "
|
||||
"compose a compelling marketing copy (like a newsletter section) that highlights these points. "
|
||||
"Output should be short (around 150 words), output just the copy as a single text block."
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
format_proof_agent = ChatCompletionAgent(
|
||||
name="FormatProofAgent",
|
||||
instructions=(
|
||||
"You are an editor. Given the draft copy, correct grammar, improve clarity, ensure consistent tone, "
|
||||
"give format and make it polished. Output the final improved copy as a single text block."
|
||||
),
|
||||
service=AzureChatCompletion(),
|
||||
)
|
||||
|
||||
# The order of the agents in the list will be the order in which they are executed
|
||||
return [concept_extractor_agent, writer_agent, format_proof_agent]
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main function to run the agents."""
|
||||
# 1. Create a sequential orchestration with multiple agents
|
||||
agents = get_agents()
|
||||
sequential_orchestration = SequentialOrchestration(members=agents)
|
||||
|
||||
# 2. Create a runtime and start it
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
# 3. Invoke the orchestration with a task and the runtime
|
||||
orchestration_result = await sequential_orchestration.invoke(
|
||||
task="An eco-friendly stainless steel water bottle that keeps drinks cold for 24 hours",
|
||||
runtime=runtime,
|
||||
)
|
||||
|
||||
# 4. Cancel the orchestration before it finishes
|
||||
await asyncio.sleep(1) # Simulate some delay before cancellation
|
||||
orchestration_result.cancel()
|
||||
|
||||
try:
|
||||
# Attempt to get the result will result in an exception due to cancellation
|
||||
_ = await orchestration_result.get(timeout=20)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
finally:
|
||||
# 5. Stop the runtime
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
"""
|
||||
Sample output:
|
||||
DEBUG:semantic_kernel.agents.orchestration.sequential:Registered agent actor of type
|
||||
FormatProofAgent_5efa69d39306414c91325ef82145ec19
|
||||
DEBUG:semantic_kernel.agents.orchestration.sequential:Registered agent actor of type
|
||||
WriterAgent_5efa69d39306414c91325ef82145ec19
|
||||
DEBUG:semantic_kernel.agents.orchestration.sequential:Registered agent actor of type
|
||||
ConceptExtractorAgent_5efa69d39306414c91325ef82145ec19
|
||||
DEBUG:semantic_kernel.agents.orchestration.sequential:Sequential actor
|
||||
(Actor ID: ConceptExtractorAgent_5efa69d39306414c91325ef82145ec19/default; Agent name: ConceptExtractorAgent)
|
||||
started processing...
|
||||
The invocation was canceled before it could complete.
|
||||
DEBUG:semantic_kernel.agents.orchestration.sequential:Sequential actor
|
||||
(Actor ID: ConceptExtractorAgent_5efa69d39306414c91325ef82145ec19/default; Agent name: ConceptExtractorAgent)
|
||||
finished processing.
|
||||
"""
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -0,0 +1,13 @@
|
||||
On a dark winter night, a ghost walks the ramparts of Elsinore Castle in Denmark. Discovered first by a pair of watchmen, then by the scholar Horatio, the ghost resembles the recently deceased King Hamlet, whose brother Claudius has inherited the throne and married the king’s widow, Queen Gertrude. When Horatio and the watchmen bring Prince Hamlet, the son of Gertrude and the dead king, to see the ghost, it speaks to him, declaring ominously that it is indeed his father’s spirit, and that he was murdered by none other than Claudius. Ordering Hamlet to seek revenge on the man who usurped his throne and married his wife, the ghost disappears with the dawn.
|
||||
|
||||
Prince Hamlet devotes himself to avenging his father’s death, but, because he is contemplative and thoughtful by nature, he delays, entering into a deep melancholy and even apparent madness. Claudius and Gertrude worry about the prince’s erratic behavior and attempt to discover its cause. They employ a pair of Hamlet’s friends, Rosencrantz and Guildenstern, to watch him. When Polonius, the pompous Lord Chamberlain, suggests that Hamlet may be mad with love for his daughter, Ophelia, Claudius agrees to spy on Hamlet in conversation with the girl. But though Hamlet certainly seems mad, he does not seem to love Ophelia: he orders her to enter a nunnery and declares that he wishes to ban marriages.
|
||||
|
||||
A group of traveling actors comes to Elsinore, and Hamlet seizes upon an idea to test his uncle’s guilt. He will have the players perform a scene closely resembling the sequence by which Hamlet imagines his uncle to have murdered his father, so that if Claudius is guilty, he will surely react. When the moment of the murder arrives in the theater, Claudius leaps up and leaves the room. Hamlet and Horatio agree that this proves his guilt. Hamlet goes to kill Claudius but finds him praying. Since he believes that killing Claudius while in prayer would send Claudius’s soul to heaven, Hamlet considers that it would be an inadequate revenge and decides to wait. Claudius, now frightened of Hamlet’s madness and fearing for his own safety, orders that Hamlet be sent to England at once.
|
||||
|
||||
Hamlet goes to confront his mother, in whose bedchamber Polonius has hidden behind a tapestry. Hearing a noise from behind the tapestry, Hamlet believes the king is hiding there. He draws his sword and stabs through the fabric, killing Polonius. For this crime, he is immediately dispatched to England with Rosencrantz and Guildenstern. However, Claudius’s plan for Hamlet includes more than banishment, as he has given Rosencrantz and Guildenstern sealed orders for the King of England demanding that Hamlet be put to death.
|
||||
|
||||
In the aftermath of her father’s death, Ophelia goes mad with grief and drowns in the river. Polonius’s son, Laertes, who has been staying in France, returns to Denmark in a rage. Claudius convinces him that Hamlet is to blame for his father’s and sister’s deaths. When Horatio and the king receive letters from Hamlet indicating that the prince has returned to Denmark after pirates attacked his ship en route to England, Claudius concocts a plan to use Laertes’ desire for revenge to secure Hamlet’s death. Laertes will fence with Hamlet in innocent sport, but Claudius will poison Laertes’ blade so that if he draws blood, Hamlet will die. As a backup plan, the king decides to poison a goblet, which he will give Hamlet to drink should Hamlet score the first or second hits of the match. Hamlet returns to the vicinity of Elsinore just as Ophelia’s funeral is taking place. Stricken with grief, he attacks Laertes and declares that he had in fact always loved Ophelia. Back at the castle, he tells Horatio that he believes one must be prepared to die, since death can come at any moment. A foolish courtier named Osric arrives on Claudius’s orders to arrange the fencing match between Hamlet and Laertes.
|
||||
|
||||
The sword-fighting begins. Hamlet scores the first hit, but declines to drink from the king’s proffered goblet. Instead, Gertrude takes a drink from it and is swiftly killed by the poison. Laertes succeeds in wounding Hamlet, though Hamlet does not die of the poison immediately. First, Laertes is cut by his own sword’s blade, and, after revealing to Hamlet that Claudius is responsible for the queen’s death, he dies from the blade’s poison. Hamlet then stabs Claudius through with the poisoned sword and forces him to drink down the rest of the poisoned wine. Claudius dies, and Hamlet dies immediately after achieving his revenge.
|
||||
|
||||
At this moment, a Norwegian prince named Fortinbras, who has led an army to Denmark and attacked Poland earlier in the play, enters with ambassadors from England, who report that Rosencrantz and Guildenstern are dead. Fortinbras is stunned by the gruesome sight of the entire royal family lying sprawled on the floor dead. He moves to take power of the kingdom. Horatio, fulfilling Hamlet’s last request, tells him Hamlet’s tragic story. Fortinbras orders that Hamlet be carried away in a manner befitting a fallen soldier.
|
||||
@@ -28,6 +28,8 @@ _AGENTS = {
|
||||
"AzureResponsesAgent": ".open_ai.azure_responses_agent",
|
||||
"ResponsesAgentThread": ".open_ai.openai_responses_agent",
|
||||
"RunPollingOptions": ".open_ai.run_polling_options",
|
||||
"ConcurrentOrchestration": ".orchestration.concurrent",
|
||||
"SequentialOrchestration": ".orchestration.sequential",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -15,6 +15,8 @@ from .open_ai.azure_responses_agent import AzureResponsesAgent
|
||||
from .open_ai.open_ai_assistant_agent import AssistantAgentThread, OpenAIAssistantAgent
|
||||
from .open_ai.openai_responses_agent import OpenAIResponsesAgent, ResponsesAgentThread
|
||||
from .open_ai.run_polling_options import RunPollingOptions
|
||||
from .orchestration.concurrent import ConcurrentOrchestration
|
||||
from .orchestration.sequential import SequentialOrchestration
|
||||
|
||||
__all__ = [
|
||||
"Agent",
|
||||
@@ -34,6 +36,7 @@ __all__ = [
|
||||
"BedrockAgentThread",
|
||||
"ChatCompletionAgent",
|
||||
"ChatHistoryAgentThread",
|
||||
"ConcurrentOrchestration",
|
||||
"CopilotStudioAgent",
|
||||
"CopilotStudioAgentAuthMode",
|
||||
"CopilotStudioAgentSettings",
|
||||
@@ -42,4 +45,5 @@ __all__ = [
|
||||
"OpenAIResponsesAgent",
|
||||
"ResponsesAgentThread",
|
||||
"RunPollingOptions",
|
||||
"SequentialOrchestration",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
|
||||
import inspect
|
||||
import sys
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
|
||||
from semantic_kernel.agents.agent import Agent, AgentThread
|
||||
from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias
|
||||
from semantic_kernel.agents.runtime.core.message_context import MessageContext
|
||||
from semantic_kernel.agents.runtime.core.routed_agent import RoutedAgent
|
||||
from semantic_kernel.contents.chat_history import ChatHistory
|
||||
|
||||
if sys.version_info >= (3, 12):
|
||||
from typing import override # pragma: no cover
|
||||
else:
|
||||
from typing_extensions import override # pragma: no cover
|
||||
|
||||
|
||||
class ActorBase(RoutedAgent):
|
||||
"""A base class for actors running in the AgentRuntime."""
|
||||
|
||||
@override
|
||||
async def on_message_impl(self, message: Any, ctx: MessageContext) -> Any | None:
|
||||
"""Handle a message.
|
||||
|
||||
Stop the handling of the message if the cancellation token is cancelled.
|
||||
"""
|
||||
if ctx.cancellation_token.is_cancelled():
|
||||
return None
|
||||
|
||||
return await super().on_message_impl(message, ctx)
|
||||
|
||||
|
||||
class AgentActorBase(ActorBase):
|
||||
"""A agent actor for multi-agent orchestration running on Agent runtime."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
agent: Agent,
|
||||
internal_topic_type: str,
|
||||
agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
|
||||
) -> None:
|
||||
"""Initialize the agent container.
|
||||
|
||||
Args:
|
||||
agent (Agent): An agent to be run in the container.
|
||||
internal_topic_type (str): The topic type of the internal topic.
|
||||
agent_response_callback (Callable | None): A function that is called when a response is produced
|
||||
by the agents.
|
||||
"""
|
||||
self._agent = agent
|
||||
self._internal_topic_type = internal_topic_type
|
||||
self._agent_response_callback = agent_response_callback
|
||||
|
||||
self._agent_thread: AgentThread | None = None
|
||||
# Chat history to temporarily store messages before the agent thread is created
|
||||
self._chat_history = ChatHistory()
|
||||
|
||||
ActorBase.__init__(self, description=agent.description or "Semantic Kernel Agent")
|
||||
|
||||
async def _call_agent_response_callback(self, message: DefaultTypeAlias) -> None:
|
||||
"""Call the agent_response_callback function if it is set.
|
||||
|
||||
Args:
|
||||
message (DefaultTypeAlias): The message to be sent to the agent_response_callback.
|
||||
"""
|
||||
# TODO(@taochen): Support streaming
|
||||
if self._agent_response_callback:
|
||||
if inspect.iscoroutinefunction(self._agent_response_callback):
|
||||
await self._agent_response_callback(message)
|
||||
else:
|
||||
self._agent_response_callback(message)
|
||||
@@ -0,0 +1,228 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sys
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
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 (
|
||||
ChatMessageContent,
|
||||
DefaultTypeAlias,
|
||||
OrchestrationBase,
|
||||
TIn,
|
||||
TOut,
|
||||
)
|
||||
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.kernel_pydantic import KernelBaseModel
|
||||
|
||||
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__)
|
||||
|
||||
|
||||
class ConcurrentRequestMessage(KernelBaseModel):
|
||||
"""A request message type for concurrent agents."""
|
||||
|
||||
body: DefaultTypeAlias
|
||||
|
||||
|
||||
class ConcurrentResponseMessage(KernelBaseModel):
|
||||
"""A response message type for concurrent agents."""
|
||||
|
||||
body: ChatMessageContent
|
||||
|
||||
|
||||
class ConcurrentAgentActor(AgentActorBase):
|
||||
"""A agent actor for concurrent agents that process tasks."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
agent: Agent,
|
||||
internal_topic_type: str,
|
||||
collection_agent_type: str,
|
||||
agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
|
||||
) -> None:
|
||||
"""Initialize the agent actor.
|
||||
|
||||
Args:
|
||||
agent: The agent to be executed.
|
||||
internal_topic_type: The internal topic type for the actor.
|
||||
collection_agent_type: The collection agent type for the actor.
|
||||
agent_response_callback: A callback function to handle the response from the agent.
|
||||
"""
|
||||
self._collection_agent_type = collection_agent_type
|
||||
super().__init__(
|
||||
agent=agent,
|
||||
internal_topic_type=internal_topic_type,
|
||||
agent_response_callback=agent_response_callback,
|
||||
)
|
||||
|
||||
@message_handler
|
||||
async def _handle_message(self, message: ConcurrentRequestMessage, ctx: MessageContext) -> None:
|
||||
"""Handle a message."""
|
||||
logger.debug(f"Concurrent actor (Actor ID: {self.id}; Agent name: {self._agent.name}) started processing...")
|
||||
|
||||
response = await self._agent.get_response(
|
||||
messages=message.body, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
logger.debug(f"Concurrent actor (Actor ID: {self.id}; Agent name: {self._agent.name}) finished processing.")
|
||||
|
||||
await self._call_agent_response_callback(response.message)
|
||||
|
||||
target_actor_id = await self.runtime.get(self._collection_agent_type)
|
||||
await self.send_message(
|
||||
ConcurrentResponseMessage(body=response.message),
|
||||
target_actor_id,
|
||||
cancellation_token=ctx.cancellation_token,
|
||||
)
|
||||
|
||||
|
||||
class CollectionActor(ActorBase):
|
||||
"""A agent container for collecting results from concurrent agents."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
description: str,
|
||||
expected_answer_count: int,
|
||||
result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None,
|
||||
) -> None:
|
||||
"""Initialize the collection agent container."""
|
||||
self._expected_answer_count = expected_answer_count
|
||||
self._result_callback = result_callback
|
||||
self._results: list[ChatMessageContent] = []
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
super().__init__(description=description)
|
||||
|
||||
@message_handler
|
||||
async def _handle_message(self, message: ConcurrentResponseMessage, _: MessageContext) -> None:
|
||||
async with self._lock:
|
||||
self._results.append(message.body)
|
||||
|
||||
if len(self._results) == self._expected_answer_count:
|
||||
logger.debug(f"Collection actor (Actor ID: {self.id}) finished processing all responses.")
|
||||
if self._result_callback:
|
||||
await self._result_callback(self._results)
|
||||
|
||||
|
||||
class ConcurrentOrchestration(OrchestrationBase[TIn, TOut]):
|
||||
"""A concurrent multi-agent pattern orchestration."""
|
||||
|
||||
@override
|
||||
async def _start(
|
||||
self,
|
||||
task: DefaultTypeAlias,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
cancellation_token: CancellationToken,
|
||||
) -> None:
|
||||
"""Start the concurrent pattern."""
|
||||
await runtime.publish_message(
|
||||
ConcurrentRequestMessage(body=task),
|
||||
TopicId(internal_topic_type, self.__class__.__name__),
|
||||
cancellation_token=cancellation_token,
|
||||
)
|
||||
|
||||
@override
|
||||
async def _prepare(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None,
|
||||
) -> None:
|
||||
"""Register the actors and orchestrations with the runtime and add the required subscriptions."""
|
||||
await asyncio.gather(*[
|
||||
self._register_members(
|
||||
runtime,
|
||||
internal_topic_type,
|
||||
),
|
||||
self._register_collection_actor(
|
||||
runtime,
|
||||
internal_topic_type,
|
||||
result_callback=result_callback,
|
||||
),
|
||||
self._add_subscriptions(
|
||||
runtime,
|
||||
internal_topic_type,
|
||||
),
|
||||
])
|
||||
|
||||
async def _register_members(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
) -> None:
|
||||
"""Register the members."""
|
||||
|
||||
async def _internal_helper(agent: Agent) -> None:
|
||||
await ConcurrentAgentActor.register(
|
||||
runtime,
|
||||
self._get_agent_actor_type(agent, internal_topic_type),
|
||||
lambda agent=agent: ConcurrentAgentActor( # type: ignore[misc]
|
||||
agent,
|
||||
internal_topic_type,
|
||||
collection_agent_type=self._get_collection_actor_type(internal_topic_type),
|
||||
agent_response_callback=self._agent_response_callback,
|
||||
),
|
||||
)
|
||||
|
||||
await asyncio.gather(*[_internal_helper(agent) for agent in self._members])
|
||||
|
||||
async def _register_collection_actor(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
result_callback: Callable[[DefaultTypeAlias], Awaitable[None]] | None = None,
|
||||
) -> None:
|
||||
await CollectionActor.register(
|
||||
runtime,
|
||||
self._get_collection_actor_type(internal_topic_type),
|
||||
lambda: CollectionActor(
|
||||
description="An internal agent that is responsible for collection results",
|
||||
expected_answer_count=len(self._members),
|
||||
result_callback=result_callback,
|
||||
),
|
||||
)
|
||||
|
||||
async def _add_subscriptions(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
) -> None:
|
||||
await asyncio.gather(*[
|
||||
runtime.add_subscription(
|
||||
TypeSubscription(
|
||||
internal_topic_type,
|
||||
self._get_agent_actor_type(agent, internal_topic_type),
|
||||
)
|
||||
)
|
||||
for agent in self._members
|
||||
])
|
||||
|
||||
def _get_agent_actor_type(self, worker: Agent, internal_topic_type: str) -> str:
|
||||
"""Get the container 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"{worker.name}_{internal_topic_type}"
|
||||
|
||||
def _get_collection_actor_type(self, internal_topic_type: str) -> str:
|
||||
"""Get the collection agent type.
|
||||
|
||||
The type is appended with the internal topic type to ensure uniqueness in the runtime
|
||||
that may be shared by multiple orchestrations.
|
||||
"""
|
||||
return f"{CollectionActor.__name__}_{internal_topic_type}"
|
||||
@@ -0,0 +1,327 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
from abc import ABC, abstractmethod
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Generic, Union, get_args
|
||||
|
||||
from pydantic import Field
|
||||
from typing_extensions import TypeVar
|
||||
|
||||
from semantic_kernel.agents.agent import Agent
|
||||
from semantic_kernel.agents.runtime.core.cancellation_token import CancellationToken
|
||||
from semantic_kernel.agents.runtime.core.core_runtime import CoreRuntime
|
||||
from semantic_kernel.contents.chat_message_content import ChatMessageContent
|
||||
from semantic_kernel.contents.utils.author_role import AuthorRole
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DefaultTypeAlias = Union[ChatMessageContent, list[ChatMessageContent]]
|
||||
|
||||
TIn = TypeVar("TIn", default=DefaultTypeAlias)
|
||||
TOut = TypeVar("TOut", default=DefaultTypeAlias)
|
||||
|
||||
|
||||
class OrchestrationResult(KernelBaseModel, Generic[TOut]):
|
||||
"""The result of an invocation of an orchestration."""
|
||||
|
||||
value: TOut | None = None
|
||||
exception: BaseException | None = None
|
||||
event: asyncio.Event = Field(default_factory=asyncio.Event)
|
||||
cancellation_token: CancellationToken = Field(default_factory=CancellationToken)
|
||||
|
||||
async def get(self, timeout: float | None = None) -> TOut:
|
||||
"""Get the result of the invocation.
|
||||
|
||||
If a timeout is specified, the method will wait for the result for the specified time.
|
||||
If the result is not available within the timeout, a TimeoutError will be raised but the
|
||||
invocation will not be aborted.
|
||||
|
||||
Args:
|
||||
timeout (int | None): The timeout (seconds) for getting the result. If None, wait indefinitely.
|
||||
|
||||
Returns:
|
||||
TOut: The result of the invocation.
|
||||
"""
|
||||
if timeout is not None:
|
||||
await asyncio.wait_for(self.event.wait(), timeout=timeout)
|
||||
else:
|
||||
await self.event.wait()
|
||||
|
||||
if self.value is None:
|
||||
if self.cancellation_token.is_cancelled():
|
||||
raise RuntimeError("The invocation was canceled before it could complete.")
|
||||
if self.exception is not None:
|
||||
raise self.exception
|
||||
raise RuntimeError("The invocation did not produce a result.")
|
||||
return self.value
|
||||
|
||||
def cancel(self) -> None:
|
||||
"""Cancel the invocation.
|
||||
|
||||
This method will cancel the invocation.
|
||||
Actors that have received messages will continue to process them, but no new messages will be processed.
|
||||
"""
|
||||
if self.cancellation_token.is_cancelled():
|
||||
raise RuntimeError("The invocation has already been canceled.")
|
||||
if self.event.is_set():
|
||||
raise RuntimeError("The invocation has already been completed.")
|
||||
|
||||
self.cancellation_token.cancel()
|
||||
self.event.set()
|
||||
|
||||
|
||||
class OrchestrationBase(ABC, Generic[TIn, TOut]):
|
||||
"""Base class for multi-agent orchestration."""
|
||||
|
||||
t_in: type[TIn] | None = None
|
||||
t_out: type[TOut] | None = None
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
members: list[Agent],
|
||||
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 orchestration base.
|
||||
|
||||
Args:
|
||||
members (list[Agent]): The list of agents to be used.
|
||||
name (str | None): A unique name of the orchestration. If None, a unique name will be generated.
|
||||
description (str | None): The description of the orchestration. If None, use a default description.
|
||||
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.
|
||||
"""
|
||||
if not members:
|
||||
raise ValueError("The members list cannot be empty.")
|
||||
self._members = members
|
||||
|
||||
self.name = name or f"{self.__class__.__name__}_{uuid.uuid4().hex}"
|
||||
self.description = description or "A multi-agent orchestration."
|
||||
|
||||
self._input_transform = input_transform or self._default_input_transform
|
||||
self._output_transform = output_transform or self._default_output_transform
|
||||
|
||||
self._agent_response_callback = agent_response_callback
|
||||
|
||||
def _set_types(self) -> None:
|
||||
"""Set the external input and output types from the class arguments.
|
||||
|
||||
This method can only be run after the class has been initialized because it relies on the
|
||||
`__orig_class__` attributes to determine the type parameters.
|
||||
|
||||
This method will first try to get the type parameters from the class itself. The `__orig_class__`
|
||||
attribute will contain the external input and output types if they are explicitly given, for example:
|
||||
```
|
||||
class MyOrchestration(OrchestrationBase[TIn, TOut]):
|
||||
pass
|
||||
|
||||
|
||||
my_orchestration = MyOrchestration[str, str](...)
|
||||
```
|
||||
If the type parameters are not explicitly given, for example when the TypeVars has defaults, for example:
|
||||
```
|
||||
TIn = TypeVar("TIn", default=str)
|
||||
TOut = TypeVar("TOut", default=str)
|
||||
|
||||
|
||||
class MyOrchestration(OrchestrationBase[TIn, TOut]):
|
||||
pass
|
||||
|
||||
|
||||
my_orchestration = MyOrchestration(...)
|
||||
```
|
||||
The type parameters can be inferred from the `__orig_bases__` attribute.
|
||||
"""
|
||||
if all([self.t_in is not None, self.t_out is not None]):
|
||||
return
|
||||
|
||||
try:
|
||||
args = self.__orig_class__.__args__ # type: ignore[attr-defined]
|
||||
if len(args) == 1:
|
||||
self.t_in = args[0]
|
||||
self.t_out = DefaultTypeAlias # type: ignore[assignment]
|
||||
elif len(args) == 2:
|
||||
self.t_in = args[0]
|
||||
self.t_out = args[1]
|
||||
else:
|
||||
raise TypeError("Orchestration must have two type parameters.")
|
||||
except AttributeError:
|
||||
args = get_args(self.__orig_bases__[0]) # type: ignore[attr-defined]
|
||||
|
||||
if len(args) != 2:
|
||||
raise TypeError("Orchestration must be subclassed with two type parameters.")
|
||||
self.t_in = args[0] if isinstance(args[0], type) else getattr(args[0], "__default__", None) # type: ignore[assignment]
|
||||
self.t_out = args[1] if isinstance(args[1], type) else getattr(args[1], "__default__", None) # type: ignore[assignment]
|
||||
|
||||
if any([self.t_in is None, self.t_out is None]):
|
||||
raise TypeError("Orchestration must have concrete types for all type parameters.")
|
||||
|
||||
async def invoke(
|
||||
self,
|
||||
task: str | DefaultTypeAlias | TIn,
|
||||
runtime: CoreRuntime,
|
||||
) -> OrchestrationResult[TOut]:
|
||||
"""Invoke the multi-agent orchestration.
|
||||
|
||||
This method is non-blocking and will return immediately.
|
||||
To wait for the result, use the `get` method of the `OrchestrationResult` object.
|
||||
|
||||
Args:
|
||||
task (str, DefaultTypeAlias, TIn): The task to be executed by the agents.
|
||||
runtime (CoreRuntime): The runtime environment for the agents.
|
||||
"""
|
||||
self._set_types()
|
||||
|
||||
orchestration_result = OrchestrationResult[self.t_out]() # type: ignore[name-defined]
|
||||
|
||||
async def result_callback(result: DefaultTypeAlias) -> None:
|
||||
nonlocal orchestration_result
|
||||
if inspect.iscoroutinefunction(self._output_transform):
|
||||
transformed_result = await self._output_transform(result)
|
||||
else:
|
||||
transformed_result = self._output_transform(result)
|
||||
|
||||
orchestration_result.value = transformed_result
|
||||
orchestration_result.event.set()
|
||||
|
||||
# This unique topic type is used to isolate the orchestration run from others.
|
||||
internal_topic_type = uuid.uuid4().hex
|
||||
|
||||
await self._prepare(
|
||||
runtime,
|
||||
internal_topic_type=internal_topic_type,
|
||||
result_callback=result_callback,
|
||||
)
|
||||
|
||||
if isinstance(task, str):
|
||||
prepared_task = ChatMessageContent(role=AuthorRole.USER, content=task)
|
||||
elif isinstance(task, ChatMessageContent) or (
|
||||
isinstance(task, list) and all(isinstance(item, ChatMessageContent) for item in task)
|
||||
):
|
||||
prepared_task = task # type: ignore[assignment]
|
||||
else:
|
||||
if inspect.iscoroutinefunction(self._input_transform):
|
||||
prepared_task = await self._input_transform(task) # type: ignore[arg-type]
|
||||
else:
|
||||
prepared_task = self._input_transform(task) # type: ignore[arg-type,assignment]
|
||||
|
||||
background_task = asyncio.create_task(
|
||||
self._start(
|
||||
prepared_task,
|
||||
runtime,
|
||||
internal_topic_type,
|
||||
orchestration_result.cancellation_token,
|
||||
)
|
||||
)
|
||||
|
||||
# Add a callback to surface any exceptions that occur during the task execution.
|
||||
def exception_callback(task: asyncio.Task) -> None:
|
||||
nonlocal orchestration_result
|
||||
try:
|
||||
task.result()
|
||||
except BaseException as e:
|
||||
orchestration_result.exception = e
|
||||
orchestration_result.event.set()
|
||||
|
||||
background_task.add_done_callback(exception_callback)
|
||||
|
||||
return orchestration_result
|
||||
|
||||
@abstractmethod
|
||||
async def _start(
|
||||
self,
|
||||
task: DefaultTypeAlias,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
cancellation_token: CancellationToken,
|
||||
) -> None:
|
||||
"""Start the multi-agent orchestration.
|
||||
|
||||
Args:
|
||||
task (ChatMessageContent | list[ChatMessageContent]): The task to be executed by the agents.
|
||||
runtime (CoreRuntime): The runtime environment for the agents.
|
||||
internal_topic_type (str): The internal topic type for the orchestration that this actor is part of.
|
||||
cancellation_token (CancellationToken): The cancellation token for the orchestration.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
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.
|
||||
|
||||
Args:
|
||||
runtime (CoreRuntime): The runtime environment for the agents.
|
||||
internal_topic_type (str): The internal topic type for the orchestration that this actor is part of.
|
||||
external_topic_type (str | None): The external topic type for the orchestration.
|
||||
direct_actor_type (str | None): The direct actor type for which this actor will relay the output message to.
|
||||
result_callback (Callable): A function that is called when the result is available.
|
||||
"""
|
||||
pass
|
||||
|
||||
def _default_input_transform(self, input_message: TIn) -> DefaultTypeAlias:
|
||||
"""Default input transform function.
|
||||
|
||||
This function transforms the external input message to chat message content(s).
|
||||
If the input message is already in the correct format, it is returned as is.
|
||||
|
||||
Args:
|
||||
input_message (TIn): The input message to be transformed.
|
||||
|
||||
Returns:
|
||||
DefaultTypeAlias: The transformed input message.
|
||||
"""
|
||||
if isinstance(input_message, ChatMessageContent):
|
||||
return input_message
|
||||
|
||||
if isinstance(input_message, list) and all(isinstance(item, ChatMessageContent) for item in input_message):
|
||||
return input_message
|
||||
|
||||
if isinstance(input_message, self.t_in): # type: ignore[arg-type]
|
||||
return ChatMessageContent(
|
||||
role=AuthorRole.USER,
|
||||
content=json.dumps(input_message.__dict__),
|
||||
)
|
||||
|
||||
raise TypeError(f"Invalid input message type: {type(input_message)}. Expected {self.t_in}.")
|
||||
|
||||
def _default_output_transform(self, output_message: DefaultTypeAlias) -> TOut:
|
||||
"""Default output transform function.
|
||||
|
||||
This function transforms the internal output message to the external output message.
|
||||
If the output message is already in the correct format, it is returned as is.
|
||||
|
||||
Args:
|
||||
output_message (DefaultTypeAlias): The output message to be transformed.
|
||||
|
||||
Returns:
|
||||
TOut: The transformed output message.
|
||||
"""
|
||||
if self.t_out == DefaultTypeAlias or self.t_out in get_args(DefaultTypeAlias):
|
||||
if isinstance(output_message, ChatMessageContent) or (
|
||||
isinstance(output_message, list)
|
||||
and all(isinstance(item, ChatMessageContent) for item in output_message)
|
||||
):
|
||||
return output_message # type: ignore[return-value]
|
||||
raise TypeError(f"Invalid output message type: {type(output_message)}. Expected {self.t_out}.")
|
||||
|
||||
if isinstance(output_message, ChatMessageContent):
|
||||
return self.t_out(**json.loads(output_message.content)) # type: ignore[misc]
|
||||
|
||||
raise TypeError(f"Unable to transform output message of type {type(output_message)} to {self.t_out}.")
|
||||
@@ -0,0 +1,187 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
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.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.contents.chat_message_content import ChatMessageContent
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
|
||||
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__)
|
||||
|
||||
|
||||
class SequentialRequestMessage(KernelBaseModel):
|
||||
"""A request message type for sequential agents."""
|
||||
|
||||
body: DefaultTypeAlias
|
||||
|
||||
|
||||
class SequentialResultMessage(KernelBaseModel):
|
||||
"""A result message type for sequential agents."""
|
||||
|
||||
body: ChatMessageContent
|
||||
|
||||
|
||||
class SequentialAgentActor(AgentActorBase):
|
||||
"""A agent actor for sequential agents that process tasks."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
agent: Agent,
|
||||
internal_topic_type: str,
|
||||
next_agent_type: str,
|
||||
agent_response_callback: Callable[[DefaultTypeAlias], Awaitable[None] | None] | None = None,
|
||||
) -> None:
|
||||
"""Initialize the agent actor."""
|
||||
self._next_agent_type = next_agent_type
|
||||
super().__init__(
|
||||
agent=agent,
|
||||
internal_topic_type=internal_topic_type,
|
||||
agent_response_callback=agent_response_callback,
|
||||
)
|
||||
|
||||
@message_handler
|
||||
async def _handle_message(self, message: SequentialRequestMessage, ctx: MessageContext) -> None:
|
||||
"""Handle a message."""
|
||||
logger.debug(f"Sequential actor (Actor ID: {self.id}; Agent name: {self._agent.name}) started processing...")
|
||||
|
||||
response = await self._agent.get_response(messages=message.body) # type: ignore[arg-type]
|
||||
|
||||
logger.debug(f"Sequential actor (Actor ID: {self.id}; Agent name: {self._agent.name}) finished processing.")
|
||||
|
||||
await self._call_agent_response_callback(response.message)
|
||||
|
||||
target_actor_id = await self.runtime.get(self._next_agent_type)
|
||||
await self.send_message(
|
||||
SequentialRequestMessage(body=response.message),
|
||||
target_actor_id,
|
||||
cancellation_token=ctx.cancellation_token,
|
||||
)
|
||||
|
||||
|
||||
class CollectionActor(ActorBase):
|
||||
"""A agent container for collection results from the last agent in the sequence."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
description: str,
|
||||
result_callback: Callable[[DefaultTypeAlias], Awaitable[None]],
|
||||
) -> None:
|
||||
"""Initialize the collection actor."""
|
||||
self._result_callback = result_callback
|
||||
|
||||
super().__init__(description=description)
|
||||
|
||||
@message_handler
|
||||
async def _handle_message(self, message: SequentialRequestMessage, _: MessageContext) -> None:
|
||||
"""Handle the last message."""
|
||||
await self._result_callback(message.body)
|
||||
|
||||
|
||||
class SequentialOrchestration(OrchestrationBase[TIn, TOut]):
|
||||
"""A sequential multi-agent pattern orchestration."""
|
||||
|
||||
@override
|
||||
async def _start(
|
||||
self,
|
||||
task: DefaultTypeAlias,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
cancellation_token: CancellationToken,
|
||||
) -> None:
|
||||
"""Start the sequential pattern."""
|
||||
target_actor_id = await runtime.get(self._get_agent_actor_type(self._members[0], internal_topic_type))
|
||||
await runtime.send_message(
|
||||
SequentialRequestMessage(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_collection_actor(runtime, internal_topic_type, result_callback)
|
||||
|
||||
async def _register_members(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
) -> None:
|
||||
"""Register the members.
|
||||
|
||||
The members will be registered in the reverse order so that the actor type of the next worker
|
||||
is available when the current worker is registered. This is important for the sequential
|
||||
orchestration, where actors need to know its next actor type to send the message to.
|
||||
|
||||
Args:
|
||||
runtime (CoreRuntime): The agent runtime.
|
||||
internal_topic_type (str): The internal topic type for the orchestration that this actor is part of.
|
||||
|
||||
Returns:
|
||||
str: The first actor type in the sequence.
|
||||
"""
|
||||
next_actor_type = self._get_collection_actor_type(internal_topic_type)
|
||||
for index, agent in enumerate(reversed(self._members)):
|
||||
await SequentialAgentActor.register(
|
||||
runtime,
|
||||
self._get_agent_actor_type(agent, internal_topic_type),
|
||||
lambda agent=agent, next_actor_type=next_actor_type: SequentialAgentActor( # type: ignore[misc]
|
||||
agent,
|
||||
internal_topic_type,
|
||||
next_agent_type=next_actor_type,
|
||||
agent_response_callback=self._agent_response_callback,
|
||||
),
|
||||
)
|
||||
logger.debug(f"Registered agent actor of type {self._get_agent_actor_type(agent, internal_topic_type)}")
|
||||
next_actor_type = self._get_agent_actor_type(agent, internal_topic_type)
|
||||
|
||||
async def _register_collection_actor(
|
||||
self,
|
||||
runtime: CoreRuntime,
|
||||
internal_topic_type: str,
|
||||
result_callback: Callable[[DefaultTypeAlias], Awaitable[None]],
|
||||
) -> None:
|
||||
"""Register the collection actor."""
|
||||
await CollectionActor.register(
|
||||
runtime,
|
||||
self._get_collection_actor_type(internal_topic_type),
|
||||
lambda: CollectionActor(
|
||||
description="An internal agent that is responsible for collection results",
|
||||
result_callback=result_callback,
|
||||
),
|
||||
)
|
||||
|
||||
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_collection_actor_type(self, internal_topic_type: str) -> str:
|
||||
"""Get the collection actor type.
|
||||
|
||||
The type is appended with the internal topic type to ensure uniqueness in the runtime
|
||||
that may be shared by multiple orchestrations.
|
||||
"""
|
||||
return f"{CollectionActor.__name__}_{internal_topic_type}"
|
||||
@@ -0,0 +1,65 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias
|
||||
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.kernel import Kernel
|
||||
|
||||
|
||||
def structured_outputs_transform(
|
||||
target_structure: type[BaseModel],
|
||||
service: ChatCompletionClientBase,
|
||||
prompt_execution_settings: PromptExecutionSettings | None = None,
|
||||
) -> Callable[[DefaultTypeAlias], Awaitable[BaseModel]]:
|
||||
"""Return a function that transforms the output of a chat completion service into a target structure.
|
||||
|
||||
Args:
|
||||
target_structure (type): The target structure to transform the output into.
|
||||
service (ChatCompletionClientBase): The chat completion service to use for the transformation. This service
|
||||
must support structured output.
|
||||
prompt_execution_settings (PromptExecutionSettings, optional): The settings to use for the prompt execution.
|
||||
|
||||
Returns:
|
||||
Callable[[DefaultTypeAlias], Awaitable[BaseModel]]: A function that takes the output of
|
||||
the chat completion service and transforms it into the target structure.
|
||||
"""
|
||||
kernel = Kernel()
|
||||
kernel.add_service(service)
|
||||
|
||||
settings = kernel.get_prompt_execution_settings_from_service_id(service.service_id)
|
||||
if prompt_execution_settings:
|
||||
settings.update_from_prompt_execution_settings(prompt_execution_settings)
|
||||
if not hasattr(settings, "response_format"):
|
||||
raise ValueError("The service must support structured output.")
|
||||
settings.response_format = target_structure
|
||||
|
||||
chat_history = ChatHistory(
|
||||
system_message=(
|
||||
"Try your best to summarize the conversation into structured format:\n"
|
||||
f"{target_structure.model_json_schema()}."
|
||||
),
|
||||
)
|
||||
|
||||
async def output_transform(output: DefaultTypeAlias) -> BaseModel:
|
||||
"""Transform the output of the chat completion service into the target structure."""
|
||||
if isinstance(output, ChatMessageContent):
|
||||
chat_history.add_message(output)
|
||||
elif isinstance(output, list) and all(isinstance(item, ChatMessageContent) for item in output):
|
||||
for item in output:
|
||||
chat_history.add_message(item)
|
||||
else:
|
||||
raise ValueError(f"Output must be {DefaultTypeAlias}.")
|
||||
|
||||
response = await service.get_chat_message_content(chat_history, settings)
|
||||
assert response is not None # nosec B101
|
||||
|
||||
return target_structure.model_validate_json(response.content)
|
||||
|
||||
return output_transform
|
||||
@@ -0,0 +1,82 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from collections.abc import AsyncIterable, Awaitable, Callable
|
||||
|
||||
from semantic_kernel.agents.agent import Agent, AgentResponseItem, AgentThread
|
||||
from semantic_kernel.agents.runtime.core.core_runtime import CoreRuntime
|
||||
from semantic_kernel.contents.chat_message_content import ChatMessageContent
|
||||
from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
|
||||
from semantic_kernel.contents.utils.author_role import AuthorRole
|
||||
|
||||
if sys.version_info >= (3, 12):
|
||||
from typing import override # pragma: no cover
|
||||
else:
|
||||
from typing_extensions import override # pragma: no cover
|
||||
|
||||
|
||||
class MockAgentThread(AgentThread):
|
||||
"""A mock agent thread for testing purposes."""
|
||||
|
||||
@override
|
||||
async def _create(self) -> str:
|
||||
return "mock_thread_id"
|
||||
|
||||
@override
|
||||
async def _delete(self) -> None:
|
||||
pass
|
||||
|
||||
@override
|
||||
async def _on_new_message(self, new_message: ChatMessageContent) -> None:
|
||||
pass
|
||||
|
||||
|
||||
class MockAgent(Agent):
|
||||
"""A mock agent for testing purposes."""
|
||||
|
||||
@override
|
||||
async def get_response(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
**kwargs,
|
||||
) -> AgentResponseItem[ChatMessageContent]:
|
||||
# Simulate some processing time
|
||||
await asyncio.sleep(0.1)
|
||||
return AgentResponseItem[ChatMessageContent](
|
||||
message=ChatMessageContent(
|
||||
role=AuthorRole.ASSISTANT,
|
||||
content="mock_response",
|
||||
),
|
||||
thread=thread or MockAgentThread(),
|
||||
)
|
||||
|
||||
@override
|
||||
async def invoke(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
|
||||
**kwargs,
|
||||
) -> AgentResponseItem[ChatMessageContent]:
|
||||
pass
|
||||
|
||||
@override
|
||||
async def invoke_stream(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
|
||||
**kwargs,
|
||||
) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
|
||||
pass
|
||||
|
||||
|
||||
class MockRuntime(CoreRuntime):
|
||||
"""A mock agent runtime for testing purposes."""
|
||||
|
||||
pass
|
||||
@@ -0,0 +1,153 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.agents.orchestration.concurrent import ConcurrentOrchestration
|
||||
from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationResult
|
||||
from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime
|
||||
from semantic_kernel.contents.chat_message_content import ChatMessageContent
|
||||
from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
|
||||
|
||||
|
||||
async def test_prepare():
|
||||
"""Test the prepare method of the ConcurrentOrchestration."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = MockRuntime()
|
||||
|
||||
package_path = "semantic_kernel.agents.orchestration.concurrent"
|
||||
with (
|
||||
patch(f"{package_path}.ConcurrentOrchestration._start"),
|
||||
patch(f"{package_path}.ConcurrentAgentActor.register") as mock_agent_actor_register,
|
||||
patch(f"{package_path}.CollectionActor.register") as mock_collection_actor_register,
|
||||
patch.object(runtime, "add_subscription") as mock_add_subscription,
|
||||
):
|
||||
orchestration = ConcurrentOrchestration(members=[agent_a, agent_b])
|
||||
await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
assert mock_agent_actor_register.call_count == 2
|
||||
assert mock_collection_actor_register.call_count == 1
|
||||
assert mock_add_subscription.call_count == 2
|
||||
|
||||
|
||||
async def test_invoke():
|
||||
"""Test the invoke method of the ConcurrentOrchestration."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = ConcurrentOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
result = await orchestration_result.get(1.0)
|
||||
|
||||
assert isinstance(orchestration_result, OrchestrationResult)
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
assert all(isinstance(item, ChatMessageContent) for item in result)
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
async def test_invoke_with_response_callback():
|
||||
"""Test the invoke method of the ConcurrentOrchestration with a response callback."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
responses: list[DefaultTypeAlias] = []
|
||||
try:
|
||||
orchestration = ConcurrentOrchestration(
|
||||
members=[agent_a, agent_b],
|
||||
agent_response_callback=lambda x: responses.append(x),
|
||||
)
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
await orchestration_result.get(1.0)
|
||||
|
||||
assert len(responses) == 2
|
||||
assert all(isinstance(item, ChatMessageContent) for item in responses)
|
||||
assert all(item.content == "mock_response" for item in responses)
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
@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_cancel_before_completion():
|
||||
"""Test the invoke method of the ConcurrentOrchestration with cancellation before completion."""
|
||||
with (
|
||||
patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
|
||||
):
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = ConcurrentOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Cancel before the collection agent gets the responses
|
||||
await asyncio.sleep(0.05)
|
||||
orchestration_result.cancel()
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
assert mock_get_response.call_count == 2
|
||||
|
||||
|
||||
async def test_invoke_cancel_after_completion():
|
||||
"""Test the invoke method of the ConcurrentOrchestration with cancellation after completion."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = ConcurrentOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Wait for the orchestration to complete
|
||||
await orchestration_result.get(1.0)
|
||||
|
||||
with pytest.raises(RuntimeError, match="The invocation has already been completed."):
|
||||
orchestration_result.cancel()
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
async def test_invoke_with_double_get_result():
|
||||
"""Test the invoke method of the ConcurrentOrchestration with double get result."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = ConcurrentOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Get result before completion
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await orchestration_result.get(0.1)
|
||||
# The invocation should still be in progress and getting the result again should not raise an error
|
||||
result = await orchestration_result.get()
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 2
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
@@ -0,0 +1,422 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from collections.abc import AsyncIterable, Awaitable, Callable
|
||||
from dataclasses import dataclass
|
||||
from unittest.mock import ANY, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.agents.agent import Agent, AgentResponseItem, AgentThread
|
||||
from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationBase, TIn, TOut
|
||||
from semantic_kernel.contents.chat_message_content import ChatMessageContent
|
||||
from semantic_kernel.contents.streaming_chat_message_content import StreamingChatMessageContent
|
||||
from semantic_kernel.contents.utils.author_role import AuthorRole
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
from tests.unit.agents.orchestration.conftest import MockRuntime
|
||||
|
||||
if sys.version_info >= (3, 12):
|
||||
from typing import override # pragma: no cover
|
||||
else:
|
||||
from typing_extensions import override # pragma: no cover
|
||||
|
||||
|
||||
class MockAgent(Agent):
|
||||
"""A mock agent for testing purposes."""
|
||||
|
||||
@override
|
||||
async def get_response(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
**kwargs,
|
||||
) -> AgentResponseItem[ChatMessageContent]:
|
||||
pass
|
||||
|
||||
@override
|
||||
async def invoke(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
|
||||
**kwargs,
|
||||
) -> AgentResponseItem[ChatMessageContent]:
|
||||
pass
|
||||
|
||||
@override
|
||||
async def invoke_stream(
|
||||
self,
|
||||
*,
|
||||
messages: str | ChatMessageContent | list[str | ChatMessageContent] | None = None,
|
||||
thread: AgentThread | None = None,
|
||||
on_intermediate_message: Callable[[ChatMessageContent], Awaitable[None]] | None = None,
|
||||
**kwargs,
|
||||
) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
|
||||
pass
|
||||
|
||||
|
||||
class MockOrchestration(OrchestrationBase[TIn, TOut]):
|
||||
"""A mock orchestration base for testing purposes."""
|
||||
|
||||
async def _start(self, task, runtime, internal_topic_type, collection_agent_type):
|
||||
pass
|
||||
|
||||
async def _prepare(self, runtime, internal_topic_type, result_callback):
|
||||
pass
|
||||
|
||||
|
||||
def test_orchestration_init():
|
||||
"""Test the initialization of the MockOrchestration."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
agent_c = MockAgent()
|
||||
|
||||
orchestration = MockOrchestration(
|
||||
members=[agent_a, agent_b, agent_c],
|
||||
name="test_orchestration",
|
||||
description="Test Orchestration",
|
||||
)
|
||||
|
||||
assert orchestration.name == "test_orchestration"
|
||||
assert orchestration.description == "Test Orchestration"
|
||||
|
||||
assert len(orchestration._members) == 3
|
||||
assert orchestration._input_transform is not None
|
||||
assert orchestration._output_transform is not None
|
||||
assert orchestration._agent_response_callback is None
|
||||
|
||||
|
||||
def test_orchestration_init_with_default_values():
|
||||
"""Test the initialization of the MockOrchestration with default values."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
orchestration = MockOrchestration(members=[agent_a, agent_b])
|
||||
|
||||
assert orchestration.name
|
||||
assert orchestration.description
|
||||
|
||||
assert len(orchestration._members) == 2
|
||||
assert orchestration._input_transform is not None
|
||||
assert orchestration._output_transform is not None
|
||||
assert orchestration._agent_response_callback is None
|
||||
|
||||
|
||||
def test_orchestration_init_with_no_members():
|
||||
"""Test the initialization of the OrchestrationBase with no members."""
|
||||
with pytest.raises(ValueError):
|
||||
_ = MockOrchestration(members=[])
|
||||
|
||||
|
||||
def test_orchestration_set_types():
|
||||
"""Test the set_types method of OrchestrationBase."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
# Test with default types
|
||||
orchestration_a = MockOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_a._set_types()
|
||||
|
||||
assert orchestration_a.t_in is DefaultTypeAlias
|
||||
assert orchestration_a.t_out is DefaultTypeAlias
|
||||
|
||||
# Test with a custom input type and default output type
|
||||
orchestration_c = MockOrchestration[int](members=[agent_a, agent_b])
|
||||
orchestration_c._set_types()
|
||||
|
||||
assert orchestration_c.t_in is int
|
||||
assert orchestration_c.t_out is DefaultTypeAlias
|
||||
|
||||
# Test with a custom input type and custom output type
|
||||
orchestration_b = MockOrchestration[str, int](members=[agent_a, agent_b])
|
||||
orchestration_b._set_types()
|
||||
|
||||
assert orchestration_b.t_in is str
|
||||
assert orchestration_b.t_out is int
|
||||
|
||||
# Test with an incorrect number of types
|
||||
with pytest.raises(TypeError):
|
||||
orchestration_d = MockOrchestration[str, str, str](members=[agent_a, agent_b])
|
||||
orchestration_d._set_types()
|
||||
|
||||
|
||||
async def test_orchestration_invoke_with_str():
|
||||
"""Test the invoke method of OrchestrationBase with a string input."""
|
||||
orchestration = MockOrchestration(members=[MockAgent(), MockAgent()])
|
||||
|
||||
with patch.object(orchestration, "_start") as mock_start:
|
||||
await orchestration.invoke("Test message", MockRuntime())
|
||||
mock_start.assert_called_once_with(
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message"), ANY, ANY, ANY
|
||||
)
|
||||
|
||||
|
||||
async def test_orchestration_invoke_with_chat_message_content():
|
||||
"""Test the invoke method of OrchestrationBase with a ChatMessageContent input."""
|
||||
orchestration = MockOrchestration(members=[MockAgent(), MockAgent()])
|
||||
|
||||
chat_message_content = ChatMessageContent(role=AuthorRole.USER, content="Test message")
|
||||
with patch.object(orchestration, "_start") as mock_start:
|
||||
await orchestration.invoke(chat_message_content, MockRuntime())
|
||||
mock_start.assert_called_once_with(chat_message_content, ANY, ANY, ANY)
|
||||
|
||||
chat_message_content_list = [
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 1"),
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 2"),
|
||||
]
|
||||
with patch.object(orchestration, "_start") as mock_start:
|
||||
await orchestration.invoke(chat_message_content_list, MockRuntime())
|
||||
mock_start.assert_called_once_with(chat_message_content_list, ANY, ANY, ANY)
|
||||
|
||||
|
||||
async def test_orchestration_invoke_with_custom_type():
|
||||
"""Test the invoke method of OrchestrationBase with a custom type."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration = MockOrchestration[CustomType, TOut](members=[MockAgent(), MockAgent()])
|
||||
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
with patch.object(orchestration, "_start") as mock_start:
|
||||
await orchestration.invoke(custom_type_instance, MockRuntime())
|
||||
mock_start.assert_called_once_with(
|
||||
ChatMessageContent(role=AuthorRole.USER, content=json.dumps(custom_type_instance.__dict__)), ANY, ANY, ANY
|
||||
)
|
||||
|
||||
|
||||
async def test_orchestration_invoke_with_custom_type_async_input_transform():
|
||||
"""Test the invoke method of OrchestrationBase with a custom type and async input transform."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
async def async_input_transform(input_data: CustomType) -> ChatMessageContent:
|
||||
await asyncio.sleep(0.1) # Simulate async processing
|
||||
return ChatMessageContent(role=AuthorRole.USER, content="Test message")
|
||||
|
||||
orchestration = MockOrchestration[CustomType, TOut](
|
||||
members=[MockAgent(), MockAgent()],
|
||||
input_transform=async_input_transform,
|
||||
)
|
||||
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
with patch.object(orchestration, "_start") as mock_start:
|
||||
await orchestration.invoke(custom_type_instance, MockRuntime())
|
||||
mock_start.assert_called_once_with(
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message"), ANY, ANY, ANY
|
||||
)
|
||||
|
||||
|
||||
def test_default_input_transform_default_type_alias():
|
||||
"""Test the default_input_transform method of OrchestrationBase."""
|
||||
orchestration = MockOrchestration(members=[MockAgent(), MockAgent()])
|
||||
orchestration._set_types()
|
||||
|
||||
chat_message_content = ChatMessageContent(role=AuthorRole.USER, content="Test message")
|
||||
transformed_input = orchestration._default_input_transform(chat_message_content)
|
||||
|
||||
assert isinstance(transformed_input, ChatMessageContent)
|
||||
|
||||
chat_message_content_list = [
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 1"),
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 2"),
|
||||
]
|
||||
transformed_input_list = orchestration._default_input_transform(chat_message_content_list)
|
||||
|
||||
assert isinstance(transformed_input_list, list) and all(
|
||||
isinstance(item, ChatMessageContent) for item in transformed_input_list
|
||||
)
|
||||
|
||||
|
||||
def test_default_input_transform_custom_type():
|
||||
"""Test the default_input_transform method of OrchestrationBase with a custom type."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration_a = MockOrchestration[CustomType, TOut](members=[MockAgent(), MockAgent()])
|
||||
orchestration_a._set_types()
|
||||
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
transformed_input_a = orchestration_a._default_input_transform(custom_type_instance)
|
||||
|
||||
assert isinstance(transformed_input_a, ChatMessageContent)
|
||||
assert CustomType(**json.loads(transformed_input_a.content)) == custom_type_instance
|
||||
assert transformed_input_a.role == AuthorRole.USER
|
||||
|
||||
class CustomModel(KernelBaseModel):
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration_b = MockOrchestration[CustomModel, TOut](members=[MockAgent(), MockAgent()])
|
||||
orchestration_b._set_types()
|
||||
|
||||
custom_model_instance = CustomModel(value="Test message", number=42)
|
||||
transformed_input_b = orchestration_b._default_input_transform(custom_model_instance)
|
||||
|
||||
assert isinstance(transformed_input_b, ChatMessageContent)
|
||||
assert CustomModel.model_validate_json(transformed_input_b.content) == custom_model_instance
|
||||
assert transformed_input_b.role == AuthorRole.USER
|
||||
|
||||
|
||||
def test_default_input_transform_custom_type_error():
|
||||
"""Test the default_input_transform method of OrchestrationBase with an incorrect type."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
class CustomModel(KernelBaseModel):
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration = MockOrchestration[CustomModel, TOut](members=[MockAgent(), MockAgent()])
|
||||
orchestration._set_types()
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
orchestration._default_input_transform(custom_type_instance)
|
||||
|
||||
|
||||
def test_default_output_transform_default_type_alias():
|
||||
"""Test the default_output_transform method of OrchestrationBase."""
|
||||
orchestration = MockOrchestration(members=[MockAgent(), MockAgent()])
|
||||
orchestration._set_types()
|
||||
|
||||
chat_message_content = ChatMessageContent(role=AuthorRole.USER, content="Test message")
|
||||
transformed_output = orchestration._default_output_transform(chat_message_content)
|
||||
|
||||
assert isinstance(transformed_output, ChatMessageContent)
|
||||
|
||||
chat_message_content_list = [
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 1"),
|
||||
ChatMessageContent(role=AuthorRole.USER, content="Test message 2"),
|
||||
]
|
||||
transformed_output_list = orchestration._default_output_transform(chat_message_content_list)
|
||||
|
||||
assert isinstance(transformed_output_list, list) and all(
|
||||
isinstance(item, ChatMessageContent) for item in transformed_output_list
|
||||
)
|
||||
|
||||
with pytest.raises(TypeError, match="Invalid output message type"):
|
||||
orchestration._default_output_transform("Invalid type")
|
||||
|
||||
|
||||
def test_default_output_transform_custom_type():
|
||||
"""Test the default_output_transform method of OrchestrationBase with a custom type."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration_a = MockOrchestration[TIn, CustomType](members=[MockAgent(), MockAgent()])
|
||||
orchestration_a._set_types()
|
||||
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
chat_message_content = ChatMessageContent(role=AuthorRole.USER, content=json.dumps(custom_type_instance.__dict__))
|
||||
|
||||
transformed_output_a = orchestration_a._default_output_transform(chat_message_content)
|
||||
|
||||
assert isinstance(transformed_output_a, CustomType)
|
||||
assert transformed_output_a == custom_type_instance
|
||||
|
||||
class CustomModel(KernelBaseModel):
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration_b = MockOrchestration[TIn, CustomModel](members=[MockAgent(), MockAgent()])
|
||||
orchestration_b._set_types()
|
||||
|
||||
custom_model_instance = CustomModel(value="Test message", number=42)
|
||||
chat_message_content = ChatMessageContent(role=AuthorRole.USER, content=custom_model_instance.model_dump_json())
|
||||
|
||||
transformed_output_b = orchestration_b._default_output_transform(chat_message_content)
|
||||
|
||||
assert isinstance(transformed_output_b, CustomModel)
|
||||
assert transformed_output_b == custom_model_instance
|
||||
|
||||
|
||||
def test_default_output_transform_custom_type_error():
|
||||
"""Test the default_output_transform method of OrchestrationBase with an incorrect type."""
|
||||
|
||||
@dataclass
|
||||
class CustomType:
|
||||
value: str
|
||||
number: int
|
||||
|
||||
class CustomModel(KernelBaseModel):
|
||||
value: str
|
||||
number: int
|
||||
|
||||
orchestration = MockOrchestration[TIn, CustomModel](members=[MockAgent(), MockAgent()])
|
||||
orchestration._set_types()
|
||||
|
||||
with pytest.raises(TypeError, match="Unable to transform output message"):
|
||||
custom_type_instance = CustomType(value="Test message", number=42)
|
||||
orchestration._default_output_transform(custom_type_instance)
|
||||
|
||||
|
||||
async def test_invoke_with_timeout_error():
|
||||
"""Test the invoke method of the MockOrchestration with a timeout error."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
orchestration = MockOrchestration(members=[agent_a, agent_b])
|
||||
# The orchestration_result will never be set by the MockOrchestration
|
||||
orchestration_result = await orchestration.invoke(
|
||||
task="test_message",
|
||||
runtime=MockRuntime(),
|
||||
)
|
||||
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await orchestration_result.get(timeout=0.1)
|
||||
|
||||
|
||||
async def test_invoke_cancel_before_completion():
|
||||
"""Test the invoke method of the MockOrchestration with cancellation before completion."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
orchestration = MockOrchestration(members=[agent_a, agent_b])
|
||||
# The orchestration_result will never be set by the MockOrchestration
|
||||
orchestration_result = await orchestration.invoke(
|
||||
task="test_message",
|
||||
runtime=MockRuntime(),
|
||||
)
|
||||
|
||||
# Cancel the orchestration before completion
|
||||
orchestration_result.cancel()
|
||||
|
||||
with pytest.raises(RuntimeError, match="The invocation was canceled before it could complete."):
|
||||
await orchestration_result.get()
|
||||
|
||||
|
||||
async def test_invoke_with_double_cancel():
|
||||
"""Test the invoke method of the MockOrchestration with double cancel."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
orchestration = MockOrchestration(members=[agent_a, agent_b])
|
||||
# The orchestration_result will never be set by the MockOrchestration
|
||||
orchestration_result = await orchestration.invoke(
|
||||
task="test_message",
|
||||
runtime=MockRuntime(),
|
||||
)
|
||||
|
||||
orchestration_result.cancel()
|
||||
# Cancelling again should raise an error
|
||||
with pytest.raises(RuntimeError, match="The invocation has already been canceled."):
|
||||
orchestration_result.cancel()
|
||||
@@ -0,0 +1,173 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from typing import Any, Literal
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel
|
||||
|
||||
from semantic_kernel.agents.orchestration.tools import structured_outputs_transform
|
||||
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_message_content import ChatMessageContent
|
||||
from semantic_kernel.kernel_pydantic import KernelBaseModel
|
||||
|
||||
if sys.version_info >= (3, 12):
|
||||
from typing import override # pragma: no cover
|
||||
else:
|
||||
from typing_extensions import override # pragma: no cover
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
class MockChatCompletionService(ChatCompletionClientBase):
|
||||
"""A mock chat completion service for testing purposes."""
|
||||
|
||||
@override
|
||||
def get_prompt_execution_settings_class(self) -> type["PromptExecutionSettings"]:
|
||||
"""Get the request settings class."""
|
||||
return MockPromptExecutionSettings
|
||||
|
||||
|
||||
class MockModel(KernelBaseModel):
|
||||
name: str
|
||||
age: int
|
||||
|
||||
|
||||
def test_structured_outputs_transform():
|
||||
"""Test the structured_outputs_transform function."""
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings()
|
||||
|
||||
transform = structured_outputs_transform(
|
||||
target_structure=MockModel,
|
||||
service=service,
|
||||
prompt_execution_settings=prompt_execution_settings,
|
||||
)
|
||||
|
||||
assert isinstance(transform, Callable)
|
||||
|
||||
|
||||
def test_structured_outputs_transform_original_settings_not_changed():
|
||||
"""Test the structured_outputs_transform function with original settings not changed."""
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings(
|
||||
extension_data={"test_key": "test_value"},
|
||||
)
|
||||
|
||||
_ = structured_outputs_transform(
|
||||
target_structure=MockModel,
|
||||
service=service,
|
||||
prompt_execution_settings=prompt_execution_settings,
|
||||
)
|
||||
|
||||
assert not hasattr(prompt_execution_settings, "response_format")
|
||||
assert prompt_execution_settings.extension_data["test_key"] == "test_value"
|
||||
|
||||
|
||||
def test_structured_outputs_transform_unsupported_service():
|
||||
"""Test the structured_outputs_transform function with unsupported service."""
|
||||
|
||||
with (
|
||||
patch.object(
|
||||
MockChatCompletionService, "get_prompt_execution_settings_class"
|
||||
) as mock_get_prompt_execution_settings_class,
|
||||
pytest.raises(ValueError),
|
||||
):
|
||||
mock_get_prompt_execution_settings_class.return_value = PromptExecutionSettings
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings()
|
||||
|
||||
_ = structured_outputs_transform(MockModel, service, prompt_execution_settings)
|
||||
|
||||
|
||||
async def test_structured_outputs_transform_invoke():
|
||||
"""Test the structured_outputs_transform function and invoke the transform."""
|
||||
mock_model = MockModel(name="John Doe", age=30)
|
||||
|
||||
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_model.model_dump_json()
|
||||
)
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings()
|
||||
|
||||
transform = structured_outputs_transform(
|
||||
target_structure=MockModel,
|
||||
service=service,
|
||||
prompt_execution_settings=prompt_execution_settings,
|
||||
)
|
||||
|
||||
result = await transform(ChatMessageContent(role="user", content="name is John Doe and age is 30"))
|
||||
|
||||
assert isinstance(result, MockModel)
|
||||
assert result == mock_model
|
||||
|
||||
mock_get_chat_message_content.assert_called_once()
|
||||
assert len(mock_get_chat_message_content.call_args[0][0].messages) == 2
|
||||
|
||||
|
||||
async def test_structured_outputs_transform_invoke_with_messages():
|
||||
"""Test the structured_outputs_transform function and invoke the transform with messages."""
|
||||
mock_model = MockModel(name="John Doe", age=30)
|
||||
|
||||
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_model.model_dump_json()
|
||||
)
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings()
|
||||
|
||||
transform = structured_outputs_transform(
|
||||
target_structure=MockModel,
|
||||
service=service,
|
||||
prompt_execution_settings=prompt_execution_settings,
|
||||
)
|
||||
|
||||
result = await transform([
|
||||
ChatMessageContent(role="user", content="name is John Doe"),
|
||||
ChatMessageContent(role="user", content="age is 30"),
|
||||
])
|
||||
|
||||
assert isinstance(result, MockModel)
|
||||
assert result == mock_model
|
||||
|
||||
mock_get_chat_message_content.assert_called_once()
|
||||
assert len(mock_get_chat_message_content.call_args[0][0].messages) == 3
|
||||
|
||||
|
||||
async def test_structured_outputs_transform_invoke_unsupported_type():
|
||||
"""Test the structured_outputs_transform function and invoke the transform with messages of unsupported type."""
|
||||
|
||||
service = MockChatCompletionService(ai_model_id="test_model")
|
||||
prompt_execution_settings = PromptExecutionSettings()
|
||||
|
||||
transform = structured_outputs_transform(
|
||||
target_structure=MockModel,
|
||||
service=service,
|
||||
prompt_execution_settings=prompt_execution_settings,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
_ = await transform("unsupported type")
|
||||
@@ -0,0 +1,151 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from semantic_kernel.agents.orchestration.orchestration_base import DefaultTypeAlias, OrchestrationResult
|
||||
from semantic_kernel.agents.orchestration.sequential import SequentialOrchestration
|
||||
from semantic_kernel.agents.runtime.in_process.in_process_runtime import InProcessRuntime
|
||||
from semantic_kernel.contents.chat_message_content import ChatMessageContent
|
||||
from tests.unit.agents.orchestration.conftest import MockAgent, MockRuntime
|
||||
|
||||
|
||||
async def test_prepare():
|
||||
"""Test the prepare method of the SequentialOrchestration."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = MockRuntime()
|
||||
|
||||
package_path = "semantic_kernel.agents.orchestration.sequential"
|
||||
with (
|
||||
patch(f"{package_path}.SequentialOrchestration._start"),
|
||||
patch(f"{package_path}.SequentialAgentActor.register") as mock_agent_actor_register,
|
||||
patch(f"{package_path}.CollectionActor.register") as mock_collection_actor_register,
|
||||
patch.object(runtime, "add_subscription") as mock_add_subscription,
|
||||
):
|
||||
orchestration = SequentialOrchestration(members=[agent_a, agent_b])
|
||||
await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
assert mock_agent_actor_register.call_count == 2
|
||||
assert mock_collection_actor_register.call_count == 1
|
||||
assert mock_add_subscription.call_count == 0
|
||||
|
||||
|
||||
async def test_invoke():
|
||||
"""Test the invoke method of the SequentialOrchestration."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = SequentialOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
result = await orchestration_result.get(1.0)
|
||||
|
||||
assert isinstance(orchestration_result, OrchestrationResult)
|
||||
assert isinstance(result, ChatMessageContent)
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
async def test_invoke_with_response_callback():
|
||||
"""Test the invoke method of the SequentialOrchestration with a response callback."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
responses: list[DefaultTypeAlias] = []
|
||||
try:
|
||||
orchestration = SequentialOrchestration(
|
||||
members=[agent_a, agent_b],
|
||||
agent_response_callback=lambda x: responses.append(x),
|
||||
)
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
await orchestration_result.get(1.0)
|
||||
|
||||
assert len(responses) == 2
|
||||
assert all(isinstance(item, ChatMessageContent) for item in responses)
|
||||
assert all(item.content == "mock_response" for item in responses)
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
@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_cancel_before_completion():
|
||||
"""Test the invoke method of the SequentialOrchestration with cancellation before completion."""
|
||||
with (
|
||||
patch.object(MockAgent, "get_response", wraps=MockAgent.get_response, autospec=True) as mock_get_response,
|
||||
):
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = SequentialOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Cancel while the first agent is processing
|
||||
await asyncio.sleep(0.05)
|
||||
orchestration_result.cancel()
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
assert mock_get_response.call_count == 1
|
||||
|
||||
|
||||
async def test_invoke_cancel_after_completion():
|
||||
"""Test the invoke method of the SequentialOrchestration with cancellation after completion."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = SequentialOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Wait for the orchestration to complete
|
||||
await orchestration_result.get(1.0)
|
||||
|
||||
with pytest.raises(RuntimeError, match="The invocation has already been completed."):
|
||||
orchestration_result.cancel()
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
|
||||
|
||||
async def test_invoke_with_double_get_result():
|
||||
"""Test the invoke method of the SequentialOrchestration with double get result."""
|
||||
agent_a = MockAgent()
|
||||
agent_b = MockAgent()
|
||||
|
||||
runtime = InProcessRuntime()
|
||||
runtime.start()
|
||||
|
||||
try:
|
||||
orchestration = SequentialOrchestration(members=[agent_a, agent_b])
|
||||
orchestration_result = await orchestration.invoke(task="test_message", runtime=runtime)
|
||||
|
||||
# Get result before completion
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await orchestration_result.get(0.1)
|
||||
# The invocation should still be in progress and getting the result again should not raise an error
|
||||
result = await orchestration_result.get()
|
||||
|
||||
assert isinstance(result, ChatMessageContent)
|
||||
assert result.content == "mock_response"
|
||||
finally:
|
||||
await runtime.stop_when_idle()
|
||||
Reference in New Issue
Block a user