Validator
This chapter describes the concept of a validator.
Idea
A Validator in the Maia Framework is used to check the whole session for specific scenarios.
Here is an example of validator usage:
@pytest.mark.asyncio
async def test_assert_latency_below(self):
session = self.create_session(
["Alice"],
validators=[performance_validator(threshold=60, unit="seconds")]
)
await session.user_says("What is the weather like today?")
await session.agent_responds("Alice")
And here is the validator implementation:
from typing import Callable
from maia_test_framework.core.session import Session
from datetime import timedelta
def performance_validator(threshold: int, unit: str = "seconds") -> Callable[[Session], None]:
"""
Returns a validator that asserts that the latency between user and agent messages is below a threshold.
"""
def latency_validator(session: Session):
"""Asserts that the latency between user and agent messages is below a threshold."""
if unit == "milliseconds":
delta = timedelta(milliseconds=threshold)
elif unit == "seconds":
delta = timedelta(seconds=threshold)
elif unit == "minutes":
delta = timedelta(minutes=threshold)
else:
raise ValueError("unit must be one of 'milliseconds', 'seconds', or 'minutes'")
messages = session.message_history
for i in range(1, len(messages)):
if messages[i-1].sender == "user":
latency = messages[i].timestamp - messages[i-1].timestamp
if latency > delta:
raise AssertionError(f"Latency of {latency} between user and agent {messages[i].sender} exceeded the threshold of {delta}.")
return latency_validator
In the example above, performance_validator is responsible for checking if every agent responded in defined time threshold.
Note:
You can create your own validator by following above example and use it in a test.
Validator vs Assertion
Validator works similarly to an Assertion, but it is executed at teardown, so after the whole session, it has access to the entire history. This means you can have more holistic checks. On the other hand, you can often achieve a similar thing with an Assertion, but it would require more checks in tests. You are free to choose your own way of testing.
Negative testing
Sometimes you want to test that some validator raises an error. In such case, you can manually run the validator instead of assinging it to the session.
Here is the example of test which checks if agent has not partcipated in the session:
@pytest.mark.asyncio
async def test_assert_agent_not_participating(self):
session = self.create_session(["Alice", "Bob"])
await session.agent_says("Alice", "Bob", "What shall I wear if it is sunny?")
await session.agent_responds("Bob")
with pytest.raises(AssertionError, match=r"Agent Bob participated in the conversation when they should not have."):
# We can call it manually for testing
self.run_validator(agent_not_participating_validator(agent_name="Bob"), session)
You can see that we are expecting that error will be raised and we are catching it in the test.