Orchestration
This guide covers orchestration idea in tests
Idea
In multi AI Agent systems you will face a situation when AI Agents shall work together to complete some task without user intervention. For instance, CrewAI framework has been built for such purpose. In Maia, tests can be written to use different kind of providers, so framework itself shall give possibility to simulate such behaviour. This is done using orchestration policies.
Note:
Maia does not mirror the orchestration frameworks like CrewAI. Maia gives you more - it gives possibility to write autonomous tests so you can simulate such behavior between different providers. This is crucial difference, because Maia can test CrewAI orchestration framework just using CrewAI provider, while simulation between different providers is not possible in CrewAI.
Policies
Maia provides Orchestration Policies which enables custom routing messages between agents.
Ignore message policy
This policy enables ignore_trigger_prompt, which can be attached to Agent. This prompt is added automatically to system_message of an Agent and Agent itself is responsible for evaluating a message broadcasted by user. The prompt need to ask Agent to respond with special message IGNORE_MESSAGE which is evaluated by the framework.
If Agent decides that it is not a target agent, then responses with IGNORE_MESSAGE. Maia evaluates it by ignoring such message and waiting for another Agent.
This policy is useful if you do not want to introduce another agent to orchestrate the routing, but you are moving the responsibility to every agent.
Example of test:
class TestConversationSessions(MaiaTest):
def setup_agents(self):
self.create_agent(
name="Alice",
provider=GenericLiteLLMProvider(config={
"model": "ollama/mistral",
"api_base": "http://localhost:11434"
}),
system_message="You are a weather assistant. Only describe the weather.",
ignore_trigger_prompt="You MUST NOT answer questions about any other topic, including what to wear. If the user asks about anything other than the weather, you MUST respond with only the exact text: IGNORE_MESSAGE",
)
self.create_agent(
name="Bob",
provider=GenericLiteLLMProvider(config={
"model": "ollama/mistral",
"api_base": "http://localhost:11434"
}),
system_message="You are a pirate assistant who only suggests clothing.",
ignore_trigger_prompt="If the question is not about what to wear, you MUST respond with only the exact text: IGNORE_MESSAGE"
)
@pytest.mark.asyncio
async def test_conversation_broadcast_ignore_message_policy(self):
session = self.create_session(["Alice", "Bob"], orchestration_policy=OrchestrationPolicy.IGNORE_MESSAGE)
# Test that only Alice responds to a weather question
response_a, responder_a = await session.user_says_and_broadcast("Please describe the usual weather in London in July, including temperature and conditions.")
print(f"{responder_a}: {response_a.content}")
assert responder_a == 'Alice'
assert_agent_participated(session, 'Alice')
# Test that only Bob responds to a clothing question
response_b, responder_b = await session.user_says_and_broadcast(f"Given the weather: {response_a.content}, what clothes should I wear?")
print(f"{responder_b}: {response_b.content}")
assert responder_b == 'Bob'
assert_agent_participated(session, 'Bob')
# Test that no one responds to an irrelevant question
response_c, responder_c = await session.user_says_and_broadcast("What is the capital of France?")
assert response_c is None
assert responder_c is None
Orchestration agent
Another way to automatically route the messages is using dedicated agent. Such orchestration agent has access to all agents in the session, together with their system messages. It means that it has full view what are the responsibility of every agent. When user broadcasts the message, it is always taken by the orchestration agent and routed to the usual agent.
Orchestration agent can be created with any provider supported by Maia.
Caution:
Please be aware that orchestration agent is autonomous AI agent - it means that there is possibility that it will route message to the wrong agent. However, it is also a part of a testing - in such case it means that system_messages assigned to agents are not enough informative to distinguish responsibility. It needs to be then corrected.
Example of test:
class TestConversationSessions(MaiaTest):
def setup_agents(self):
self.create_agent(
name="Alice",
provider=GenericLiteLLMProvider(config={
"model": "ollama/mistral",
"api_base": "http://localhost:11434"
}),
system_message="You are a weather assistant. Only describe the weather.",
)
self.create_agent(
name="Bob",
provider=GenericLiteLLMProvider(config={
"model": "ollama/mistral",
"api_base": "http://localhost:11434"
}),
system_message="You are a pirate assistant who only suggests clothing.",
)
@pytest.mark.asyncio
async def test_conversation_broadcast_orchestration_policy(self):
orchestration_agent = OrchestrationAgent(self.get_provider("ollama"))
session = self.create_session(["Alice", "Bob"], orchestration_agent=orchestration_agent, orchestration_policy=OrchestrationPolicy.ORCHESTRATION_AGENT)
# Test that only Alice responds to a weather question
response_a, responder_a = await session.user_says_and_broadcast("Please describe the usual weather in London in July, including temperature and conditions.")
print(f"{responder_a}: {response_a.content}")
assert responder_a == 'Alice'
assert_agent_participated(session, 'Alice')
# Test that only Bob responds to a clothing question
response_b, responder_b = await session.user_says_and_broadcast(f"Given the weather: {response_a.content}, what clothes should I wear?")
print(f"{responder_b}: {response_b.content}")
assert responder_b == 'Bob'
assert_agent_participated(session, 'Bob')
# Test that no one responds to an irrelevant question
response_c, responder_c = await session.user_says_and_broadcast("What is the capital of France?")
assert response_c is None
assert responder_c is None