Use database operations to track and monitor data changes in AI agent workflows. Create database schemas to log changes, track agent actions, and analyze data patterns using natural language queries.

Key Features

Change Tracking Database

  • Data Change Logs: Track all data changes with timestamps and context
  • Agent Action Logging: Log agent actions and their data impacts
  • Version History: Maintain version history of data changes
  • Audit Trail: Complete audit trail of all data modifications

Natural Language Monitoring

  • Query Change Data: Use natural language to query change logs
  • Pattern Analysis: Analyze data change patterns and trends
  • Impact Assessment: Assess the impact of data changes on system behavior
  • Reporting: Generate reports on data changes and agent activity

Database Operations

  • Schema Management: Create schemas to track various types of changes
  • REST API Access: Access change data through auto-generated APIs
  • Real-time Logging: Log changes as they occur in real-time
  • Flexible Queries: Query change data using natural language

Implementation Examples

Creating Change Tracking Schema

# Using Gibson CLI to create change tracking database
# Create tables for tracking data changes
# gibson modify data_changes "Create data_changes table with id, table_name, record_id, change_type, old_value, new_value, changed_by, timestamp"
# gibson modify agent_actions "Create agent_actions table with id, agent_id, action_type, target_table, target_id, action_data, timestamp"
# gibson modify system_events "Create system_events table with id, event_type, event_data, source, severity, timestamp"
# gibson modify change_summaries "Create change_summaries table with id, date, table_name, change_count, agent_id, summary"

# Generate models and deploy
# gibson code models
# gibson merge

Change Tracking System

import requests
from datetime import datetime
import json

class ChangeTracker:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.gibsonai.com/v1/-"
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def log_data_change(self, table_name, record_id, change_type, old_value, new_value, changed_by):
        """Log a data change"""
        change_data = {
            "table_name": table_name,
            "record_id": record_id,
            "change_type": change_type,
            "old_value": json.dumps(old_value) if old_value else None,
            "new_value": json.dumps(new_value) if new_value else None,
            "changed_by": changed_by,
            "timestamp": datetime.now().isoformat()
        }

        response = requests.post(
            f"{self.base_url}/data-changes",
            json=change_data,
            headers=self.headers
        )

        if response.status_code == 201:
            print(f"Data change logged: {change_type} on {table_name}")
            return response.json()
        else:
            print(f"Failed to log data change: {response.status_code}")
            return None

    def log_agent_action(self, agent_id, action_type, target_table, target_id, action_data):
        """Log an agent action"""
        action_data_record = {
            "agent_id": agent_id,
            "action_type": action_type,
            "target_table": target_table,
            "target_id": target_id,
            "action_data": json.dumps(action_data),
            "timestamp": datetime.now().isoformat()
        }

        response = requests.post(
            f"{self.base_url}/agent-actions",
            json=action_data_record,
            headers=self.headers
        )

        if response.status_code == 201:
            print(f"Agent action logged: {action_type} by {agent_id}")
            return response.json()
        else:
            print(f"Failed to log agent action: {response.status_code}")
            return None

    def log_system_event(self, event_type, event_data, source, severity="info"):
        """Log a system event"""
        event_record = {
            "event_type": event_type,
            "event_data": json.dumps(event_data),
            "source": source,
            "severity": severity,
            "timestamp": datetime.now().isoformat()
        }

        response = requests.post(
            f"{self.base_url}/system-events",
            json=event_record,
            headers=self.headers
        )

        if response.status_code == 201:
            print(f"System event logged: {event_type}")
            return response.json()
        else:
            print(f"Failed to log system event: {response.status_code}")
            return None

    def create_change_summary(self, date, table_name, change_count, agent_id, summary):
        """Create a change summary"""
        summary_data = {
            "date": date,
            "table_name": table_name,
            "change_count": change_count,
            "agent_id": agent_id,
            "summary": summary
        }

        response = requests.post(
            f"{self.base_url}/change-summaries",
            json=summary_data,
            headers=self.headers
        )

        if response.status_code == 201:
            print(f"Change summary created for {date}")
            return response.json()
        else:
            print(f"Failed to create change summary: {response.status_code}")
            return None

Data Change Monitoring

