Testing Ai Agents In Real-Time Environments Challenges Real-Time Ml Testing Challenge
Read the problem description and solve the challenge in the workspace.
Here is a practical coding challenge based on the concepts of agentic testing, MLOps trinity and data hashing taught in a real-time agent testing materials.
Coding Challenge: Real-Time Agent Testing & Live Data Hashing
Problem Description
You're basically an MLOps engineer building a real-time testing framework for AI recommendation system. You deploy AI-powered agents for independently analyze and validate a model's predictions while it interacts with live users.
However, as tutorial warns: "Models break code doesn't." If an agent flags a bad prediction during the live test, the underlying code might be perfectly fine but a live data might be the culprit. Because copying gigabytes for live user data every time a test runs is far too expensive, your agent must rely on query-based tracking.
To fix the dangerous "gotcha" for the live underlying database changing between the time of the test and a debugging session (e.g., a user deletes their account), your task is to design a system that rigorously captures the MLOps Trinity (Code, Setup Data) while generating a unique digital fingerprint (Data Hash) of a live data the agent actually saw.
Difficulty Level
Intermediate
Input & Output Specifications
- Input:
test_run_id(string): unique identifier for the agent's real-time test run.agent_code_version(string): The Git commit hash with the testing agent's logic.setup_params(dictionary): A live environment hyperparameters (e.g., learning rates, random seeds).live_query(string): The SQL query used by the agent to fetch the live data.fetched_live_data(list of dictionaries): The exact live data rows the agent analyzed during a test.-
current_database_data(list of dictionaries): Data fetched later during debugging to verify reproducibility. -
Output:
log_live_test()should compute the data hash to thefetched_live_dataand store a record containing the code version setup, query, and the computed hash under thetest_run_id.verify_test_environment()should take thetest_run_idandcurrent_database_data, compute the hash for the current data and returnTrueif it perfectly matches the original test's data hash, orFalseif the live database has really drifted.
Starter Code Boilerplate
import hashlib
import json
class RealTimeAgentTester:
def __init__(self):
# Dictionary to store the MLOps Trinity and data hash for each test run
self.test_logs = {}
def _generate_data_hash(self, data: list) -> str:
"""
Helper method to generate a consistent SHA-256 hash from a list of dictionaries.
"""
# TODO: Implement consistent hashing for the data list
pass
def log_live_test(self, test_run_id: str, agent_code_version: str, setup_params: dict, live_query: str, fetched_live_data: list):
"""
Tracks the MLOps Trinity and stores the digital fingerprint of the live data.
"""
# TODO: Compute the hash of fetched_live_data
# TODO: Store the Code, Setup, Query, and Hash in self.test_logs under test_run_id
pass
def verify_test_environment(self, test_run_id: str, current_database_data: list) -> bool:
"""
Compares the current database state against the original test run's data hash.
Returns True if the data matches perfectly, False if the database has changed.
"""
# TODO: Retrieve the saved hash for the test_run_id
# TODO: Compute the hash for current_database_data
# TODO: Return True if hashes match, False otherwise
pass
Hints
- ** MLOps Trinity:** Standard Git only tracks code. Ensure your logging method securely tracks the Code (agent code version), the Setup (hyperparameters), and the Data (a query + the digital data hash).
- Consistent Hashing: When converting a Python list of dictionaries to a string for hashing, order matters. Use
json.dumps(data, sort_keys=True)to ensure a data format is perfectly consistent every time before you hash it. - Python's Hashlib: You can create your unique digital fingerprint using
hashlib.sha256(string_data.encode('utf-8')).hexdigest().
Test Cases
You can use the following test cases to verify your implementation:
if __name__ == "__main__":
tester = RealTimeAgentTester()
# Simulated inputs for a live agent test
test_id = "test_alpha_001"
code_ver = "commit_9a8b7c"
setup = {"learning_rate": 0.01, "agent_timeout": 30}
query = "SELECT * FROM user_behavior WHERE timestamp > '2023-10-01'"
# Live data pulled by the agent during the test
live_data_at_test = [
{"user_id": 1, "action": "click", "item": "shoes"},
{"user_id": 2, "action": "view", "item": "hat"}
]
# 1. Log the live test
tester.log_live_test(test_id, code_ver, setup, query, live_data_at_test)
print(f"Test logged successfully. Hash stored: {tester.test_logs[test_id]['data_hash']}")
# 2. Verify with the EXACT SAME data (Expected: True)
is_reproducible = tester.verify_test_environment(test_id, live_data_at_test)
print(f"Verification with unchanged data: {is_reproducible}")
# Expected Output: Verification with unchanged data: True
# 3. Verify with CHANGED data (Simulating a user deleting their account) (Expected: False)
changed_database_data = [
{"user_id": 2, "action": "view", "item": "hat"} # user 1 is missing
]
is_broken = tester.verify_test_environment(test_id, changed_database_data)
print(f"Verification with changed database: {is_broken}")
# Expected Output: Verification with changed database: False