Monitoring AI Agents in Production

Coding Challenge: Monitoring AI Agents & Live Data Hashing

Problem Description

You're an MLOps engineer tasked with building an observability pipeline for AI agent in the live production environment. Your agent independently executes tool selections, decision chains, and model calls while interacting with the dynamic, live database.

Because AI systems change based upon live user input, your team knows that "models break, code doesn't." If the agent starts making poor decisions the underlying Python code might be perfectly fine. The live data it ingested could be culprit.

To recreate production bugs without saving gigabytes of raw user data every time the agent runs, you implement Query-Based Version Control. But, this introduces the dangerous edge case: if a live database changes later (e.g., a user deletes their account), running a same query will return different data, and your reproducibility is broken.

Your task is to build robust monitoring logger that safely tracks the MLOps Trinity (Code, Setup, Data) while utilizing Data Hashing to generate a unique digital fingerprint of the live data the agent actually saw. Later, your system must be able to verify if the database has drifted by checking this fingerprint.

Difficulty Level

Intermediate

Input & Output Specifications

Input: * agent_run_id (string): The unique identifier to an agent's production run. * code_version (string): The Git commit hash of agent's logic. * setup_params (dictionary): A live environment hyperparameters (e.g., random seeds token limits). * sql_query (string): An exact query an agent used to fetch the live data. * fetched_live_data (list with dictionaries): The exact live data rows the agent analyzed during its run. * current_database_data (list of dictionaries): Data fetched later during the debugging session to verify if a live database has changed.

Output: * log_agent_run(agent_run_id, code_version, setup_params, sql_query, fetched_live_data): Computes the digital fingerprint (hash) about the fetched_live_data and securely stores a record about the MLOps Trinity (Code, Setup Query + Hash) under the agent_run_id. * verify_data_integrity(agent_run_id, current_database_data): Retrieves the original hash for the given run computes the hash of the current_database_data. Returns True if they match perfectly or False if the live production database has drifted.

Starter Code Boilerplate

import hashlib
import json

class AgentObservabilityLogger:
    def __init__(self):
        # A mock database to store our observability logs
        self.observability_logs = {}

    def log_agent_run(self, agent_run_id, code_version, setup_params, sql_query, fetched_live_data):
        # TODO: Step 1 - Generate a consistent JSON string from fetched_live_data

        # TODO: Step 2 - Compute the SHA-256 hash of the formatted string

        # TODO: Step 3 - Store the MLOps Trinity and the data hash in self.observability_logs
        pass

    def verify_data_integrity(self, agent_run_id, current_database_data):
        # TODO: Step 1 - Retrieve the stored log using the agent_run_id

        # TODO: Step 2 - Generate the hash for the current_database_data

        # TODO: Step 3 - Compare the new hash with the stored original hash and return True/False
        pass

Hints

• A MLOps Trinity: Standard monitoring isn't enough. Ensure your log groups a Code (code_version), a Setup (setup_params), and the Data (sql_query + the computed hash).
• Consistent Formatting: When hashing a list of dictionaries in Python, order and spacing matter immensely; use json.dumps(data_payload, sort_keys=True) to ensure the payload format is absolutely identical every time before hashing it.
• Generating the Digital Fingerprint: Use Python's built-into hashlib library. You can generate your secure hash like this: hashlib.sha256(formatted_string.encode('utf-8')).hexdigest().

Test Cases

You can really run the following test script towards verify your implementation works correctly:

# 1. Initialize Logger
logger = AgentObservabilityLogger()

# 2. Mock Production Inputs
run_id = "agent_run_9942"
code_commit = "a1b2c3d4"
setup = {"learning_rate": 0.01, "max_tokens": 1500}
query = "SELECT * FROM user_behavior WHERE date < '2026-10-01'"

# Data the agent sees in production today
live_data_today = [
    {"user_id": 1, "action": "click", "timestamp": "10:00"},
    {"user_id": 2, "action": "purchase", "timestamp": "10:05"}
]

# 3. Log the run
logger.log_agent_run(run_id, code_commit, setup, query, live_data_today)

# 4. Verify Reproducibility (No changes to database)
# Should return True
is_consistent = logger.verify_data_integrity(run_id, live_data_today)
print(f"Test 1 - Data Intact: {'Passed' if is_consistent else 'Failed'}")

# 5. Verify Reproducibility (Database changed - User 2 deleted their account)
live_data_tomorrow = [
    {"user_id": 1, "action": "click", "timestamp": "10:00"}
]

# Should return False due to data drift
is_consistent_drift = logger.verify_data_integrity(run_id, live_data_tomorrow)
print(f"Test 2 - Data Drift Detected: {'Passed' if not is_consistent_drift else 'Failed'}")