Real-Time Agent Testing

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 the fetched_live_data and store a record containing the code version setup, query, and the computed hash under the test_run_id.
• verify_test_environment() should take the test_run_id and current_database_data, compute the hash for the current data and return True if it perfectly matches the original test's data hash, or False if 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