class DataChangeMonitor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.gibsonai.com/v1/-"
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def monitor_recent_changes(self, hours=24):
        """Monitor recent data changes"""
        query_request = {
            "query": f"Show all data changes from the last {hours} hours grouped by table and change type"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print(f"Recent changes in last {hours} hours:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def analyze_change_patterns(self):
        """Analyze data change patterns"""
        query_request = {
            "query": "Analyze data change patterns by agent, table, and time of day for the last 7 days"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print("Change pattern analysis:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def detect_unusual_activity(self):
        """Detect unusual data change activity"""
        query_request = {
            "query": "Find unusual data change activity including high-frequency changes, bulk operations, and changes outside normal hours"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print("Unusual activity detected:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def get_agent_activity_summary(self, agent_id):
        """Get activity summary for specific agent"""
        query_request = {
            "query": f"Show activity summary for agent {agent_id} including actions performed, data changes, and time patterns"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print(f"Activity summary for agent {agent_id}:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

Agent Integration with Change Tracking

class MonitoredAgent:
    def __init__(self, agent_id, api_key):
        self.agent_id = agent_id
        self.api_key = api_key
        self.base_url = "https://api.gibsonai.com/v1/-"
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.change_tracker = ChangeTracker(api_key)

    def create_record(self, table_name, record_data):
        """Create a record with change tracking"""
        # Create the record
        response = requests.post(
            f"{self.base_url}/{table_name}",
            json=record_data,
            headers=self.headers
        )

        if response.status_code == 201:
            created_record = response.json()

            # Log the change
            self.change_tracker.log_data_change(
                table_name=table_name,
                record_id=created_record["id"],
                change_type="CREATE",
                old_value=None,
                new_value=record_data,
                changed_by=self.agent_id
            )

            # Log the agent action
            self.change_tracker.log_agent_action(
                agent_id=self.agent_id,
                action_type="CREATE_RECORD",
                target_table=table_name,
                target_id=created_record["id"],
                action_data=record_data
            )

            print(f"Record created and logged: {table_name}")
            return created_record
        else:
            print(f"Failed to create record: {response.status_code}")
            return None

    def update_record(self, table_name, record_id, update_data):
        """Update a record with change tracking"""
        # Get current record for old_value
        current_response = requests.get(
            f"{self.base_url}/{table_name}/{record_id}",
            headers=self.headers
        )

        old_value = current_response.json() if current_response.status_code == 200 else None

        # Update the record
        response = requests.put(
            f"{self.base_url}/{table_name}/{record_id}",
            json=update_data,
            headers=self.headers
        )

        if response.status_code == 200:
            updated_record = response.json()

            # Log the change
            self.change_tracker.log_data_change(
                table_name=table_name,
                record_id=record_id,
                change_type="UPDATE",
                old_value=old_value,
                new_value=update_data,
                changed_by=self.agent_id
            )

            # Log the agent action
            self.change_tracker.log_agent_action(
                agent_id=self.agent_id,
                action_type="UPDATE_RECORD",
                target_table=table_name,
                target_id=record_id,
                action_data=update_data
            )

            print(f"Record updated and logged: {table_name}/{record_id}")
            return updated_record
        else:
            print(f"Failed to update record: {response.status_code}")
            return None

    def delete_record(self, table_name, record_id):
        """Delete a record with change tracking"""
        # Get current record for old_value
        current_response = requests.get(
            f"{self.base_url}/{table_name}/{record_id}",
            headers=self.headers
        )

        old_value = current_response.json() if current_response.status_code == 200 else None

        # Delete the record
        response = requests.delete(
            f"{self.base_url}/{table_name}/{record_id}",
            headers=self.headers
        )

        if response.status_code == 200:
            # Log the change
            self.change_tracker.log_data_change(
                table_name=table_name,
                record_id=record_id,
                change_type="DELETE",
                old_value=old_value,
                new_value=None,
                changed_by=self.agent_id
            )

            # Log the agent action
            self.change_tracker.log_agent_action(
                agent_id=self.agent_id,
                action_type="DELETE_RECORD",
                target_table=table_name,
                target_id=record_id,
                action_data={"deleted_record": old_value}
            )

            print(f"Record deleted and logged: {table_name}/{record_id}")
            return True
        else:
            print(f"Failed to delete record: {response.status_code}")
            return False

Change Analysis and Reporting

class ChangeAnalyzer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.gibsonai.com/v1/-"
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def generate_daily_report(self, date):
        """Generate daily change report"""
        query_request = {
            "query": f"Generate a comprehensive report of all data changes on {date} including counts by table, agent, and change type"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print(f"Daily report for {date}:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def analyze_agent_impact(self, agent_id, days=7):
        """Analyze the impact of a specific agent"""
        query_request = {
            "query": f"Analyze the impact of agent {agent_id} over the last {days} days including records created, updated, deleted, and affected tables"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print(f"Agent {agent_id} impact analysis:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def find_data_anomalies(self):
        """Find data anomalies and unusual patterns"""
        query_request = {
            "query": "Find data anomalies including unusual change volumes, unexpected change types, and irregular timing patterns"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print("Data anomalies found:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

    def track_data_evolution(self, table_name, record_id):
        """Track the evolution of a specific record"""
        query_request = {
            "query": f"Show the complete change history for record {record_id} in table {table_name} ordered by timestamp"
        }

        response = requests.post(
            f"{self.base_url}/query",
            json=query_request,
            headers=self.headers
        )

        if response.status_code == 200:
            results = response.json()
            print(f"Change history for {table_name}/{record_id}:")
            for result in results:
                print(f"  {result}")
            return results
        else:
            print(f"Query failed: {response.status_code}")
            return None

Use Cases

Agent Monitoring

Perfect for:

  • Tracking agent actions and their data impacts
  • Monitoring agent performance and behavior
  • Auditing agent operations for compliance
  • Identifying agent-related issues or problems

Data Governance

Enable:

  • Maintaining complete audit trails of data changes
  • Tracking data lineage and transformation
  • Ensuring data quality and consistency
  • Supporting compliance requirements

System Analysis

Support:

  • Analyzing data change patterns and trends
  • Identifying system performance issues
  • Understanding data usage patterns
  • Optimizing system performance

Problem Diagnosis

Allow:

  • Investigating data-related issues
  • Tracking down the source of problems
  • Analyzing the impact of changes
  • Supporting troubleshooting efforts

Benefits for AI Agent Systems

Comprehensive Tracking

  • Complete Audit Trail: Full record of all data changes and agent actions
  • Natural Language Queries: Query change data using natural language
  • Pattern Analysis: Analyze patterns in data changes and agent behavior
  • Impact Assessment: Understand the impact of changes on system behavior

Monitoring and Analysis

  • Real-time Logging: Log changes as they occur
  • Historical Analysis: Analyze historical change patterns
  • Anomaly Detection: Identify unusual or suspicious activity
  • Performance Monitoring: Track system performance over time

Flexible Architecture

  • Database Storage: Store change data in structured database format
  • REST API Access: Access change data through auto-generated APIs
  • Flexible Schema: Adapt to different monitoring needs
  • Integration Support: Easy integration with existing systems

Important Limitations

What This Approach Does NOT Provide

  • Automated Alerts: No automatic alert or notification system
  • Real-time Monitoring: No real-time monitoring or alerting capabilities
  • Threshold Management: No automatic threshold monitoring
  • Workflow Automation: No automated response to changes

External Integration Required

For complete monitoring solutions, you'll need:

  • Monitoring Tools: Use external monitoring and alerting tools
  • Notification Systems: Implement notification systems separately
  • Workflow Automation: Use external workflow automation tools
  • Dashboard Tools: Use external dashboard and visualization tools

Best Practices

Change Tracking Design

  • Comprehensive Logging: Log all relevant changes and actions
  • Consistent Format: Use consistent format for change records
  • Appropriate Detail: Include appropriate level of detail in logs
  • Performance Consideration: Consider performance impact of logging

Data Management

  • Retention Policies: Implement appropriate data retention policies
  • Archive Strategy: Archive old change data as needed
  • Data Quality: Maintain high-quality change data
  • Privacy Considerations: Consider privacy requirements for change data

Analysis and Reporting

  • Regular Analysis: Regularly analyze change data for insights
  • Trend Monitoring: Monitor trends in data changes
  • Anomaly Detection: Look for unusual patterns or anomalies
  • Actionable Insights: Focus on actionable insights from change data

Getting Started

  1. Design Change Schema: Plan your change tracking database structure
  2. Create Database: Use Gibson CLI to create change tracking schema
  3. Implement Tracking: Add change tracking to your agent systems
  4. Analyze Data: Use natural language queries to analyze change data
  5. Monitor and Improve: Continuously monitor and improve your tracking

Gibson CLI Commands

# Create change tracking schema
gibson modify table_name "description of change tracking table"
gibson code models
gibson merge

# Generate models for change tracking
gibson code models
gibson code schemas

Sample Schema for Change Tracking

# Basic change tracking schema
change_tables = {
    "data_changes": "Create data_changes table with id, table_name, record_id, change_type, old_value, new_value, changed_by, timestamp",
    "agent_actions": "Create agent_actions table with id, agent_id, action_type, target_table, target_id, action_data, timestamp",
    "system_events": "Create system_events table with id, event_type, event_data, source, severity, timestamp"
}

Ready to implement data change tracking for your AI agent systems? Get started with GibsonAI.