Monitoring the AI Itself

class QualitySLIs:
    def __init__(self):
        self.metrics = defaultdict(list)

    def record_response(self, feature, response, context):
        """Record quality metrics for a response."""
        # Hallucination rate: check if response cites context
        grounded = self.check_groundedness(response, context)
        self.metrics[f"{feature}_groundedness"].append(grounded)

        # Refusal rate: did the model say it does not know?
        refused = "I do not have enough information" in response
        self.metrics[f"{feature}_refusal"].append(refused)

        # Length: track response length for anomalies
        self.metrics[f"{feature}_length"].append(len(response))

        # User feedback: thumbs up/down
        # (Recorded separately via reaction handler)

    def check_groundedness(self, response, context):
        """Simple groundedness check: does response share words with context?"""
        response_words = set(response.lower().split())
        context_words = set(context.lower().split())
        overlap = len(response_words & context_words) / len(response_words)
        return overlap > 0.3  # At least 30% word overlap

    def get_sli_summary(self, feature, window_hours=24):
        """Get SLI summary for a feature."""
        return {
            "groundedness": np.mean(self.metrics[f"{feature}_groundedness"]),
            "refusal_rate": np.mean(self.metrics[f"{feature}_refusal"]),
            "avg_length": np.mean(self.metrics[f"{feature}_length"]),
        }

from scipy import stats

def detect_output_drift(recent_outputs, baseline_outputs):
    """Detect if model output distribution has changed."""
    # Compare response lengths
    recent_lengths = [len(o) for o in recent_outputs]
    baseline_lengths = [len(o) for o in baseline_outputs]

    # KS test for distribution change
    ks_stat, p_value = stats.ks_2samp(recent_lengths, baseline_lengths)

    # Compare groundedness scores
    recent_grounded = [check_groundedness(o) for o in recent_outputs]
    baseline_grounded = [check_groundedness(o) for o in baseline_outputs]

    grounding_drift = np.mean(recent_grounded) - np.mean(baseline_grounded)

    return {
        "length_drift": {"ks_statistic": ks_stat, "p_value": p_value},
        "grounding_drift": grounding_drift,
        "alert": p_value < 0.01 or abs(grounding_drift) > 0.1,
    }

class CostTracker:
    PRICING = {
        "gpt-4": {"input": 0.00003, "output": 0.00006},
        "gpt-4-turbo": {"input": 0.00001, "output": 0.00003},
        "gpt-3.5-turbo": {"input": 0.0000005, "output": 0.0000015},
    }

    def record_cost(self, feature, model, input_tokens, output_tokens):
        """Record cost for a request."""
        pricing = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = (input_tokens * pricing["input"] +
                output_tokens * pricing["output"])

        self.db.record({
            "feature": feature,
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost_usd": cost,
            "timestamp": datetime.utcnow()
        })

    def get_daily_cost(self, feature=None):
        """Get daily cost breakdown."""
        query = "SELECT SUM(cost_usd) FROM ai_costs WHERE date = today()"
        if feature:
            query += f" AND feature = '{feature}'"
        return self.db.execute(query)